From b9841b32c510a93417435eb4b99aab5c0119a5eb Mon Sep 17 00:00:00 2001 From: Glen Date: Thu, 18 Jun 2026 12:48:46 +0200 Subject: [PATCH 1/6] Stabilize Mocha saga integration tests against timing flakes --- .../Mocha.Sagas.Tests/IntegrationTests.cs | 112 ++++++++++++++++-- 1 file changed, 99 insertions(+), 13 deletions(-) diff --git a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs index cdc0396bd15..d65aff7e602 100644 --- a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs +++ b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs @@ -50,14 +50,29 @@ public async Task Saga_Should_ExecuteThroughSteps() // send first TriggerEvent to transition Started -> Triggered await bus.PublishAsync(new TriggerEvent(sagaId), CancellationToken.None); - // allow time for the first transition to complete before sending the second event - await Task.Delay(500, TestContext.Current.CancellationToken); + // wait until the first transition is persisted (state == "Triggered") before sending the + // second event, so the two events are applied to the saga in order + var runtime = (MessagingRuntime)provider.GetRequiredService(); + var sagaName = runtime.Naming.GetSagaName(typeof(StepThroughSaga)); + var transitionDeadline = DateTime.UtcNow + s_timeout; + while (storage.Load(sagaName, sagaId)?.State != "Triggered" + && DateTime.UtcNow < transitionDeadline) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + } + + // fail fast if the first transition never happened, rather than sending the second event blindly + Assert.Equal("Triggered", storage.Load(sagaName, sagaId)?.State); // send second TriggerEvent to transition Triggered -> Completed (final) await bus.PublishAsync(new TriggerEvent(sagaId), CancellationToken.None); - // allow time for final state processing - await Task.Delay(1000, TestContext.Current.CancellationToken); + // wait for the saga to reach its final state and be deleted from the store + var deadline = DateTime.UtcNow + s_timeout; + while (storage.Count != 0 && DateTime.UtcNow < deadline) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + } // assert - saga should be deleted from store after reaching final state Assert.Equal(0, storage.Count); @@ -88,11 +103,28 @@ public async Task Saga_Should_Timeout() var recorded = Assert.Single(recorder.Messages); var sagaId = Assert.IsType(recorded).CorrelationId!.Value; + // wait until the saga instance is persisted before sending the timeout, so the + // SagaTimedOutEvent is applied to the stored instance + var runtime = (MessagingRuntime)provider.GetRequiredService(); + var sagaName = runtime.Naming.GetSagaName(typeof(TimeoutSaga)); + var persistDeadline = DateTime.UtcNow + s_timeout; + while (storage.Load(sagaName, sagaId) is null && DateTime.UtcNow < persistDeadline) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + } + + // fail fast if the saga was never persisted, so the timeout below targets the stored instance + Assert.NotNull(storage.Load(sagaName, sagaId)); + // simulate timeout by sending SagaTimedOutEvent with the saga ID await bus.SendAsync(new SagaTimedOutEvent(sagaId), CancellationToken.None); - // allow time for final state processing - await Task.Delay(500, TestContext.Current.CancellationToken); + // wait for the saga to reach its final state and be deleted from the store + var deadline = DateTime.UtcNow + s_timeout; + while (storage.Load(sagaName, sagaId) is not null && DateTime.UtcNow < deadline) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + } // assert - saga should be deleted from store after reaching final state Assert.Equal(0, storage.Count); @@ -123,11 +155,28 @@ public async Task Saga_Should_TimeoutWithCustomResponse() var recorded = Assert.Single(recorder.Messages); var sagaId = Assert.IsType(recorded).CorrelationId!.Value; + // wait until the saga instance is persisted before sending the timeout, so the + // SagaTimedOutEvent is applied to the stored instance + var runtime = (MessagingRuntime)provider.GetRequiredService(); + var sagaName = runtime.Naming.GetSagaName(typeof(TimeoutWithResponseSaga)); + var persistDeadline = DateTime.UtcNow + s_timeout; + while (storage.Load(sagaName, sagaId) is null && DateTime.UtcNow < persistDeadline) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + } + + // fail fast if the saga was never persisted, so the timeout below targets the stored instance + Assert.NotNull(storage.Load(sagaName, sagaId)); + // simulate timeout by sending SagaTimedOutEvent with the saga ID await bus.SendAsync(new SagaTimedOutEvent(sagaId), CancellationToken.None); - // allow time for final state processing - await Task.Delay(500, TestContext.Current.CancellationToken); + // wait for the saga to reach its final state and be deleted from the store + var deadline = DateTime.UtcNow + s_timeout; + while (storage.Load(sagaName, sagaId) is not null && DateTime.UtcNow < deadline) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + } // assert - saga should be deleted from store after reaching final state Assert.Equal(0, storage.Count); @@ -137,9 +186,13 @@ public async Task Saga_Should_TimeoutWithCustomResponse() public async Task Saga_Should_SupportRequestResponse() { // arrange + var recorder = new MessageRecorder(); + var replyGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); await using var provider = await CreateBusAsync(b => { - b.AddRequestHandler(); + b.Services.AddSingleton(recorder); + b.Services.AddSingleton(replyGate); + b.AddRequestHandler(); b.AddSaga(); }); @@ -147,12 +200,31 @@ public async Task Saga_Should_SupportRequestResponse() var bus = scope.ServiceProvider.GetRequiredService(); var storage = provider.GetRequiredService(); - // act - publish StartRequestEvent to start the saga + // act - publish StartRequestEvent to start the saga, which sends TriggerRequest via .Send await bus.PublishAsync(new StartRequestEvent(), CancellationToken.None); - // allow time for: saga starts -> sends TriggerRequest -> handler responds - // -> saga receives reply via OnAnyReply -> transitions to Completed (final) - await Task.Delay(2000, TestContext.Current.CancellationToken); + // the handler parks on the reply gate, holding the saga in AwaitingResponse so the test can + // observe its persisted state before releasing the reply to finalize the saga + Assert.True(await recorder.WaitAsync(s_timeout), "request handler never executed"); + + // wait until the held saga's state is persisted, and assert it actually appeared + var appearDeadline = DateTime.UtcNow + s_timeout; + while (storage.Count == 0 && DateTime.UtcNow < appearDeadline) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + } + + Assert.True(storage.Count > 0, "saga state was not persisted while the reply was held"); + + // release the reply so it routes back via OnAnyReply and finalizes the saga + replyGate.SetResult(); + + // wait for the reply to finalize the saga and delete its state + var deadline = DateTime.UtcNow + s_timeout; + while (storage.Count != 0 && DateTime.UtcNow < deadline) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + } // assert - saga should be deleted from store after reaching final state Assert.Equal(0, storage.Count); @@ -547,6 +619,20 @@ public ValueTask HandleAsync(TriggerRequest request, Cancellati } } + // Signals that the request was received, then parks until the test releases the reply gate. + // Holding the reply keeps the saga in its AwaitingResponse state so the test can deterministically + // observe the persisted state before triggering finalization. + private sealed class GatedTriggerRequestHandler(MessageRecorder recorder, TaskCompletionSource replyGate) + : IEventRequestHandler + { + public async ValueTask HandleAsync(TriggerRequest request, CancellationToken cancellationToken) + { + recorder.Record(request); + await replyGate.Task.WaitAsync(cancellationToken); + return new TriggerResponse(); + } + } + // ========================================================================= // Test Infrastructure // ========================================================================= From 91566f29686222ede3b6568fe093260dd5d4dd83 Mon Sep 17 00:00:00 2001 From: Glen Date: Thu, 18 Jun 2026 13:10:21 +0200 Subject: [PATCH 2/6] Drop unnecessary MessagingRuntime cast in saga tests --- src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs index d65aff7e602..6cececd0ed6 100644 --- a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs +++ b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs @@ -52,7 +52,7 @@ public async Task Saga_Should_ExecuteThroughSteps() // wait until the first transition is persisted (state == "Triggered") before sending the // second event, so the two events are applied to the saga in order - var runtime = (MessagingRuntime)provider.GetRequiredService(); + var runtime = provider.GetRequiredService(); var sagaName = runtime.Naming.GetSagaName(typeof(StepThroughSaga)); var transitionDeadline = DateTime.UtcNow + s_timeout; while (storage.Load(sagaName, sagaId)?.State != "Triggered" @@ -105,7 +105,7 @@ public async Task Saga_Should_Timeout() // wait until the saga instance is persisted before sending the timeout, so the // SagaTimedOutEvent is applied to the stored instance - var runtime = (MessagingRuntime)provider.GetRequiredService(); + var runtime = provider.GetRequiredService(); var sagaName = runtime.Naming.GetSagaName(typeof(TimeoutSaga)); var persistDeadline = DateTime.UtcNow + s_timeout; while (storage.Load(sagaName, sagaId) is null && DateTime.UtcNow < persistDeadline) @@ -157,7 +157,7 @@ public async Task Saga_Should_TimeoutWithCustomResponse() // wait until the saga instance is persisted before sending the timeout, so the // SagaTimedOutEvent is applied to the stored instance - var runtime = (MessagingRuntime)provider.GetRequiredService(); + var runtime = provider.GetRequiredService(); var sagaName = runtime.Naming.GetSagaName(typeof(TimeoutWithResponseSaga)); var persistDeadline = DateTime.UtcNow + s_timeout; while (storage.Load(sagaName, sagaId) is null && DateTime.UtcNow < persistDeadline) From c8ca138949014e28a47c9d4341cdfddfbfec13b3 Mon Sep 17 00:00:00 2001 From: Glen Date: Fri, 9 Oct 2026 16:41:13 +0200 Subject: [PATCH 3/6] Wait for the saga to persist before the first step-through event --- .../test/Mocha.Sagas.Tests/IntegrationTests.cs | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs index 2a0d18558bb..f23352297a1 100644 --- a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs +++ b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs @@ -48,13 +48,22 @@ public async Task Saga_Should_ExecuteThroughSteps() var testMessage = Assert.Single(recorder.Messages); var sagaId = Assert.IsType(testMessage).Id; + // wait until the saga instance is persisted (state == "Started") before sending the first + // event, so the TriggerEvent is applied to the stored instance + var runtime = provider.GetRequiredService(); + var sagaName = runtime.Naming.GetSagaName(typeof(StepThroughSaga)); + var persistDeadline = DateTime.UtcNow + s_timeout; + while (storage.Load(sagaName, sagaId)?.State != "Started" + && DateTime.UtcNow < persistDeadline) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + } + // send first TriggerEvent to transition Started -> Triggered await bus.PublishAsync(new TriggerEvent(sagaId), CancellationToken.None); // wait until the first transition is persisted (state == "Triggered") before sending the // second event, so the two events are applied to the saga in order - var runtime = provider.GetRequiredService(); - var sagaName = runtime.Naming.GetSagaName(typeof(StepThroughSaga)); var transitionDeadline = DateTime.UtcNow + s_timeout; while (storage.Load(sagaName, sagaId)?.State != "Triggered" && DateTime.UtcNow < transitionDeadline) From 1d2d385dfe1591a0a6a0e3513d59c098a433dde3 Mon Sep 17 00:00:00 2001 From: Glen Date: Fri, 9 Oct 2026 17:06:15 +0200 Subject: [PATCH 4/6] Fail fast on unpersisted saga state and always release the reply gate --- .../Mocha.Sagas.Tests/IntegrationTests.cs | 33 ++++++++++++------- 1 file changed, 21 insertions(+), 12 deletions(-) diff --git a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs index f23352297a1..4f6174236fe 100644 --- a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs +++ b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs @@ -59,6 +59,9 @@ public async Task Saga_Should_ExecuteThroughSteps() await Task.Delay(50, TestContext.Current.CancellationToken); } + // fail fast if the saga was never persisted, so the first event targets the stored instance + Assert.Equal("Started", storage.Load(sagaName, sagaId)?.State); + // send first TriggerEvent to transition Started -> Triggered await bus.PublishAsync(new TriggerEvent(sagaId), CancellationToken.None); @@ -213,21 +216,27 @@ public async Task Saga_Should_SupportRequestResponse() // act - publish StartRequestEvent to start the saga, which sends TriggerRequest via .Send await bus.PublishAsync(new StartRequestEvent(), CancellationToken.None); - // the handler parks on the reply gate, holding the saga in AwaitingResponse so the test can - // observe its persisted state before releasing the reply to finalize the saga - Assert.True(await recorder.WaitAsync(s_timeout), "request handler never executed"); - - // wait until the held saga's state is persisted, and assert it actually appeared - var appearDeadline = DateTime.UtcNow + s_timeout; - while (storage.Count == 0 && DateTime.UtcNow < appearDeadline) + try { - await Task.Delay(50, TestContext.Current.CancellationToken); - } + // the handler parks on the reply gate, holding the saga in AwaitingResponse so the test + // can observe its persisted state before releasing the reply to finalize the saga + Assert.True(await recorder.WaitAsync(s_timeout), "request handler never executed"); - Assert.True(storage.Count > 0, "saga state was not persisted while the reply was held"); + // wait until the held saga's state is persisted, and assert it actually appeared + var appearDeadline = DateTime.UtcNow + s_timeout; + while (storage.Count == 0 && DateTime.UtcNow < appearDeadline) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + } - // release the reply so it routes back via OnAnyReply and finalizes the saga - replyGate.SetResult(); + Assert.True(storage.Count > 0, "saga state was not persisted while the reply was held"); + } + finally + { + // release the reply so it routes back via OnAnyReply and finalizes the saga, and so the + // handler never stays parked when an assertion above fails + replyGate.TrySetResult(); + } // wait for the reply to finalize the saga and delete its state var deadline = DateTime.UtcNow + s_timeout; From ef5dc19569d21457b0b02b031ba64ccf2e54714b Mon Sep 17 00:00:00 2001 From: Glen Date: Fri, 9 Oct 2026 17:35:27 +0200 Subject: [PATCH 5/6] Wait for the saga to consume the first event before sending the second --- .../Mocha.Sagas.Tests/IntegrationTests.cs | 28 +++++++++++++------ 1 file changed, 20 insertions(+), 8 deletions(-) diff --git a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs index 4f6174236fe..ca378af791d 100644 --- a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs +++ b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs @@ -28,11 +28,28 @@ public async Task Saga_Should_ExecuteThroughSteps() { // arrange var recorder = new MessageRecorder(); + var consumedTriggers = new MessageRecorder(); await using var provider = await CreateBusAsync(b => { b.Services.AddSingleton(recorder); b.AddEventHandler(); b.AddSaga(); + + // record each TriggerEvent after the saga consumer returns, by which point the saga + // transaction has committed + b.ConfigureMessageBus(h => + h.UseConsume( + new ConsumerMiddlewareConfiguration( + (_, next) => async ctx => + { + await next(ctx); + + if (ctx.GetMessage() is TriggerEvent triggerEvent) + { + consumedTriggers.Record(triggerEvent); + } + }, + "RecordConsumedTriggerEvent"))); }); using var scope = provider.CreateScope(); @@ -65,14 +82,9 @@ public async Task Saga_Should_ExecuteThroughSteps() // send first TriggerEvent to transition Started -> Triggered await bus.PublishAsync(new TriggerEvent(sagaId), CancellationToken.None); - // wait until the first transition is persisted (state == "Triggered") before sending the - // second event, so the two events are applied to the saga in order - var transitionDeadline = DateTime.UtcNow + s_timeout; - while (storage.Load(sagaName, sagaId)?.State != "Triggered" - && DateTime.UtcNow < transitionDeadline) - { - await Task.Delay(50, TestContext.Current.CancellationToken); - } + // wait until the saga has consumed the first event and committed its transition before + // sending the second, so the two events are applied to the saga in order + await consumedTriggers.WaitAsync(s_timeout); // fail fast if the first transition never happened, rather than sending the second event blindly Assert.Equal("Triggered", storage.Load(sagaName, sagaId)?.State); From 90aaa9b20eb3b2d86859fb91589ecc005bf07211 Mon Sep 17 00:00:00 2001 From: Glen Date: Fri, 9 Oct 2026 17:49:14 +0200 Subject: [PATCH 6/6] Assert that the saga consumed the first event before sending the second --- src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs index ca378af791d..7f93344f187 100644 --- a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs +++ b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs @@ -84,7 +84,9 @@ public async Task Saga_Should_ExecuteThroughSteps() // wait until the saga has consumed the first event and committed its transition before // sending the second, so the two events are applied to the saga in order - await consumedTriggers.WaitAsync(s_timeout); + Assert.True( + await consumedTriggers.WaitAsync(s_timeout), + "Saga did not finish consuming the first TriggerEvent within timeout"); // fail fast if the first transition never happened, rather than sending the second event blindly Assert.Equal("Triggered", storage.Load(sagaName, sagaId)?.State);