Interrupt-driven message fetch; merge FetchedMessages into QueueManager (#150)

Author: stidsborg

.gitignore

@@ -385,4 +385,7 @@ FodyWeavers.xsd
 
 # JetBrains Rider
 .idea/
-*.sln.iml
+*.sln.iml
+
+# macOS
+.DS_Store

Core/Cleipnir.ResilientFunctions.Tests/Messaging/TestTemplates/MessagesTests.cs

@@ -256,8 +256,8 @@ await registration.SendMessages(
         var controlPanel1 = await registration.ControlPanel("Instance#1").ShouldNotBeNullAsync();
         var controlPanel2 = await registration.ControlPanel("Instance#2").ShouldNotBeNullAsync();
 
-        await controlPanel1.WaitForCompletion(allowPostponeAndSuspended: true, maxWait: TimeSpan.FromSeconds(2));
-        await controlPanel2.WaitForCompletion(allowPostponeAndSuspended: true, maxWait: TimeSpan.FromSeconds(2));
+        await controlPanel1.WaitForCompletion(allowPostponeAndSuspended: true, maxWait: TimeSpan.FromSeconds(5));
+        await controlPanel2.WaitForCompletion(allowPostponeAndSuspended: true, maxWait: TimeSpan.FromSeconds(5));
 
         unhandledExceptionCatcher.ShouldNotHaveExceptions();
     }

Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowState.cs

@@ -1,6 +1,6 @@
 using System.Threading;
 using System.Threading.Tasks;
-using Cleipnir.ResilientFunctions.Helpers;
+using Cleipnir.ResilientFunctions.Queuing;
 using Cleipnir.ResilientFunctions.Storage;
 
 namespace Cleipnir.ResilientFunctions.CoreRuntime;
@@ -14,7 +14,7 @@ public class FlowState
     public int Subflows { get; private set; }
     public int WaitingSubflows { get; private set; }
     public FlowTimeouts Timeouts { get; }
-    public AsyncSignal InterruptSignal { get; } = new();
+    internal QueueManager? QueueManager { get; set; }
     public bool Suspended { get; private set; }
     public Task SuspendedTask { get; }
 
@@ -67,8 +67,8 @@ public void Interrupt()
                 return;
             else
                 WaitingSubflows = 0;
-        
-        InterruptSignal.Fire();
+
+        QueueManager?.Interrupt();
     }
 
     public bool Suspend()

Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowsManager.cs

@@ -14,7 +14,7 @@ public class FlowsManager
     public FlowState CreateFlow(StoredId id, FlowTimeouts timeouts)
     {
         lock (_lock)
-            return _dict[id] = new FlowState(id, subflows: 1, waitingSubflows: 0, timeouts);;
+            return _dict[id] = new FlowState(id, subflows: 1, waitingSubflows: 0, timeouts);
     }
 
     public void RemoveFlow(StoredId id, FlowState flowState)
@@ -30,16 +30,12 @@ public IReadOnlyList<StoredId> FilterOwned(IEnumerable<StoredId> ids)
             return ids.Where(_dict.ContainsKey).ToList();
     }
 
-    public Task Interrupt(IReadOnlyList<StoredId> ids)
+    public void Interrupt(IReadOnlyList<StoredId> ids)
     {
-        /*
-         * lock (_lock)
-           foreach (var id in ids)
-               if (_dict.TryGetValue(id, out var flowState))
-                   flowState.Interrupt();
-
-         */
-        return Task.CompletedTask;
+        lock (_lock)
+            foreach (var id in ids)
+                if (_dict.TryGetValue(id, out var flowState))
+                    flowState.Interrupt();
     }
 
     public void StartThread(StoredId id)

Core/Cleipnir.ResilientFunctions/CoreRuntime/Watchdogs/InterruptedWatchdog.cs

@@ -49,7 +49,10 @@ public async Task Start()
                 var interrupted = await _functionStore.GetInterruptedFunctions();
                 var owned = _flowsManager.FilterOwned(interrupted);
                 if (owned.Count > 0)
-                    await _flowsManager.Interrupt(owned);
+                {
+                    await _functionStore.ResetInterrupted(owned);
+                    _flowsManager.Interrupt(owned);
+                }
 
                 var timeElapsed = _utcNow() - now;
                 var delay = (_checkFrequency - timeElapsed).RoundUpToZero();

Core/Cleipnir.ResilientFunctions/FunctionsRegistry.cs

@@ -76,7 +76,7 @@ public FunctionsRegistry(IFunctionStore functionStore, Settings? settings = null
         if (_settings.EnableWatchdogs)
         {
             _replicaWatchdog.Initialize().GetAwaiter().GetResult();
-            _ = _replicaWatchdog.Start();            
+            _ = _replicaWatchdog.Start();
             _ = Task.Run(_interruptedWatchdog.Start);
         }
     }

Core/Cleipnir.ResilientFunctions/Queuing/FetchedMessages.cs

@@ -55,7 +55,7 @@ public Task<Envelope> PullEnvelope<T>(Workflow workflow, EffectId parentId, Func
 
         if (!effect.Contains(messageId))
         {
-            var result = await queueManager.Subscribe(
+            return await queueManager.Subscribe(
                 envelope => filter?.Invoke(envelope) ?? true,
                 timeout,
                 timeoutId,
@@ -74,8 +74,6 @@ public Task<Envelope> PullEnvelope<T>(Workflow workflow, EffectId parentId, Func
                             EffectResult.Create(senderId, msg.Sender),
                         ]
             );
-            
-            return result;
         }
 
         if (!effect.TryGet<byte[]>(messageTypeId, out var typeNameBytes))