Skip to content

[Mocha] Stabilize saga integration tests against timing flakes #9947

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
glen-84 merged 8 commits into from
Oct 9, 2026
Merged

[Mocha] Stabilize saga integration tests against timing flakes #9947

Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
b9841b3
Stabilize Mocha saga integration tests against timing flakes
glen-84 Jun 18, 2026
91566f2
Drop unnecessary MessagingRuntime cast in saga tests
glen-84 Jun 18, 2026
adef396
Merge branch 'main' into gai/stabilize-saga-integration-tests
glen-84 Oct 9, 2026
3a57314
Merge branch 'main' into gai/stabilize-saga-integration-tests
glen-84 Oct 9, 2026
c8ca138
Wait for the saga to persist before the first step-through event
glen-84 Oct 9, 2026
1d2d385
Fail fast on unpersisted saga state and always release the reply gate
glen-84 Oct 9, 2026
ef5dc19
Wait for the saga to consume the first event before sending the second
glen-84 Oct 9, 2026
90aaa9b
Assert that the saga consumed the first event before sending the second
glen-84 Oct 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
144 changes: 131 additions & 13 deletions src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<TestMessageHandler>();
b.AddSaga<StepThroughSaga>();

// 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();
Expand All @@ -48,17 +65,41 @@ public async Task Saga_Should_ExecuteThroughSteps()
var testMessage = Assert.Single(recorder.Messages);
var sagaId = Assert.IsType<TestMessage>(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<IMessagingRuntime>();
var sagaName = runtime.Naming.GetSagaName(typeof(StepThroughSaga));
var persistDeadline = DateTime.UtcNow + s_timeout;
while (storage.Load<StepThroughState>(sagaName, sagaId)?.State != "Started"
&& DateTime.UtcNow < persistDeadline)
{
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<StepThroughState>(sagaName, sagaId)?.State);

// 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 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
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<StepThroughState>(sagaName, sagaId)?.State);
Comment thread
glen-84 marked this conversation as resolved.

// 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);
Expand Down Expand Up @@ -89,11 +130,28 @@ public async Task Saga_Should_Timeout()
var recorded = Assert.Single(recorder.Messages);
var sagaId = Assert.IsType<TriggerEvent>(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 = provider.GetRequiredService<IMessagingRuntime>();
var sagaName = runtime.Naming.GetSagaName(typeof(TimeoutSaga));
var persistDeadline = DateTime.UtcNow + s_timeout;
while (storage.Load<TimeoutState>(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<TimeoutState>(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<TimeoutState>(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);
Expand Down Expand Up @@ -124,11 +182,28 @@ public async Task Saga_Should_TimeoutWithCustomResponse()
var recorded = Assert.Single(recorder.Messages);
var sagaId = Assert.IsType<TriggerEvent>(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 = provider.GetRequiredService<IMessagingRuntime>();
var sagaName = runtime.Naming.GetSagaName(typeof(TimeoutWithResponseSaga));
var persistDeadline = DateTime.UtcNow + s_timeout;
while (storage.Load<TimeoutState>(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<TimeoutState>(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<TimeoutState>(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);
Expand All @@ -138,22 +213,51 @@ 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<TriggerRequestHandler>();
b.Services.AddSingleton(recorder);
b.Services.AddSingleton(replyGate);
b.AddRequestHandler<GatedTriggerRequestHandler>();
b.AddSaga<RequestResponseSaga>();
});

using var scope = provider.CreateScope();
var bus = scope.ServiceProvider.GetRequiredService<IMessageBus>();
var storage = provider.GetRequiredService<InMemorySagaStateStorage>();

// 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);
try
{
// 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");
}
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;
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);
Expand Down Expand Up @@ -620,6 +724,20 @@ public ValueTask<TriggerResponse> 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<TriggerRequest, TriggerResponse>
{
public async ValueTask<TriggerResponse> HandleAsync(TriggerRequest request, CancellationToken cancellationToken)
{
recorder.Record(request);
await replyGate.Task.WaitAsync(cancellationToken);
return new TriggerResponse();
}
}

// =========================================================================
// Test Infrastructure
// =========================================================================
Expand Down
Loading