Simplified Workflow.Delay method

Author: stidsborg

Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/RFunctionTests/SuspensionTests.cs

@@ -99,4 +99,8 @@ public override Task DelayedFlowIsRestartedOnce()
     [TestMethod]
     public override Task InterruptedExecutingFlowIsRestartedOnce()
         => InterruptedExecutingFlowIsRestartedOnce(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task TwoDelaysFlowCompletesSuccessfully()
+        => TwoDelaysFlowCompletesSuccessfully(FunctionStoreFactory.Create());
 }

Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/SuspensionTests.cs

@@ -826,6 +826,46 @@ protected async Task InterruptedExecutingFlowIsRestartedOnce(Task<IFunctionStore
         await cp.WaitForCompletion(allowPostponeAndSuspended: true);
 
         syncedCounter.Current.ShouldBe(2);
+
+        unhandledExceptionHandler.ShouldNotHaveExceptions();
+    }
+
+    public abstract Task TwoDelaysFlowCompletesSuccessfully();
+    protected async Task TwoDelaysFlowCompletesSuccessfully(Task<IFunctionStore> storeTask)
+    {
+        var store = await storeTask;
+        var id = TestFlowId.Create();
+        var (flowType, flowInstance) = id;
+
+        var unhandledExceptionHandler = new UnhandledExceptionCatcher();
+        using var functionsRegistry = new FunctionsRegistry
+        (
+            store,
+            new Settings(unhandledExceptionHandler.Catch)
+        );
+
+        var counter = new SyncedCounter();
+        var registration = functionsRegistry.RegisterParamless(
+            flowType,
+            inner: async Task (workflow) =>
+            {
+                counter.Increment();
+                await workflow.Delay(TimeSpan.FromMilliseconds(100), alias: "first");
+                await workflow.Delay(TimeSpan.FromMilliseconds(100), alias: "second");
+            }
+        );
+
+        await registration.Schedule(flowInstance);
+
+        var cp = await registration.ControlPanel(flowInstance);
+        cp.ShouldNotBeNull();
+        
+        await cp.WaitForCompletion(allowPostponeAndSuspended: true);
+        await cp.Refresh();
+
+        cp.Status.ShouldBe(Status.Succeeded);
+
+        counter.Current.ShouldBe(3);
 
         unhandledExceptionHandler.ShouldNotHaveExceptions();
     }

Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowsManager.cs

@@ -0,0 +1,63 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Cleipnir.ResilientFunctions.Domain;
+using Cleipnir.ResilientFunctions.Queuing;
+using Cleipnir.ResilientFunctions.Storage;
+
+namespace Cleipnir.ResilientFunctions.CoreRuntime;
+
+public record FlowStatus(StoredId Id, Action Suspend, QueueManager QueueManager, int Threads, int AwaitingThreads);
+
+public class FlowsManager
+{
+    private readonly Dictionary<StoredId, FlowStatus> _dict = new();
+    private readonly Lock _lock = new();
+
+    public void AddFlow(StoredId id, Action suspend, QueueManager queueManager)
+    {
+        lock (_lock)
+        {
+            if (!_dict.ContainsKey(id))
+                return;
+
+            _dict[id] = new FlowStatus(id, suspend, queueManager, Threads: 1, AwaitingThreads: 0);
+        }
+    }
+
+    public async Task Interrupt(IEnumerable<StoredId> ids)
+    {
+        lock (_lock)
+        {
+            foreach (var id in ids)
+            {
+                if (!_dict.ContainsKey(id))
+                    continue;
+                
+                var queueManager = _dict[id].QueueManager;
+                Task.Run(() => queueManager.FetchMessagesOnce());
+            }
+        }
+    }
+
+    public void AddThread(StoredId id)
+    {
+        throw new NotImplementedException();
+    }
+
+    public void CompletedThread(StoredId id)
+    {
+        throw new NotImplementedException();
+    }
+
+    public void StartedThread(StoredId id)
+    {
+        throw new NotImplementedException();
+    }
+
+    public void SuspendedThread(StoredId id)
+    {
+        throw new NotImplementedException();
+    }
+}

Core/Cleipnir.ResilientFunctions/CoreRuntime/Invocation/Workflow.cs

@@ -40,34 +40,36 @@ public Workflow(FlowId flowId, StoredId storedId, Effect effect, Utilities utili
     public async Task RegisterCorrelation(string correlation) => await Correlations.Register(correlation);
 
     public Task Delay(TimeSpan @for, bool suspend = true, string? alias = null) => Delay(until: _utcNow() + @for, suspend, alias);
-    public async Task Delay(DateTime until, bool suspend = true, string? alias = null)
+    public Task Delay(DateTime until, bool suspend = true, string? alias = null)
     {
         var effectId = Effect.TakeNextImplicitId();
         var timeoutId = EffectId.CreateWithCurrentContext(effectId);
 
-        var (status, expiry) = await RegisteredTimeouts.RegisterTimeout(
-            timeoutId,
-            until,
-            publishMessage: false,
-            alias
-        );
-
-        if (status is TimeoutStatus.Completed or TimeoutStatus.Cancelled)
-        {
-            return;
-        }
-
-        var delay =  (expiry - _utcNow()).RoundUpToZero();
-        if (!suspend)
+        async Task Inner()
         {
-            await Task.Delay(delay);
-            delay = TimeSpan.Zero;
+            var expiry = await Effect.CreateOrGet(timeoutId, until.ToUniversalTime().Ticks, alias, flush: false);
+
+            if (expiry == -1)
+            {
+                return;
+            }
+
+            Effect.FlowMinimumTimeout.AddTimeout(timeoutId, expiry.ToDateTime());
+            var delay = (expiry.ToDateTime() - _utcNow()).RoundUpToZero();
+            if (!suspend)
+            {
+                await Task.Delay(delay);
+                delay = TimeSpan.Zero;
+            }
+
+            if (delay > TimeSpan.Zero)
+                throw new SuspendInvocationException();
+        
+            await Effect.Upsert(timeoutId, -1L, alias, flush: false);
+            Effect.FlowMinimumTimeout.RemoveTimeout(timeoutId);
         }
 
-        if (delay == TimeSpan.Zero)
-            await RegisteredTimeouts.CompleteTimeout(timeoutId, alias);
-        else
-            throw new SuspendInvocationException();
+        return Inner();
     }
 
     public Task<T> Message<T>(TimeSpan? maxWait = null) where T : class

Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/RFunctionTests/SuspensionTests.cs

@@ -87,4 +87,8 @@ public override Task DelayedFlowIsRestartedOnce()
     [TestMethod]
     public override Task InterruptedExecutingFlowIsRestartedOnce()
         => InterruptedExecutingFlowIsRestartedOnce(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task TwoDelaysFlowCompletesSuccessfully()
+        => TwoDelaysFlowCompletesSuccessfully(FunctionStoreFactory.Create());
 }

Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/RFunctionTests/SuspensionTests.cs

@@ -88,4 +88,8 @@ public override Task DelayedFlowIsRestartedOnce()
     [TestMethod]
     public override Task InterruptedExecutingFlowIsRestartedOnce()
         => InterruptedExecutingFlowIsRestartedOnce(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task TwoDelaysFlowCompletesSuccessfully()
+        => TwoDelaysFlowCompletesSuccessfully(FunctionStoreFactory.Create());
 }

Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/RFunctionTests/SuspensionTests.cs

@@ -88,4 +88,8 @@ public override Task DelayedFlowIsRestartedOnce()
     [TestMethod]
     public override Task InterruptedExecutingFlowIsRestartedOnce()
         => InterruptedExecutingFlowIsRestartedOnce(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task TwoDelaysFlowCompletesSuccessfully()
+        => TwoDelaysFlowCompletesSuccessfully(FunctionStoreFactory.Create());
 }