Restart not-running flows from FlowsManagers.Push and deliver in-hand messages

When fetched messages target flows that are not currently running on this
replica, route them through RestartExecutions instead of dropping them:

- FlowsManagers.Push now materializes the not-running messages and forwards
  them to RestartExecutions, which claims the flows for this replica and
  resumes them with the in-hand messages. Messages for a stored type not
  registered on this replica are reported via the UnhandledExceptionHandler
  (TODO: downgrade to a warning once an ILogger abstraction exists).
- FunctionsRegistry passes the UnhandledExceptionHandler into FlowsManagers.
- Invoker pushes the in-hand messages straight into the queue manager on
  restart so the flow does not re-fetch them from the store.
- QueueManager.Push initializes the queue manager first so the idempotency-key
  state is loaded before the pushed messages are processed.

Author: stidsborg

Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowsManagers.cs

@@ -4,6 +4,7 @@
 using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
 using Cleipnir.ResilientFunctions.Domain;
+using Cleipnir.ResilientFunctions.Domain.Exceptions;
 using Cleipnir.ResilientFunctions.Messaging;
 using Cleipnir.ResilientFunctions.Storage;
 
@@ -22,14 +23,16 @@ public class FlowsManagers
     private readonly IFunctionStore _functionStore;
     private readonly MessageClearer _messageClearer;
     private readonly ClusterInfo _clusterInfo;
+    private readonly UnhandledExceptionHandler _unhandledExceptionHandler;
 
     private readonly Lock _lock = new();
 
-    internal FlowsManagers(IFunctionStore functionStore, MessageClearer messageClearer, ClusterInfo clusterInfo)
+    internal FlowsManagers(IFunctionStore functionStore, MessageClearer messageClearer, ClusterInfo clusterInfo, UnhandledExceptionHandler unhandledExceptionHandler)
     {
         _functionStore = functionStore;
         _messageClearer = messageClearer;
         _clusterInfo = clusterInfo;
+        _unhandledExceptionHandler = unhandledExceptionHandler;
     }
 
     public FlowsManager GetOrCreate(StoredType storedType)
@@ -52,10 +55,12 @@ public FlowsManager GetOrCreate(StoredType storedType)
     public Task Push(IReadOnlyList<StoredMessages> messages)
     {
         List<Task> messageDeliveries;
+        List<StoredMessages> notRunning;
         lock (_lock)
         {
-            var notRunning = messages
-                .Where(msg => !_managers.ContainsKey(msg.StoredId.Type));
+            notRunning = messages
+                .Where(msg => !_managers.ContainsKey(msg.StoredId.Type))
+                .ToList();
 
             var running = messages
                 .Where(msg => _managers.ContainsKey(msg.StoredId.Type))
@@ -65,10 +70,45 @@ public Task Push(IReadOnlyList<StoredMessages> messages)
 
             messageDeliveries = running;
         }
-        
+
+        messageDeliveries.Add(RestartExecutions(notRunning));
+
         return Task.WhenAll(messageDeliveries);
     }
 
+    /// <summary>
+    /// Routes restart-and-deliver requests to each flow type's manager: groups the messages by
+    /// <see cref="StoredId.Type"/> and forwards each group to the matching <see cref="FlowsManager.RestartExecutions"/>,
+    /// which claims the not-yet-owned flows for this replica and resumes them with the in-hand messages. Messages for
+    /// a type not registered on this replica are reported to the unhandled-exception handler: this replica fetched
+    /// them (it owns or published them) yet has no flow of that type to restart, which signals a deployment mismatch.
+    /// </summary>
+    public Task RestartExecutions(IReadOnlyList<StoredMessages> messages)
+    {
+        var restarts = new List<Task>();
+        foreach (var group in messages.GroupBy(msg => msg.StoredId.Type))
+        {
+            if (TryGet(group.Key) is { } manager)
+            {
+                restarts.Add(manager.RestartExecutions(group.ToList()));
+                continue;
+            }
+
+            var unrunnable = group.ToList();
+            // TODO: this is an expected condition during deployments, not a framework fault - it should be logged as a
+            // warning rather than surfaced through the UnhandledExceptionHandler. Revisit once the framework has an
+            // ILogger abstraction.
+            _unhandledExceptionHandler.Invoke(
+                new FrameworkException(
+                    $"Fetched messages for {unrunnable.Count} flow(s) of stored type '{group.Key.Value}' which is " +
+                    $"not registered on this replica - the targeted flow(s) cannot be restarted here"
+                )
+            );
+        }
+
+        return Task.WhenAll(restarts);
+    }
+
     public IReadOnlyList<StoredId> FilterOwned(IEnumerable<StoredId> ids)
     {
         var owned = new List<StoredId>();

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

@@ -337,6 +337,12 @@ private async Task<PreparedReInvocation> PrepareForReInvocation(StoredId storedI
 
             var queueManager = _invocationHelper.CreateQueueManager(flowId, storedId, effect, flowState, flowTimeouts, _unhandledExceptionHandler);
             disposables.Add(queueManager);
+
+            // Deliver the in-hand messages handed over by the restart straight into the queue manager's pipeline so
+            // the flow does not have to re-fetch them from the store. Push initializes the queue manager first, which
+            // loads the idempotency-key state before these messages are processed.
+            await queueManager.Push(storedMessages);
+
             var messageWriter = _invocationHelper.CreateMessageWriter(storedId);
 
             var workflow = new Workflow(

Core/Cleipnir.ResilientFunctions/FunctionsRegistry.cs

@@ -50,7 +50,7 @@ public FunctionsRegistry(IFunctionStore functionStore, Settings? settings = null
         );
         ClusterInfo = new ClusterInfo(ReplicaId.NewId());
 
-        _flowsManagers = new FlowsManagers(_functionStore, _messageClearer, ClusterInfo);
+        _flowsManagers = new FlowsManagers(_functionStore, _messageClearer, ClusterInfo, _settings.UnhandledExceptionHandler);
 
         _postponedWatchdog = new PostponedWatchdog(
             _functionStore,

Core/Cleipnir.ResilientFunctions/Queuing/QueueManager.cs

@@ -122,14 +122,19 @@ internal void Interrupt()
     }
 
     /// <summary>
-    /// Pushes messages fetched elsewhere (the MessageWatchdog) straight into the delivery pipeline,
-    /// avoiding a per-flow re-fetch. Idempotent: positions already processed are skipped by ProcessMessages.
+    /// Pushes messages fetched elsewhere (the MessageWatchdog, or the in-hand messages handed over on restart)
+    /// straight into the delivery pipeline, avoiding a per-flow re-fetch. Ensures the queue manager is initialized
+    /// first so the idempotency-key state is loaded before the messages are processed. Idempotent: positions
+    /// already processed are skipped by ProcessMessages.
     /// </summary>
     public async Task Push(IReadOnlyList<StoredMessage> messages)
     {
         if (_disposed || messages.Count == 0)
             return;
 
+        if (!_initialized)
+            await Initialize();
+
         await _fetchSemaphore.WaitAsync();
         try
         {