Strip empty messages when building the restart hand-over instead of in QueueManager

Author: stidsborg

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

@@ -29,4 +29,8 @@ public override Task EmptyMessagesRestartSuspendedFlowsWithoutDeliveryAndAreRemo
     [TestMethod]
     public override Task EmptyMessageIsNotDeliveredToRestartedFlowWhileNonEmptyMessageIs()
         => EmptyMessageIsNotDeliveredToRestartedFlowWhileNonEmptyMessageIs(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel()
+        => EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel(FunctionStoreFactory.Create());
 }

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

@@ -228,4 +228,59 @@ await store.MessageStore.AppendMessages([
 
         unhandledExceptionHandler.ShouldNotHaveExceptions();
     }
+
+    public abstract Task EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel();
+    public async Task EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel(Task<IFunctionStore> functionStore)
+    {
+        var store = await functionStore;
+
+        var flowId = TestFlowId.Create();
+        var unhandledExceptionHandler = new UnhandledExceptionCatcher();
+        // Watchdogs are disabled so the pending messages can only reach the flow via the control panel restart's
+        // hand-over (the RestartExecution route) - not via the MessageWatchdog route.
+        using var functionsRegistry = new FunctionsRegistry(
+            store,
+            new Settings(unhandledExceptionHandler.Catch, enableWatchdogs: false)
+        );
+
+        var registration = functionsRegistry.RegisterFunc(
+            flowId.Type,
+            inner: async Task<string> (string _, Workflow workflow) => await workflow.Message<string>()
+        );
+
+        await registration.Schedule(flowId.Instance.Value, "");
+
+        var controlPanel = await registration.ControlPanel(flowId.Instance);
+        controlPanel.ShouldNotBeNull();
+        await controlPanel.BusyWaitUntil(c => c.Status == Status.Suspended);
+
+        var storedId = registration.MapToStoredId(flowId.Instance);
+        var replicaId = functionsRegistry.ClusterInfo.ReplicaId;
+        var serializer = DefaultSerializer.Instance;
+        await store.MessageStore.AppendMessages([
+            new StoredIdAndMessage(storedId, StoredMessage.CreateEmpty(replicaId)),
+            new StoredIdAndMessage(
+                storedId,
+                new StoredMessage(
+                    serializer.Serialize("hello world", typeof(string)),
+                    serializer.SerializeType(typeof(string)),
+                    Position: 0,
+                    Replica: replicaId
+                )
+            )
+        ]);
+
+        await controlPanel.ScheduleRestart();
+
+        // The restarted flow receives only the non-empty message.
+        await controlPanel.WaitForCompletion(allowPostponeAndSuspended: true);
+        await controlPanel.Refresh();
+        controlPanel.Status.ShouldBe(Status.Succeeded);
+        controlPanel.Result.ShouldBe("hello world");
+
+        // The empty message is deleted by the restart hand-over; the delivered one is deleted after delivery.
+        await BusyWait.Until(async () => (await store.MessageStore.GetMessages(storedId)).Count == 0);
+
+        unhandledExceptionHandler.ShouldNotHaveExceptions();
+    }
 }

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

@@ -221,15 +221,25 @@ public async Task PublishCompletionMessageToParent(StoredId? parent, FlowId chil
             flowId,
             _replicaId
         );
+        if (restarted == null)
+            return null;
 
-        return restarted != null
-            ? new RestartedFunction(
-                restarted.StoredFlow, 
-                restarted.Effects,
-                restarted.Messages,
-                restarted.StorageSession
-            ) 
-            : null;
+        // The message snapshot comes straight from the store and can contain empty messages - restart-pokes with
+        // nothing to deliver. The restart they were appended to force is happening right now, so delete them and
+        // hand over only the deliverable messages.
+        var messages = restarted.Messages;
+        if (messages.Any(m => m.IsEmpty))
+        {
+            await _messageClearer.Clear(messages.Where(m => m.IsEmpty).Select(m => m.Position).ToList());
+            messages = messages.Where(m => !m.IsEmpty).ToList();
+        }
+
+        return new RestartedFunction(
+            restarted.StoredFlow,
+            restarted.Effects,
+            messages,
+            restarted.StorageSession
+        );
     }
 
     public async Task<PreparedReInvocation> PrepareForReInvocation(StoredId storedId, RestartedFunction restartedFunction)

Core/Cleipnir.ResilientFunctions/Queuing/QueueManager.cs

@@ -133,7 +133,8 @@ public Task FetchMessagesOnce()
     /// 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.
+    /// already processed are skipped by ProcessMessages. Both routes strip empty (restart-poke) messages before
+    /// handing over, so every message arriving here carries a deliverable payload.
     ///
     /// A push that hits a disposed (dying) instance reopens its positions instead of dropping them: the
     /// MessageWatchdog already marked them as pushed, so a silent drop would strand the messages in the
@@ -163,19 +164,6 @@ public async Task Push(IReadOnlyList<StoredMessage> messages)
             }
         }
 
-        // Empty messages exist only to force a restart and carry nothing to deliver (deserializing one would
-        // poison the queue manager). The only push that can contain them is a restart hand-over - the watchdog
-        // route strips them before delivery - so the restart they were appended to force has happened: delete
-        // them instead of delivering.
-        if (messages.Any(m => m.IsEmpty))
-        {
-            var emptyPositions = messages.Where(m => m.IsEmpty).Select(m => m.Position).ToList();
-            messages = messages.Where(m => !m.IsEmpty).ToList();
-            await _messageClearer.Clear(emptyPositions);
-            if (messages.Count == 0)
-                return;
-        }
-
         await _fetchSemaphore.WaitAsync();
         try
         {

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

@@ -24,4 +24,8 @@ public override Task EmptyMessagesRestartSuspendedFlowsWithoutDeliveryAndAreRemo
     [TestMethod]
     public override Task EmptyMessageIsNotDeliveredToRestartedFlowWhileNonEmptyMessageIs()
         => EmptyMessageIsNotDeliveredToRestartedFlowWhileNonEmptyMessageIs(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel()
+        => EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel(FunctionStoreFactory.Create());
 }

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

@@ -25,4 +25,8 @@ public override Task EmptyMessagesRestartSuspendedFlowsWithoutDeliveryAndAreRemo
     [TestMethod]
     public override Task EmptyMessageIsNotDeliveredToRestartedFlowWhileNonEmptyMessageIs()
         => EmptyMessageIsNotDeliveredToRestartedFlowWhileNonEmptyMessageIs(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel()
+        => EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel(FunctionStoreFactory.Create());
 }

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

@@ -25,4 +25,8 @@ public override Task EmptyMessagesRestartSuspendedFlowsWithoutDeliveryAndAreRemo
     [TestMethod]
     public override Task EmptyMessageIsNotDeliveredToRestartedFlowWhileNonEmptyMessageIs()
         => EmptyMessageIsNotDeliveredToRestartedFlowWhileNonEmptyMessageIs(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel()
+        => EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel(FunctionStoreFactory.Create());
 }