From b9841b32c510a93417435eb4b99aab5c0119a5eb Mon Sep 17 00:00:00 2001 From: Glen Date: Thu, 18 Jun 2026 12:48:46 +0200 Subject: [PATCH 1/2] 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/2] 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)