Partial payments script

ExampleRunner/Program.cs

@@ -22,8 +22,9 @@ public class AppArguments
     [Option('f', "framework", Default = Framework.Automatic, Required = false, HelpText = "Type of AI Framework to run")]
     public required Framework AiFramework { get; set; }
     [Option('m', "mainnet", Default = false, Required = false, HelpText = "Enables usage of mainnet")]
-    public required bool Mainnet { get;  set; }
-
+    public required bool Mainnet { get; set; }
+    [Option('p', "pay-interval", Default = null, Required = false, HelpText = "Interval between partial payments in seconds")]
+    public UInt32? PaymentInterval { get; set; }
 }
 
 
@@ -41,6 +42,8 @@ static void Main(string[] args)
         GolemRelay.SetEnv(parsed.Relay);
 
         var App = new FullExample(workDir, "Requestor", loggerFactory, runtime: parsed.AiFramework.ToString().ToLower(), parsed.Mainnet);
+
+        App.PaymentInterval = parsed.PaymentInterval;
         var logger = loggerFactory.CreateLogger("Example");
 
         _ = Task.Run(async () =>

Golem.Package/Args.cs

@@ -5,9 +5,9 @@ public class BuildArgs
 {
     [Option('t', "target", Default = "package", Required = false, HelpText = "Directory where binaries will be generated relative to working dir")]
     public required string Target { get; set; }
-    [Option('y', "yagna-version", Default = "pre-rel-v0.16.0-ai-rc15", Required = false, HelpText = "Yagna version github tag")]
+    [Option('y', "yagna-version", Default = "pre-rel-v0.16.0-ai-rc16", Required = false, HelpText = "Yagna version github tag")]
     public required string GolemVersion { get; set; }
-    [Option('r', "runtime-version", Default = "v0.2.0", Required = false, HelpText = "Runtime version github tag")]
+    [Option('r', "runtime-version", Default = "pre-rel-v0.2.1-rc1", Required = false, HelpText = "Runtime version github tag")]
     public required string RuntimeVersion { get; set; }
     [Option('c', "dont-clean", Default = false, Required = false, HelpText = "Remove temporary directories")]
     public required bool DontClean { get; set; }

Golem.Tools/App/app.py

@@ -3,7 +3,7 @@
 import json
 
 from dataclasses import dataclass
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 
 from yapapi import Golem
 from yapapi.payload import Payload
@@ -15,7 +15,6 @@
 import argparse
 import asyncio
 import tempfile
-from datetime import datetime, timezone
 from pathlib import Path
 from typing import Optional
 
@@ -26,6 +25,7 @@
 from yapapi import windows_event_loop_fix
 from yapapi.log import enable_default_logger
 from yapapi.strategy import SCORE_TRUSTED, SCORE_REJECTED, MarketStrategy
+from yapapi.strategy.base import PropValueRange, PROP_DEBIT_NOTE_INTERVAL_SEC, PROP_PAYMENT_TIMEOUT_SEC
 
 # Utils
 
@@ -61,6 +61,7 @@ def build_parser(description: str) -> argparse.ArgumentParser:
     )
     parser.add_argument("--runtime", default="dummy", help="Runtime name, for example `automatic`")
     parser.add_argument("--descriptor", default=None, help="Path to node descriptor file")
+    parser.add_argument("--pay-interval", default=180, help="Interval of making partial payments")
     return parser
 
 
@@ -131,8 +132,12 @@ class ProviderOnceStrategy(MarketStrategy):
     """Hires provider only once.
     """
 
-    def __init__(self):
+    def __init__(self, pay_interval=180):
         self.history = set(())
+        self.acceptable_prop_value_range_overrides =  {
+            PROP_DEBIT_NOTE_INTERVAL_SEC: PropValueRange(60, None),
+            PROP_PAYMENT_TIMEOUT_SEC: PropValueRange(int(pay_interval), None),
+        }
 
     async def score_offer(self, offer):
         if offer.issuer not in self.history:
@@ -193,12 +198,12 @@ async def start(self):
     def __init__(self, strategy: ProviderOnceStrategy):
         super().__init__()
         self.strategy = strategy
+
 
-
-async def main(subnet_tag, descriptor, driver=None, network=None, runtime="dummy"):
-    strategy = ProviderOnceStrategy()
+async def main(subnet_tag, descriptor, driver=None, network=None, runtime="dummy", args=None):
+    strategy = ProviderOnceStrategy(pay_interval=args.pay_interval)
     async with Golem(
-        budget=1.0,
+        budget=4.0,
         subnet_tag=subnet_tag,
         strategy=strategy,
         payment_driver=driver,
@@ -212,6 +217,7 @@ async def main(subnet_tag, descriptor, driver=None, network=None, runtime="dummy
                 {"strategy": strategy}
             ],
             num_instances=1,
+            expiration=datetime.now(timezone.utc) + timedelta(days=10),
         )
 
         async def print_usage():
@@ -264,7 +270,6 @@ def instances():
                 } for s in cluster.instances
             ]
 
-
         usage_printed = False
         while True:
             await asyncio.sleep(3)
@@ -278,6 +283,7 @@ def instances():
 
             print(f"""instances: {[f"{r['name']}: {r['state']}" for r in i]}""")
 
+
 if __name__ == "__main__":
     parser = build_parser("Run AI runtime task")
     now = datetime.now().strftime("%Y-%m-%d_%H.%M.%S")
@@ -291,6 +297,7 @@ def instances():
             driver=args.payment_driver,
             network=args.payment_network,
             runtime=args.runtime,
+            args=args
         ),
         log_file=args.log_file,
     )

Golem.Tools/GolemPackageBuilder.cs

@@ -18,8 +18,8 @@ namespace Golem.Tools
 {
     public class PackageBuilder
     {
-        public static string CURRENT_GOLEM_VERSION = "pre-rel-v0.16.0-ai-rc15";
-        public static string CURRENT_RUNTIME_VERSION = "v0.2.0";
+        public static string CURRENT_GOLEM_VERSION = "pre-rel-v0.16.0-ai-rc16";
+        public static string CURRENT_RUNTIME_VERSION = "pre-rel-v0.2.1-rc1";
 
         internal static string InitTestDirectory(string name, bool cleanupData = true)
         {

Golem.Tools/SampleApp.cs

@@ -53,6 +53,7 @@ public class FullExample : IAsyncDisposable, INotifyPropertyChanged
         private readonly bool _mainnet;
 
         private readonly string _runtime;
+        public UInt32? PaymentInterval { get; set; }
 
         private string _message;
         public string Message
@@ -97,6 +98,10 @@ private string ExtraArgs()
             {
                 args += $" --descriptor {GetNodeDescriptor()}";
             }
+            if (PaymentInterval.HasValue)
+            {
+                args += $" --pay-interval {PaymentInterval.Value}";
+            }
             return args;
         }
 

Golem/ActivityLoop.cs

@@ -53,12 +53,12 @@ public async Task Start(
                     await foreach (var trackingEvent in _yagnaApi.ActivityMonitorStream(token))
                     {
                         var activities = trackingEvent?.Activities ?? new List<ActivityState>();
-                        
+
                         List<Job> currentJobs = await UpdateJobs(_jobs, activities);
-                        
+
                         var job = SelectCurrentJob(currentJobs);
                         setCurrentJob(job);
-                        if(job == null)
+                        if (job == null)
                         {
                             // Sometimes finished jobs end up in Idle state
                             _jobs.SetAllJobsFinished();
@@ -97,7 +97,7 @@ public async Task Start(
         if (currentJobs.Count == 0)
         {
             _logger.LogDebug("Cleaning current job field");
-            
+
             return null;
         }
         else if (currentJobs.Count == 1)
@@ -139,12 +139,13 @@ public async Task<List<Job>> UpdateJobs(IJobs jobs, List<ActivityState> activity
     {
         var result = await Task.WhenAll(
             activityStates
-                .Select(async d => {
-                        var usage = d.Usage!=null
-                                ? GolemUsage.From(d.Usage)
-                                : null;
-                        return await jobs.UpdateJob(d.Id, null, usage);
-                    }
+                .Select(async d =>
+                {
+                    var usage = d.Usage != null
+                            ? GolemUsage.From(d.Usage)
+                            : null;
+                    return await jobs.UpdateJobByActivity(d.Id, null, usage);
+                }
                 )
         );
 

Golem/Golem.cs

@@ -383,7 +383,7 @@ private void SetCurrentJob(Job? job)
             }
             else
             {
-                _logger.LogInformation("Job has not changed.");
+                _logger.LogDebug("Job has not changed.");
             }
         }
 

Golem/InvoiceEventsLoop.cs

@@ -1,6 +1,8 @@
 using Golem;
 using Golem.Yagna;
+
 using GolemLib.Types;
+
 using Microsoft.Extensions.Logging;
 
 
@@ -10,9 +12,8 @@ class InvoiceEventsLoop
     private readonly CancellationToken _token;
     private readonly ILogger _logger;
     private readonly IJobs _jobs;
-    private DateTime _since = DateTime.MinValue;
     private readonly EventsPublisher _events;
-    
+
 
     public InvoiceEventsLoop(YagnaApi yagnaApi, IJobs jobs, CancellationToken token, EventsPublisher events, ILogger logger)
     {
@@ -23,50 +24,91 @@ public InvoiceEventsLoop(YagnaApi yagnaApi, IJobs jobs, CancellationToken token,
         _events = events;
     }
 
-    public async Task Start()
+    public Task Start()
     {
-        _logger.LogInformation("Starting monitoring invoice events");
+        return Task.WhenAll(Task.Run(PaymentsLoop), Task.Run(InvoiceLoop));
+    }
 
-        DateTime newReconnect = DateTime.Now;
+    public async Task InvoiceLoop()
+    {
+        _logger.LogInformation("Starting monitoring invoice events");
 
-        await Task.Yield();
-
+        DateTime since = DateTime.Now;
         while (true)
         {
             _token.ThrowIfCancellationRequested();
             try
             {
-                var invoiceEvents = await _yagnaApi.GetInvoiceEvents(_since, _token);
+                _token.ThrowIfCancellationRequested();
+                _logger.LogDebug("Checking for new invoice events since: {}", since);
+
+                var invoiceEvents = await _yagnaApi.GetInvoiceEvents(since, _token);
                 if (invoiceEvents != null && invoiceEvents.Count > 0)
                 {
-                    _since = invoiceEvents.Max(x => x.EventDate);
-
-                    foreach(var invoiceEvent in invoiceEvents.OrderBy(e => e.EventDate))
+                    foreach (var invoiceEvent in invoiceEvents.OrderBy(e => e.EventDate))
                     {
+                        _logger.LogDebug("Invoice event for: {}", invoiceEvent.InvoiceId);
+
                         await UpdatesForInvoice(invoiceEvent);
+                        since = invoiceEvent.EventDate;
                     }
                 }
             }
-            catch(OperationCanceledException e)
+            catch (OperationCanceledException)
             {
-                _events.Raise(new ApplicationEventArgs("ActivityLoop", $"OperationCanceledException", ApplicationEventArgs.SeverityLevel.Error, e));
                 _logger.LogInformation("Invoice events loop cancelled");
                 return;
             }
-            catch(Exception e)
+            catch (Exception e)
             {
-                _events.Raise(new ApplicationEventArgs("ActivityLoop", $"Exception {e.Message}", ApplicationEventArgs.SeverityLevel.Error, e));
+                _events.Raise(new ApplicationEventArgs("InvoiceLoop", $"Exception {e.Message}", ApplicationEventArgs.SeverityLevel.Error, e));
                 _logger.LogError("Error in invoice events loop: {e}", e.Message);
                 await Task.Delay(TimeSpan.FromSeconds(5), _token);
             }
-        }        
+        }
+    }
+
+    public async Task PaymentsLoop()
+    {
+        _logger.LogInformation("Starting monitoring payments");
+
+        DateTime since = DateTime.Now;
+        while (true)
+        {
+            try
+            {
+
+                _token.ThrowIfCancellationRequested();
+                _logger.LogDebug("Checking for new payments since: {}", since);
+
+                var payments = await _yagnaApi.GetPayments(since, _token);
+                foreach (var payment in payments.OrderBy(pay => pay.Timestamp))
+                {
+                    _logger.LogDebug("New Payment, id: {}", payment.PaymentId);
+
+                    await _jobs.UpdatePartialPayment(payment);
+                    since = payment.Timestamp;
+                }
+            }
+            catch (OperationCanceledException)
+            {
+                _logger.LogInformation("Payments loop cancelled");
+                return;
+            }
+            catch (Exception e)
+            {
+                _events.Raise(new ApplicationEventArgs("PaymentsLoop", $"Exception {e.Message}", ApplicationEventArgs.SeverityLevel.Error, e));
+                _logger.LogError("Error in payments loop: {e}", e.Message);
+                await Task.Delay(TimeSpan.FromSeconds(5), _token);
+            }
+        }
     }
 
     private async Task UpdatesForInvoice(InvoiceEvent invoiceEvent)
     {
         var invoice = await _yagnaApi.GetInvoice(invoiceEvent.InvoiceId, _token);
 
-        foreach(var activityId in invoice.ActivityIds)
-            await _jobs.UpdateJob(activityId, invoice, null);
+        _logger.LogDebug("Update Invoice info for Job: {}, status: {}", invoice.AgreementId, invoice.Status);
+        await _jobs.UpdateJob(invoice.AgreementId, invoice, null);
     }
 }