diff --git a/Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/StoreCrudTests.cs b/Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/StoreCrudTests.cs index b9a1df835..cb14fb864 100644 --- a/Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/StoreCrudTests.cs +++ b/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()); } \ No newline at end of file diff --git a/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/StoreCrudTests.cs b/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/StoreCrudTests.cs index 082f759f9..87641d689 100644 --- a/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/StoreCrudTests.cs +++ b/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 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 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 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 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()); + } } \ No newline at end of file diff --git a/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/WatchDogsTests/CrashableFunctionStore.cs b/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/WatchDogsTests/CrashableFunctionStore.cs index 6c2aeaff3..a83a36f2b 100644 --- a/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/WatchDogsTests/CrashableFunctionStore.cs +++ b/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/WatchDogsTests/CrashableFunctionStore.cs @@ -81,6 +81,11 @@ public Task> RestartExecu ? Task.FromException>(new TimeoutException()) : _inner.RestartExecutions(storedIds, owner); + public Task> RestartExecutionsWithoutMessages(IReadOnlyList storedIds, ReplicaId owner) + => _crashed + ? Task.FromException>(new TimeoutException()) + : _inner.RestartExecutionsWithoutMessages(storedIds, owner); + public Task> GetExpiredFunctions(long isEligibleBefore) => _crashed ? Task.FromException>(new TimeoutException()) diff --git a/Core/Cleipnir.ResilientFunctions/Storage/IFunctionStore.cs b/Core/Cleipnir.ResilientFunctions/Storage/IFunctionStore.cs index d272f00a2..86b0ede66 100644 --- a/Core/Cleipnir.ResilientFunctions/Storage/IFunctionStore.cs +++ b/Core/Cleipnir.ResilientFunctions/Storage/IFunctionStore.cs @@ -33,6 +33,7 @@ Task BulkScheduleFunctions( Task RestartExecution(StoredId storedId, ReplicaId owner); Task> RestartExecutions(IReadOnlyList storedIds, ReplicaId owner); + Task> RestartExecutionsWithoutMessages(IReadOnlyList storedIds, ReplicaId owner); Task> GetExpiredFunctions(long expiresBefore); Task> GetSucceededFunctions(long completedBefore); diff --git a/Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs b/Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs index af648b51b..b1886b9f6 100644 --- a/Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs +++ b/Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs @@ -201,7 +201,64 @@ public virtual async Task return result; } - + + public virtual async Task> RestartExecutionsWithoutMessages( + IReadOnlyList storedIds, + ReplicaId owner) + { + if (storedIds.Count == 0) + return new Dictionary(); + + var restartedIds = new List(); + + // 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(); + 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(); + + 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> GetExpiredFunctions(long expiresBefore) { lock (_sync) diff --git a/Core/Cleipnir.ResilientFunctions/Storage/Types.cs b/Core/Cleipnir.ResilientFunctions/Storage/Types.cs index f88c9ec8a..357390153 100644 --- a/Core/Cleipnir.ResilientFunctions/Storage/Types.cs +++ b/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 Effects, + IStorageSession StorageSession +); + public record StoredFlowWithEffectsAndMessages( StoredFlow StoredFlow, IReadOnlyList Effects, diff --git a/Samples/Sample.ConsoleApp/Utils/CrashableFunctionStore.cs b/Samples/Sample.ConsoleApp/Utils/CrashableFunctionStore.cs index 8a0f807e5..cfc9a4b2d 100644 --- a/Samples/Sample.ConsoleApp/Utils/CrashableFunctionStore.cs +++ b/Samples/Sample.ConsoleApp/Utils/CrashableFunctionStore.cs @@ -63,6 +63,11 @@ public Task> RestartExecu ? Task.FromException>(new TimeoutException()) : _inner.RestartExecutions(storedIds, owner); + public Task> RestartExecutionsWithoutMessages(IReadOnlyList storedIds, ReplicaId owner) + => _crashed + ? Task.FromException>(new TimeoutException()) + : _inner.RestartExecutionsWithoutMessages(storedIds, owner); + public Task> GetExpiredFunctions(long expiresBefore) => _crashed ? Task.FromException>(new TimeoutException()) diff --git a/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/StoreCrudTests.cs b/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/StoreCrudTests.cs index 3ab9feafb..de0b24355 100644 --- a/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/StoreCrudTests.cs +++ b/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()); } \ No newline at end of file diff --git a/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbFunctionStore.cs b/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbFunctionStore.cs index acfc1911e..d3740e65a 100644 --- a/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbFunctionStore.cs +++ b/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbFunctionStore.cs @@ -277,6 +277,43 @@ public async Task> Restar return result; } + public async Task> RestartExecutionsWithoutMessages( + IReadOnlyList storedIds, + ReplicaId owner) + { + if (storedIds.Count == 0) + return new Dictionary(); + + // Execute restart (effects are returned inline) + var restartedFlows = await RestartFlowsAsync(storedIds, owner); + + // Build result dictionary - only for successfully restarted flows + var result = new Dictionary(); + foreach (var (flow, effectsBytes, session) in restartedFlows) + { + var effects = new List(); + if (effectsBytes != null) + { + var effectsBytesArray = BinaryPacker.Split(effectsBytes); + foreach (var effectBytes in effectsBytesArray) + { + if (effectBytes != null) + { + var storedEffect = StoredEffect.Deserialize(effectBytes); + effects.Add(storedEffect); + session.Effects[storedEffect.EffectId] = storedEffect; + } + } + } + + result[flow.StoredId] = new StoredFlowWithEffects( + flow, effects, session + ); + } + + return result; + } + private async Task> RestartFlowsAsync( IReadOnlyList storedIds, ReplicaId owner) diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/StoreCrudTests.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/StoreCrudTests.cs index 634fd6752..6f56bffa5 100644 --- a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/StoreCrudTests.cs +++ b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/StoreCrudTests.cs @@ -73,4 +73,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()); } \ No newline at end of file diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlFunctionStore.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlFunctionStore.cs index 4f40f2f04..9e3e853a8 100644 --- a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlFunctionStore.cs +++ b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlFunctionStore.cs @@ -267,6 +267,41 @@ public async Task> Restar return result; } + public async Task> RestartExecutionsWithoutMessages( + IReadOnlyList storedIds, + ReplicaId owner) + { + if (storedIds.Count == 0) + return new Dictionary(); + + // Execute 2 queries in parallel + var restartTask = RestartFlowsAsync(storedIds, owner); + var effectsTask = FetchEffectsAsync(storedIds, owner); + + await Task.WhenAll(restartTask, effectsTask); + + var restartedFlows = await restartTask; + var effectsMap = await effectsTask; + + // Build result dictionary - only for successfully restarted flows + var result = new Dictionary(); + foreach (var flow in restartedFlows) + { + var effects = effectsMap.TryGetValue(flow.StoredId, out var session) + ? session.Effects.Values.ToList() + : new List(); + var storageSession = effectsMap.TryGetValue(flow.StoredId, out var s) + ? s + : new SnapshotStorageSession(owner); + + result[flow.StoredId] = new StoredFlowWithEffects( + flow, effects, storageSession + ); + } + + return result; + } + private async Task> RestartFlowsAsync( IReadOnlyList storedIds, ReplicaId owner) diff --git a/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/StoreCrudTests.cs b/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/StoreCrudTests.cs index f56ddee12..46e3e96f9 100644 --- a/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/StoreCrudTests.cs +++ b/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/StoreCrudTests.cs @@ -73,4 +73,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()); } \ No newline at end of file diff --git a/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerFunctionStore.cs b/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerFunctionStore.cs index 7b8de094c..9c5ac585f 100644 --- a/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerFunctionStore.cs +++ b/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerFunctionStore.cs @@ -343,6 +343,43 @@ public async Task> Restar return result; } + public async Task> RestartExecutionsWithoutMessages( + IReadOnlyList storedIds, + ReplicaId owner) + { + if (storedIds.Count == 0) + return new Dictionary(); + + // Execute restart (effects are returned inline) + var restartedFlows = await RestartFlowsAsync(storedIds, owner); + + // Build result dictionary - only for successfully restarted flows + var result = new Dictionary(); + foreach (var (flow, effectsBytes, session) in restartedFlows) + { + var effects = new List(); + if (effectsBytes != null) + { + var effectsBytesArray = BinaryPacker.Split(effectsBytes); + foreach (var effectBytes in effectsBytesArray) + { + if (effectBytes != null) + { + var storedEffect = StoredEffect.Deserialize(effectBytes); + effects.Add(storedEffect); + session.Effects[storedEffect.EffectId] = storedEffect; + } + } + } + + result[flow.StoredId] = new StoredFlowWithEffects( + flow, effects, session + ); + } + + return result; + } + private async Task> RestartFlowsAsync( IReadOnlyList storedIds, ReplicaId owner)