Remove message hand-over from restarts - the MessageWatchdog is the sole message-fetch path

RestartExecution returns effects only; the with-messages RestartExecutions,
StoredFlowWithEffectsAndMessages and the RestartFunction delegate are removed.
Completed flows' undeliverable positions are parked per flow (park-then-check
ordering to avoid stranding them when a concurrent explicit restart reopens
parked positions) and released again on explicit re-invocation. ExistingMessages
stamps appended messages with the publisher replica so the watchdog routes them;
QueueManager reopens delivered-but-not-deleted positions on unwind.

Author: stidsborg

Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/StoreCrudTests.cs

@@ -56,25 +56,13 @@ public override Task StateCanBeUpdatedOnExistingFunction()
     public override Task ParameterAndStateAreNotUpdatedWhenEpochDoesNotMatch()
         => ParameterAndStateAreNotUpdatedWhenEpochDoesNotMatch(FunctionStoreFactory.Create());
 
-    [TestMethod]
-    public override Task RestartExecutionsRestartsMultipleUnownedFlows()
-        => RestartExecutionsRestartsMultipleUnownedFlows(FunctionStoreFactory.Create());
 
     [TestMethod]
-    public override Task RestartExecutionsReturnsEmptyDictionaryWhenNoFlowsAreEligible()
-        => RestartExecutionsReturnsEmptyDictionaryWhenNoFlowsAreEligible(FunctionStoreFactory.Create());
+    public override Task RestartExecutionsWithoutMessagesReturnsEmptyDictionaryWhenNoFlowsAreEligible()
+        => RestartExecutionsWithoutMessagesReturnsEmptyDictionaryWhenNoFlowsAreEligible(FunctionStoreFactory.Create());
 
-    [TestMethod]
-    public override Task RestartExecutionsRestartsOnlyUnownedFlows()
-        => RestartExecutionsRestartsOnlyUnownedFlows(FunctionStoreFactory.Create());
 
-    [TestMethod]
-    public override Task RestartExecutionsReturnsEmptyDictionaryForEmptyInput()
-        => RestartExecutionsReturnsEmptyDictionaryForEmptyInput(FunctionStoreFactory.Create());
 
-    [TestMethod]
-    public override Task RestartExecutionsIncludesExistingEffectsAndMessages()
-        => RestartExecutionsIncludesExistingEffectsAndMessages(FunctionStoreFactory.Create());
 
     [TestMethod]
     public override Task RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows()

Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/StoreTests.cs

@@ -171,12 +171,12 @@ public override Task FunctionCanBeCreatedWithEffectsOnly()
         => FunctionCanBeCreatedWithEffectsOnly(FunctionStoreFactory.Create());
 
     [TestMethod]
-    public override Task RestartExecutionReturnsEffectsAndMessages()
-        => RestartExecutionReturnsEffectsAndMessages(FunctionStoreFactory.Create());
+    public override Task RestartExecutionReturnsEffects()
+        => RestartExecutionReturnsEffects(FunctionStoreFactory.Create());
 
     [TestMethod]
-    public override Task RestartExecutionWorksWithEmptyEffectsAndMessages()
-        => RestartExecutionWorksWithEmptyEffectsAndMessages(FunctionStoreFactory.Create());
+    public override Task RestartExecutionWorksWithEmptyEffects()
+        => RestartExecutionWorksWithEmptyEffects(FunctionStoreFactory.Create());
 
     [TestMethod]
     public override Task FunctionOwnedByReplicaIsPostponedAfterRescheduleFunctionsInvocation()
@@ -210,9 +210,6 @@ public override Task GetResultsReturnsEmptyDictionaryForEmptyInput()
     public override Task GetResultsReturnsOnlyExistingFunctionResults()
         => GetResultsReturnsOnlyExistingFunctionResults(FunctionStoreFactory.Create());
 
-    [TestMethod]
-    public override Task RestartExecutionsDoesNotReturnFlowClaimedByPreviousCall()
-        => RestartExecutionsDoesNotReturnFlowClaimedByPreviousCall(FunctionStoreFactory.Create());
 
     [TestMethod]
     public override Task RestartExecutionsWithoutMessagesDoesNotReturnFlowClaimedByPreviousCall()

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

@@ -1,4 +1,5 @@
 using System;
+using System.Linq;
 using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.CoreRuntime.Invocation;
 using Cleipnir.ResilientFunctions.CoreRuntime.Serialization;
@@ -236,23 +237,20 @@ public async Task EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel(T
 
         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)
+            new Settings(unhandledExceptionHandler.Catch, messagesPullFrequency: TimeSpan.FromMilliseconds(100))
         );
 
+        var awaitMessage = new SyncedFlag();
         var registration = functionsRegistry.RegisterFunc(
             flowId.Type,
-            inner: async Task<string> (string _, Workflow workflow) => await workflow.Message<string>()
+            inner: async Task<string> (string _, Workflow workflow) =>
+                awaitMessage.IsRaised ? await workflow.Message<string>() : "no message awaited"
         );
 
-        await registration.Schedule(flowId.Instance.Value, "");
-
-        var controlPanel = await registration.ControlPanel(flowId.Instance);
-        controlPanel.ShouldNotBeNull();
-        await controlPanel.BusyWaitUntil(c => c.Status == Status.Suspended);
+        // Complete the flow without it consuming any messages.
+        await registration.Run(flowId.Instance.Value, "");
 
         var storedId = registration.MapToStoredId(flowId.Instance);
         var replicaId = functionsRegistry.ClusterInfo.ReplicaId;
@@ -270,16 +268,29 @@ await store.MessageStore.AppendMessages([
             )
         ]);
 
-        await controlPanel.ScheduleRestart();
+        // The pending messages must not resurrect the completed flow.
+        await Task.Delay(500);
+        var controlPanel = await registration.ControlPanel(flowId.Instance);
+        controlPanel.ShouldNotBeNull();
+        controlPanel.Status.ShouldBe(Status.Succeeded);
+        controlPanel.Result.ShouldBe("no message awaited");
 
-        // The restarted flow receives only the non-empty message.
+        // An explicit restart does not pull messages itself - the MessageWatchdog delivers the pending non-empty
+        // message to the restarted flow, while the empty message is never delivered.
+        awaitMessage.Raise();
+        await controlPanel.ScheduleRestart();
         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);
+        // The delivered message is deleted after delivery. The empty message is either consumed by a watchdog
+        // restart along the way (deleted) or remains parked for the completed flow - but is never delivered.
+        await BusyWait.Until(async () =>
+        {
+            var remaining = await store.MessageStore.GetMessages(storedId);
+            return remaining.All(m => m.IsEmpty);
+        });
 
         unhandledExceptionHandler.ShouldNotHaveExceptions();
     }

Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/StoreCrudTests.cs

@@ -312,63 +312,8 @@ await store.SetParameters(
         sf.Parameter.ShouldNotBeNull();
     }
 
-    public abstract Task RestartExecutionsRestartsMultipleUnownedFlows();
-    public async Task RestartExecutionsRestartsMultipleUnownedFlows(Task<IFunctionStore> storeTask)
-    {
-        var store = await storeTask;
-        var storedId1 = TestStoredId.Create();
-        var storedId2 = TestStoredId.Create();
-        var storedId3 = TestStoredId.Create();
-        var owner = ReplicaId.NewId();
-
-        // Create 3 functions without owners (postponed)
-        await store.CreateFunction(
-            storedId1,
-            "instance1",
-            Param.ToUtf8Bytes(),
-            postponeUntil: DateTime.UtcNow.Ticks,
-            timestamp: DateTime.UtcNow.Ticks,
-            parent: null,
-            owner: null
-        );
-        await store.CreateFunction(
-            storedId2,
-            "instance2",
-            Param.ToUtf8Bytes(),
-            postponeUntil: DateTime.UtcNow.Ticks,
-            timestamp: DateTime.UtcNow.Ticks,
-            parent: null,
-            owner: null
-        );
-        await store.CreateFunction(
-            storedId3,
-            "instance3",
-            Param.ToUtf8Bytes(),
-            postponeUntil: DateTime.UtcNow.Ticks,
-            timestamp: DateTime.UtcNow.Ticks,
-            parent: null,
-            owner: null
-        );
-
-        // Restart all three
-        var result = await store.RestartExecutions([storedId1, storedId2, storedId3], owner);
-
-        // All three should be restarted
-        result.Count.ShouldBe(3);
-        result.ContainsKey(storedId1).ShouldBeTrue();
-        result.ContainsKey(storedId2).ShouldBeTrue();
-        result.ContainsKey(storedId3).ShouldBeTrue();
-
-        result[storedId1].StoredFlow.OwnerId.ShouldBe(owner);
-        result[storedId1].StoredFlow.Status.ShouldBe(Status.Executing);
-        result[storedId2].StoredFlow.OwnerId.ShouldBe(owner);
-        result[storedId2].StoredFlow.Status.ShouldBe(Status.Executing);
-        result[storedId3].StoredFlow.OwnerId.ShouldBe(owner);
-        result[storedId3].StoredFlow.Status.ShouldBe(Status.Executing);
-    }
-
-    public abstract Task RestartExecutionsReturnsEmptyDictionaryWhenNoFlowsAreEligible();
-    public async Task RestartExecutionsReturnsEmptyDictionaryWhenNoFlowsAreEligible(Task<IFunctionStore> storeTask)
+    public abstract Task RestartExecutionsWithoutMessagesReturnsEmptyDictionaryWhenNoFlowsAreEligible();
+    public async Task RestartExecutionsWithoutMessagesReturnsEmptyDictionaryWhenNoFlowsAreEligible(Task<IFunctionStore> storeTask)
     {
         var store = await storeTask;
         var storedId1 = TestStoredId.Create();
@@ -397,165 +342,11 @@ await store.CreateFunction(
         );
 
         // Try to restart - should return empty dictionary
-        var result = await store.RestartExecutions([storedId1, storedId2], newOwner);
+        var result = await store.RestartExecutionsWithoutMessages([storedId1, storedId2], newOwner);
 
         result.Count.ShouldBe(0);
     }
 
-    public abstract Task RestartExecutionsRestartsOnlyUnownedFlows();
-    public async Task RestartExecutionsRestartsOnlyUnownedFlows(Task<IFunctionStore> storeTask)
-    {
-        var store = await storeTask;
-        var storedId1 = TestStoredId.Create();
-        var storedId2 = TestStoredId.Create();
-        var storedId3 = TestStoredId.Create();
-        var existingOwner = ReplicaId.NewId();
-        var newOwner = ReplicaId.NewId();
-
-        // Create 2 functions without owners and 1 with owner
-        await store.CreateFunction(
-            storedId1,
-            "instance1",
-            Param.ToUtf8Bytes(),
-            postponeUntil: DateTime.UtcNow.Ticks,
-            timestamp: DateTime.UtcNow.Ticks,
-            parent: null,
-            owner: null
-        );
-        await store.CreateFunction(
-            storedId2,
-            "instance2",
-            Param.ToUtf8Bytes(),
-            postponeUntil: null,
-            timestamp: DateTime.UtcNow.Ticks,
-            parent: null,
-            owner: existingOwner
-        );
-        await store.CreateFunction(
-            storedId3,
-            "instance3",
-            Param.ToUtf8Bytes(),
-            postponeUntil: DateTime.UtcNow.Ticks,
-            timestamp: DateTime.UtcNow.Ticks,
-            parent: null,
-            owner: null
-        );
-
-        // Restart all three - only 1 and 3 should succeed
-        var result = await store.RestartExecutions([storedId1, storedId2, storedId3], newOwner);
-
-        result.Count.ShouldBe(2);
-        result.ContainsKey(storedId1).ShouldBeTrue();
-        result.ContainsKey(storedId2).ShouldBeFalse();
-        result.ContainsKey(storedId3).ShouldBeTrue();
-
-        result[storedId1].StoredFlow.OwnerId.ShouldBe(newOwner);
-        result[storedId3].StoredFlow.OwnerId.ShouldBe(newOwner);
-    }
-
-    public abstract Task RestartExecutionsReturnsEmptyDictionaryForEmptyInput();
-    public async Task RestartExecutionsReturnsEmptyDictionaryForEmptyInput(Task<IFunctionStore> storeTask)
-    {
-        var store = await storeTask;
-        var owner = ReplicaId.NewId();
-
-        var result = await store.RestartExecutions([], owner);
-
-        result.Count.ShouldBe(0);
-    }
-
-    public abstract Task RestartExecutionsIncludesExistingEffectsAndMessages();
-    public async Task RestartExecutionsIncludesExistingEffectsAndMessages(Task<IFunctionStore> storeTask)
-    {
-        var store = await storeTask;
-        var storedId1 = TestStoredId.Create();
-        var storedId2 = TestStoredId.Create();
-        var owner = ReplicaId.NewId();
-
-        // Create effects
-        var effect1 = new StoredEffect(
-            EffectId: "effect1".GetHashCode().ToEffectId(),
-            WorkStatus: WorkStatus.Completed,
-            Result: "result1".ToUtf8Bytes(),
-            StoredException: null,
-            Alias: null
-        );
-        var effect2 = new StoredEffect(
-            EffectId: "effect2".GetHashCode().ToEffectId(),
-            WorkStatus: WorkStatus.Completed,
-            Result: "result2".ToUtf8Bytes(),
-            StoredException: null,
-            Alias: null
-        );
-
-        // Create messages
-        var message1 = new StoredMessage(
-            MessageContent: "message1".ToUtf8Bytes(),
-            MessageType: "Type1".ToUtf8Bytes(),
-            Position: 0,
-            Replica: ReplicaId.Empty,
-            IdempotencyKey: null
-        );
-        var message2 = new StoredMessage(
-            MessageContent: "message2".ToUtf8Bytes(),
-            MessageType: "Type2".ToUtf8Bytes(),
-            Position: 1,
-            Replica: ReplicaId.Empty,
-            IdempotencyKey: null
-        );
-
-        // Create 2 functions with effects and messages
-        await store.CreateFunction(
-            storedId1,
-            "instance1",
-            Param.ToUtf8Bytes(),
-            postponeUntil: DateTime.UtcNow.Ticks,
-            timestamp: DateTime.UtcNow.Ticks,
-            parent: null,
-            owner: null,
-            effects: [effect1],
-            messages: [message1]
-        );
-        await store.CreateFunction(
-            storedId2,
-            "instance2",
-            Param.ToUtf8Bytes(),
-            postponeUntil: DateTime.UtcNow.Ticks,
-            timestamp: DateTime.UtcNow.Ticks,
-            parent: null,
-            owner: null,
-            effects: [effect2],
-            messages: [message2]
-        );
-
-        // Restart both
-        var result = await store.RestartExecutions([storedId1, storedId2], owner);
-
-        // Verify both flows returned with their effects and messages
-        result.Count.ShouldBe(2);
-
-        result.ContainsKey(storedId1).ShouldBeTrue();
-        result.ContainsKey(storedId2).ShouldBeTrue();
-
-        // Verify flow 1 has correct effects and messages
-        var flow1 = result[storedId1];
-        flow1.Effects.Count.ShouldBe(1);
-        flow1.Effects[0].EffectId.ShouldBe("effect1".GetHashCode().ToEffectId());
-        flow1.Effects[0].Result.ShouldBe("result1".ToUtf8Bytes());
-        flow1.Messages.Count.ShouldBe(1);
-        flow1.Messages[0].MessageContent.ShouldBe("message1".ToUtf8Bytes());
-        flow1.Messages[0].MessageType.ShouldBe("Type1".ToUtf8Bytes());
-
-        // Verify flow 2 has correct effects and messages
-        var flow2 = result[storedId2];
-        flow2.Effects.Count.ShouldBe(1);
-        flow2.Effects[0].EffectId.ShouldBe("effect2".GetHashCode().ToEffectId());
-        flow2.Effects[0].Result.ShouldBe("result2".ToUtf8Bytes());
-        flow2.Messages.Count.ShouldBe(1);
-        flow2.Messages[0].MessageContent.ShouldBe("message2".ToUtf8Bytes());
-        flow2.Messages[0].MessageType.ShouldBe("Type2".ToUtf8Bytes());
-    }
-
     public abstract Task RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows();
     public async Task RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows(Task<IFunctionStore> storeTask)
     {