Flush pending effects before persisting new function status

stidsborg · stidsborg · commit d4a2ad76efdd · 2025-04-12T09:05:55.000+02:00
diff --git a/Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/RFunctionTests/EffectTests.cs b/Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/RFunctionTests/EffectTests.cs
@@ -73,4 +73,8 @@ public override Task DelayedFlushIsReflectedInUnderlyingStoreForSet()
     [TestMethod]
     public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush()
         => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow()
+        => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow(FunctionStoreFactory.Create());
 }
diff --git a/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/EffectTests.cs b/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/EffectTests.cs
@@ -665,4 +665,46 @@ public async Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush(Task<
         storedEffects.Single(se => se.EffectId.Id == "2").Result!.ToStringFromUtf8Bytes().DeserializeFromJsonTo<string>().ShouldBe("hello universe");
     }
 
+    public abstract Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow();
+    public async Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow(Task<IFunctionStore> storeTask)
+    {  
+        var store = await storeTask;
+        var (type, instance) = TestFlowId.Create();
+        var storedId = TestStoredId.Create();
+        var someEffectIdValue = Guid.NewGuid();
+
+        var unhandledExceptionCatcher = new UnhandledExceptionCatcher();
+        var registry = new FunctionsRegistry(store, settings: new Settings(unhandledExceptionHandler: unhandledExceptionCatcher.Catch));
+
+        var writtenEffectFlag = new SyncedFlag();
+        var continueFlag = new SyncedFlag();
+        
+        var flow = registry.RegisterParamless(
+            type,
+            async Task (workflow) =>
+            {
+                await workflow.Effect.Capture("SomeEffectId", () => someEffectIdValue, ResiliencyLevel.AtLeastOnceDelayFlush);
+                writtenEffectFlag.Raise();
+                await continueFlag.WaitForRaised();
+            }
+        );
+
+        await flow.Schedule(instance);
+        
+        await writtenEffectFlag.WaitForRaised();
+        var cp = await flow.ControlPanel(instance).ShouldNotBeNullAsync();
+        var effectIds = await cp.Effects.AllIds;
+        effectIds.Any().ShouldBeFalse();
+        
+        continueFlag.Raise();
+
+        await cp.WaitForCompletion();
+
+        await cp.Refresh();
+        effectIds = (await cp.Effects.AllIds).ToList();
+        effectIds.Count().ShouldBe(1);
+        effectIds.Single().Id.ShouldBe("SomeEffectId");
+        (await cp.Effects.GetValue<Guid>("SomeEffectId")).ShouldBe(someEffectIdValue);
+    }
+    
 }
diff --git a/Core/Cleipnir.ResilientFunctions/CoreRuntime/Invocation/Invoker.cs b/Core/Cleipnir.ResilientFunctions/CoreRuntime/Invocation/Invoker.cs
@@ -55,7 +55,7 @@ public async Task<TReturn> Invoke(FlowInstance instance, TParam param, InitialSt
         }
         finally{ disposables.Dispose(); }
 
-        await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent: null);
+        await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent: null, workflow);
         return result.SucceedWithValue!;
     }
 
@@ -90,7 +90,7 @@ public async Task<InnerScheduled<TReturn>> ScheduleInvoke(FlowInstance flowInsta
                 catch (Exception exception) { var fwe = FatalWorkflowException.CreateNonGeneric(flowId, exception); await PersistFailure(storedId, flowId, fwe, param, parent?.StoredId); throw fwe; }
                 finally{ disposables.Dispose(); }
 
-                await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent?.StoredId, allowPostponedOrSuspended: true);
+                await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent?.StoredId, workflow, allowPostponedOrSuspended: true);
             }
             catch (Exception exception) { _unhandledExceptionHandler.Invoke(_flowType, exception); }
         });
@@ -141,7 +141,7 @@ public async Task<TReturn> Restart(StoredInstance instanceId, int expectedEpoch)
         catch (Exception exception) { var fwe = FatalWorkflowException.CreateNonGeneric(flowId, exception); await PersistFailure(storedId, flowId, fwe, param, parent, epoch); throw fwe; }
         finally{ disposables.Dispose(); }
         
-        await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent, epoch);
+        await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent, workflow, epoch);
         return result.SucceedWithValue!;
     }
 
@@ -167,7 +167,7 @@ public async Task ScheduleRestart(StoredInstance instance, int expectedEpoch)
                 catch (Exception exception) { var fwe = FatalWorkflowException.CreateNonGeneric(flowId, exception); await PersistFailure(storedId, flowId, fwe, param, parent, epoch); throw fwe; }
                 finally{ disposables.Dispose(); }
                 
-                await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent, epoch, allowPostponedOrSuspended: true);
+                await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent, workflow, epoch, allowPostponedOrSuspended: true);
             }
             catch (Exception exception) { _unhandledExceptionHandler.Invoke(_flowType, exception); }
         });
@@ -198,7 +198,7 @@ internal async Task ScheduleRestart(StoredInstance instance, RestartedFunction r
                 catch (Exception exception) { var fwe = FatalWorkflowException.CreateNonGeneric(flowId, exception); await PersistFailure(storedId, flowId, fwe, param, parent, epoch); throw fwe; }
                 finally { disposables.Dispose(); }
                 
-                await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent, epoch, allowPostponedOrSuspended: true);
+                await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent, workflow, epoch, allowPostponedOrSuspended: true);
             }
             catch (Exception exception) { _unhandledExceptionHandler.Invoke(_flowType, exception); }
             finally{ onCompletion(); }
@@ -342,8 +342,9 @@ private async Task PersistFailure(StoredId storedId, FlowId flowId, FatalWorkflo
         await _invocationHelper.PersistFailure(storedId, flowId, exception, param, parent, expectedEpoch);
     }
 
-    private async Task PersistResultAndEnsureSuccess(StoredId storedId, FlowId flowId, Result<TReturn> result, TParam param, StoredId? parent, int expectedEpoch = 0, bool allowPostponedOrSuspended = false)
+    private async Task PersistResultAndEnsureSuccess(StoredId storedId, FlowId flowId, Result<TReturn> result, TParam param, StoredId? parent, Workflow workflow, int expectedEpoch = 0, bool allowPostponedOrSuspended = false)
     {
+        await workflow.Effect.Flush();
         await _invocationHelper.PublishCompletionMessageToParent(parent, childId: flowId, result);
         
         var outcome = await _invocationHelper.PersistResult(storedId, flowId, result, param, parent, expectedEpoch);
diff --git a/Core/Cleipnir.ResilientFunctions/Domain/Effect.cs b/Core/Cleipnir.ResilientFunctions/Domain/Effect.cs
@@ -98,4 +98,6 @@ public Task<T[]> WhenAll<T>(params Task<T>[] tasks)
 
     internal EffectId CreateEffectId(string id, EffectType? type = null) 
         => id.ToEffectId(type, context: EffectContext.CurrentContext.Parent?.Serialize());
+
+    public Task Flush() => effectResults.Flush();
 }
diff --git a/Core/Cleipnir.ResilientFunctions/Domain/EffectResults.cs b/Core/Cleipnir.ResilientFunctions/Domain/EffectResults.cs
@@ -361,7 +361,7 @@ private async Task FlushOrAddToPending(EffectId effectId, StoredEffectId storedE
     }
     
     private readonly SemaphoreSlim _flushSync = new(initialCount: 1, maxCount: 1);
-    private async Task Flush()
+    public async Task Flush()
     {
         await _flushSync.WaitAsync();
 
diff --git a/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/RFunctionTests/EffectTests.cs b/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/RFunctionTests/EffectTests.cs
@@ -72,4 +72,8 @@ public override Task DelayedFlushIsReflectedInUnderlyingStoreForSet()
     [TestMethod]
     public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush()
         => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush(FunctionStoreFactory.Create());
+    
+    [TestMethod]
+    public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow()
+        => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow(FunctionStoreFactory.Create());
 }
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/RFunctionTests/EffectTests.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/RFunctionTests/EffectTests.cs
@@ -73,4 +73,8 @@ public override Task DelayedFlushIsReflectedInUnderlyingStoreForSet()
     [TestMethod]
     public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush()
         => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush(FunctionStoreFactory.Create());
+    
+    [TestMethod]
+    public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow()
+        => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow(FunctionStoreFactory.Create());
 }
diff --git a/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/EffectTests.cs b/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/EffectTests.cs
@@ -73,4 +73,8 @@ public override Task DelayedFlushIsReflectedInUnderlyingStoreForSet()
     [TestMethod]
     public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush()
         => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush(FunctionStoreFactory.Create());
+    
+    [TestMethod]
+    public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow()
+        => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow(FunctionStoreFactory.Create());
 }

Original file line number	Diff line number	Diff line change
`@@ -73,4 +73,8 @@ public override Task DelayedFlushIsReflectedInUnderlyingStoreForSet()`
`73`	`73`	`[TestMethod]`
`74`	`74`	`public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush()`
`75`	`75`	`=> CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush(FunctionStoreFactory.Create());`
	`76`	`+`
	`77`	`+ [TestMethod]`
	`78`	`+ public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow()`
	`79`	`+ => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow(FunctionStoreFactory.Create());`
`76`	`80`	`}`
Original file line number	Diff line number	Diff line change
`@@ -55,7 +55,7 @@ public async Task<TReturn> Invoke(FlowInstance instance, TParam param, InitialSt`
`55`	`55`	`}`
`56`	`56`	`finally{ disposables.Dispose(); }`
`57`	`57`
`58`		`- await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent: null);`
	`58`	`+ await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent: null, workflow);`
`59`	`59`	`return result.SucceedWithValue!;`
`60`	`60`	`}`
`61`	`61`
`@@ -90,7 +90,7 @@ public async Task<InnerScheduled<TReturn>> ScheduleInvoke(FlowInstance flowInsta`
`90`	`90`	`catch (Exception exception) { var fwe = FatalWorkflowException.CreateNonGeneric(flowId, exception); await PersistFailure(storedId, flowId, fwe, param, parent?.StoredId); throw fwe; }`
`91`	`91`	`finally{ disposables.Dispose(); }`
`92`	`92`
`93`		`- await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent?.StoredId, allowPostponedOrSuspended: true);`
	`93`	`+ await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent?.StoredId, workflow, allowPostponedOrSuspended: true);`
`94`	`94`	`}`
`95`	`95`	`catch (Exception exception) { _unhandledExceptionHandler.Invoke(_flowType, exception); }`
`96`	`96`	`});`
`@@ -141,7 +141,7 @@ public async Task<TReturn> Restart(StoredInstance instanceId, int expectedEpoch)`
`141`	`141`	`catch (Exception exception) { var fwe = FatalWorkflowException.CreateNonGeneric(flowId, exception); await PersistFailure(storedId, flowId, fwe, param, parent, epoch); throw fwe; }`
`142`	`142`	`finally{ disposables.Dispose(); }`
`143`	`143`
`144`		`- await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent, epoch);`
	`144`	`+ await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent, workflow, epoch);`
`145`	`145`	`return result.SucceedWithValue!;`
`146`	`146`	`}`
`147`	`147`
`@@ -167,7 +167,7 @@ public async Task ScheduleRestart(StoredInstance instance, int expectedEpoch)`
`167`	`167`	`catch (Exception exception) { var fwe = FatalWorkflowException.CreateNonGeneric(flowId, exception); await PersistFailure(storedId, flowId, fwe, param, parent, epoch); throw fwe; }`
`168`	`168`	`finally{ disposables.Dispose(); }`
`169`	`169`
`170`		`- await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent, epoch, allowPostponedOrSuspended: true);`
	`170`	`+ await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent, workflow, epoch, allowPostponedOrSuspended: true);`
`171`	`171`	`}`
`172`	`172`	`catch (Exception exception) { _unhandledExceptionHandler.Invoke(_flowType, exception); }`
`173`	`173`	`});`
`@@ -198,7 +198,7 @@ internal async Task ScheduleRestart(StoredInstance instance, RestartedFunction r`
`198`	`198`	`catch (Exception exception) { var fwe = FatalWorkflowException.CreateNonGeneric(flowId, exception); await PersistFailure(storedId, flowId, fwe, param, parent, epoch); throw fwe; }`
`199`	`199`	`finally { disposables.Dispose(); }`
`200`	`200`
`201`		`- await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent, epoch, allowPostponedOrSuspended: true);`
	`201`	`+ await PersistResultAndEnsureSuccess(storedId, flowId, result, param, parent, workflow, epoch, allowPostponedOrSuspended: true);`
`202`	`202`	`}`
`203`	`203`	`catch (Exception exception) { _unhandledExceptionHandler.Invoke(_flowType, exception); }`
`204`	`204`	`finally{ onCompletion(); }`
`@@ -342,8 +342,9 @@ private async Task PersistFailure(StoredId storedId, FlowId flowId, FatalWorkflo`
`342`	`342`	`await _invocationHelper.PersistFailure(storedId, flowId, exception, param, parent, expectedEpoch);`
`343`	`343`	`}`
`344`	`344`
`345`		`- private async Task PersistResultAndEnsureSuccess(StoredId storedId, FlowId flowId, Result<TReturn> result, TParam param, StoredId? parent, int expectedEpoch = 0, bool allowPostponedOrSuspended = false)`
	`345`	`+ private async Task PersistResultAndEnsureSuccess(StoredId storedId, FlowId flowId, Result<TReturn> result, TParam param, StoredId? parent, Workflow workflow, int expectedEpoch = 0, bool allowPostponedOrSuspended = false)`
`346`	`346`	`{`
	`347`	`+ await workflow.Effect.Flush();`
`347`	`348`	`await _invocationHelper.PublishCompletionMessageToParent(parent, childId: flowId, result);`
`348`	`349`
`349`	`350`	`var outcome = await _invocationHelper.PersistResult(storedId, flowId, result, param, parent, expectedEpoch);`
Original file line number	Diff line number	Diff line change
`@@ -98,4 +98,6 @@ public Task<T[]> WhenAll<T>(params Task<T>[] tasks)`
`98`	`98`
`99`	`99`	`internal EffectId CreateEffectId(string id, EffectType? type = null)`
`100`	`100`	`=> id.ToEffectId(type, context: EffectContext.CurrentContext.Parent?.Serialize());`
	`101`	`+`
	`102`	`+ public Task Flush() => effectResults.Flush();`
`101`	`103`	`}`
Original file line number	Diff line number	Diff line change
`@@ -361,7 +361,7 @@ private async Task FlushOrAddToPending(EffectId effectId, StoredEffectId storedE`
`361`	`361`	`}`
`362`	`362`
`363`	`363`	`private readonly SemaphoreSlim _flushSync = new(initialCount: 1, maxCount: 1);`
`364`		`- private async Task Flush()`
	`364`	`+ public async Task Flush()`
`365`	`365`	`{`
`366`	`366`	`await _flushSync.WaitAsync();`
`367`	`367`
Original file line number	Diff line number	Diff line change
`@@ -72,4 +72,8 @@ public override Task DelayedFlushIsReflectedInUnderlyingStoreForSet()`
`72`	`72`	`[TestMethod]`
`73`	`73`	`public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush()`
`74`	`74`	`=> CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush(FunctionStoreFactory.Create());`
	`75`	`+`
	`76`	`+ [TestMethod]`
	`77`	`+ public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow()`
	`78`	`+ => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow(FunctionStoreFactory.Create());`
`75`	`79`	`}`