Add RestartExecutionsWithoutMessages to IFunctionStore

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

@@ -75,4 +75,20 @@ public override Task RestartExecutionsReturnsEmptyDictionaryForEmptyInput()
     [TestMethod]
     public override Task RestartExecutionsIncludesExistingEffectsAndMessages()
         => RestartExecutionsIncludesExistingEffectsAndMessages(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows()
+        => RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task RestartExecutionsWithoutMessagesRestartsOnlyUnownedFlows()
+        => RestartExecutionsWithoutMessagesRestartsOnlyUnownedFlows(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task RestartExecutionsWithoutMessagesReturnsEmptyDictionaryForEmptyInput()
+        => RestartExecutionsWithoutMessagesReturnsEmptyDictionaryForEmptyInput(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task RestartExecutionsWithoutMessagesIncludesExistingEffects()
+        => RestartExecutionsWithoutMessagesIncludesExistingEffects(FunctionStoreFactory.Create());
 }

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

@@ -555,4 +555,207 @@ await store.CreateFunction(
         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)
+    {
+        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.RestartExecutionsWithoutMessages([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 RestartExecutionsWithoutMessagesRestartsOnlyUnownedFlows();
+    public async Task RestartExecutionsWithoutMessagesRestartsOnlyUnownedFlows(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.RestartExecutionsWithoutMessages([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 RestartExecutionsWithoutMessagesReturnsEmptyDictionaryForEmptyInput();
+    public async Task RestartExecutionsWithoutMessagesReturnsEmptyDictionaryForEmptyInput(Task<IFunctionStore> storeTask)
+    {
+        var store = await storeTask;
+        var owner = ReplicaId.NewId();
+
+        var result = await store.RestartExecutionsWithoutMessages([], owner);
+
+        result.Count.ShouldBe(0);
+    }
+
+    public abstract Task RestartExecutionsWithoutMessagesIncludesExistingEffects();
+    public async Task RestartExecutionsWithoutMessagesIncludesExistingEffects(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 - these must NOT be fetched by RestartExecutionsWithoutMessages
+        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.RestartExecutionsWithoutMessages([storedId1, storedId2], owner);
+
+        // Verify both flows returned with their effects
+        result.Count.ShouldBe(2);
+
+        result.ContainsKey(storedId1).ShouldBeTrue();
+        result.ContainsKey(storedId2).ShouldBeTrue();
+
+        // Verify flow 1 has correct effects
+        var flow1 = result[storedId1];
+        flow1.Effects.Count.ShouldBe(1);
+        flow1.Effects[0].EffectId.ShouldBe("effect1".GetHashCode().ToEffectId());
+        flow1.Effects[0].Result.ShouldBe("result1".ToUtf8Bytes());
+
+        // Verify flow 2 has correct effects
+        var flow2 = result[storedId2];
+        flow2.Effects.Count.ShouldBe(1);
+        flow2.Effects[0].EffectId.ShouldBe("effect2".GetHashCode().ToEffectId());
+        flow2.Effects[0].Result.ShouldBe("result2".ToUtf8Bytes());
+    }
 }

Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/WatchDogsTests/CrashableFunctionStore.cs

@@ -81,6 +81,11 @@ public Task<Dictionary<StoredId, StoredFlowWithEffectsAndMessages>> RestartExecu
             ? Task.FromException<Dictionary<StoredId, StoredFlowWithEffectsAndMessages>>(new TimeoutException())
             : _inner.RestartExecutions(storedIds, owner);
 
+    public Task<Dictionary<StoredId, StoredFlowWithEffects>> RestartExecutionsWithoutMessages(IReadOnlyList<StoredId> storedIds, ReplicaId owner)
+        => _crashed
+            ? Task.FromException<Dictionary<StoredId, StoredFlowWithEffects>>(new TimeoutException())
+            : _inner.RestartExecutionsWithoutMessages(storedIds, owner);
+
     public Task<IReadOnlyList<StoredId>> GetExpiredFunctions(long isEligibleBefore)
         => _crashed
             ? Task.FromException<IReadOnlyList<StoredId>>(new TimeoutException())

Core/Cleipnir.ResilientFunctions/Storage/IFunctionStore.cs

@@ -33,6 +33,7 @@ Task<int> BulkScheduleFunctions(
 
     Task<StoredFlowWithEffectsAndMessages?> RestartExecution(StoredId storedId, ReplicaId owner);
     Task<Dictionary<StoredId, StoredFlowWithEffectsAndMessages>> RestartExecutions(IReadOnlyList<StoredId> storedIds, ReplicaId owner);
+    Task<Dictionary<StoredId, StoredFlowWithEffects>> RestartExecutionsWithoutMessages(IReadOnlyList<StoredId> storedIds, ReplicaId owner);
 
     Task<IReadOnlyList<StoredId>> GetExpiredFunctions(long expiresBefore);
     Task<IReadOnlyList<StoredId>> GetSucceededFunctions(long completedBefore);

Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs

@@ -201,7 +201,64 @@ public virtual async Task<Dictionary<StoredId, StoredFlowWithEffectsAndMessages>
 
         return result;
     }
-
+
+    public virtual async Task<Dictionary<StoredId, StoredFlowWithEffects>> RestartExecutionsWithoutMessages(
+        IReadOnlyList<StoredId> storedIds,
+        ReplicaId owner)
+    {
+        if (storedIds.Count == 0)
+            return new Dictionary<StoredId, StoredFlowWithEffects>();
+
+        var restartedIds = new List<StoredId>();
+
+        // Restart eligible flows
+        lock (_sync)
+        {
+            foreach (var storedId in storedIds)
+            {
+                if (!_states.ContainsKey(storedId))
+                    continue;
+
+                var state = _states[storedId];
+                if (state.Owner != null)
+                    continue; // Skip already owned flows
+
+                // Restart this flow
+                state.Status = Status.Executing;
+                state.Expires = 0;
+                state.Interrupted = false;
+                state.Owner = owner;
+
+                restartedIds.Add(storedId);
+            }
+        }
+
+        var effectsDict = await EffectsStore.GetEffectResults(storedIds);
+
+        // Build result dictionary - only for restarted flows
+        var result = new Dictionary<StoredId, StoredFlowWithEffects>();
+        foreach (var storedId in restartedIds)
+        {
+            var sf = await GetFunction(storedId);
+            if (sf == null) continue;
+
+            var effects = effectsDict.TryGetValue(storedId, out var effs)
+                ? effs
+                : new List<StoredEffect>();
+
+            var session = new SnapshotStorageSession(owner);
+            foreach (var effect in effects)
+                session.Effects[effect.EffectId] = effect;
+
+            session.Version = _effectsStore.GetVersion(storedId);
+            session.RowExists = effects.Any();
+
+            result[storedId] = new StoredFlowWithEffects(sf, effects, session);
+        }
+
+        return result;
+    }
+
     public virtual Task<IReadOnlyList<StoredId>> GetExpiredFunctions(long expiresBefore)
     {
         lock (_sync)

Core/Cleipnir.ResilientFunctions/Storage/Types.cs

@@ -182,6 +182,12 @@ public static StoredEffect Deserialize(byte[] bytes)
 
 public record IdWithParam(StoredId StoredId, string HumanInstanceId, byte[]? Param);
 
+public record StoredFlowWithEffects(
+    StoredFlow StoredFlow,
+    IReadOnlyList<StoredEffect> Effects,
+    IStorageSession StorageSession
+);
+
 public record StoredFlowWithEffectsAndMessages(
     StoredFlow StoredFlow,
     IReadOnlyList<StoredEffect> Effects,

Samples/Sample.ConsoleApp/Utils/CrashableFunctionStore.cs

@@ -63,6 +63,11 @@ public Task<Dictionary<StoredId, StoredFlowWithEffectsAndMessages>> RestartExecu
             ? Task.FromException<Dictionary<StoredId, StoredFlowWithEffectsAndMessages>>(new TimeoutException())
             : _inner.RestartExecutions(storedIds, owner);
 
+    public Task<Dictionary<StoredId, StoredFlowWithEffects>> RestartExecutionsWithoutMessages(IReadOnlyList<StoredId> storedIds, ReplicaId owner)
+        => _crashed
+            ? Task.FromException<Dictionary<StoredId, StoredFlowWithEffects>>(new TimeoutException())
+            : _inner.RestartExecutionsWithoutMessages(storedIds, owner);
+
     public Task<IReadOnlyList<StoredId>> GetExpiredFunctions(long expiresBefore)
         => _crashed
             ? Task.FromException<IReadOnlyList<StoredId>>(new TimeoutException())

Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/StoreCrudTests.cs

@@ -72,4 +72,20 @@ public override Task RestartExecutionsReturnsEmptyDictionaryForEmptyInput()
     [TestMethod]
     public override Task RestartExecutionsIncludesExistingEffectsAndMessages()
         => RestartExecutionsIncludesExistingEffectsAndMessages(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows()
+        => RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task RestartExecutionsWithoutMessagesRestartsOnlyUnownedFlows()
+        => RestartExecutionsWithoutMessagesRestartsOnlyUnownedFlows(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task RestartExecutionsWithoutMessagesReturnsEmptyDictionaryForEmptyInput()
+        => RestartExecutionsWithoutMessagesReturnsEmptyDictionaryForEmptyInput(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task RestartExecutionsWithoutMessagesIncludesExistingEffects()
+        => RestartExecutionsWithoutMessagesIncludesExistingEffects(FunctionStoreFactory.Create());
 }