From 2a1ae5f250cb48d2302c21b9edf4262a385d32ce Mon Sep 17 00:00:00 2001 From: Alexander Bohle Date: Sun, 6 Sep 2026 18:26:13 +0200 Subject: [PATCH] Create the saga feature per attempt so a retried saga does not reuse a disposed store --- .../Sagas/Execution/ISagaExecutionContext.cs | 17 +-- .../src/Mocha/Sagas/SagaEventListener.cs | 7 +- .../SagaRetryScopeWarmupTests.cs | 137 ++++++++++++++++++ 3 files changed, 145 insertions(+), 16 deletions(-) create mode 100644 src/Mocha/test/Mocha.Sagas.Tests/SagaRetryScopeWarmupTests.cs diff --git a/src/Mocha/src/Mocha/Sagas/Execution/ISagaExecutionContext.cs b/src/Mocha/src/Mocha/Sagas/Execution/ISagaExecutionContext.cs index 0640df5bdda..958e58be08c 100644 --- a/src/Mocha/src/Mocha/Sagas/Execution/ISagaExecutionContext.cs +++ b/src/Mocha/src/Mocha/Sagas/Execution/ISagaExecutionContext.cs @@ -3,26 +3,15 @@ namespace Mocha.Sagas; /// -/// A pooled feature that provides access to the saga store during saga event processing. +/// Carries the saga store for the current consume attempt. It is created on the attempt's own +/// feature collection and is never shared with the receive context or pooled. /// -public class SagaFeature : IPooledFeature +public class SagaFeature { /// /// Gets or sets the saga store used for persisting saga state. /// public ISagaStore Store { get; set; } = null!; - - /// - public void Initialize(object state) - { - Store = null!; - } - - /// - public void Reset() - { - Store = null!; - } } internal static class ConsumeContextSagaExtensions diff --git a/src/Mocha/src/Mocha/Sagas/SagaEventListener.cs b/src/Mocha/src/Mocha/Sagas/SagaEventListener.cs index f72dfc7d8d8..de12f4b0996 100644 --- a/src/Mocha/src/Mocha/Sagas/SagaEventListener.cs +++ b/src/Mocha/src/Mocha/Sagas/SagaEventListener.cs @@ -86,8 +86,11 @@ public override ConsumerDescription Describe() /// protected override async ValueTask ConsumeAsync(IConsumeContext context) { - var sagaFeature = context.Features.GetOrSet(); - sagaFeature.Store ??= context.Services.GetRequiredService(); + // The store is scoped to this attempt, so the feature is created here rather than read + // through the attempt's fallback to the receive context, where a previous attempt's + // instance would still be visible. + var sagaFeature = new SagaFeature { Store = context.Services.GetRequiredService() }; + context.Features.Set(sagaFeature); var ct = context.CancellationToken; diff --git a/src/Mocha/test/Mocha.Sagas.Tests/SagaRetryScopeWarmupTests.cs b/src/Mocha/test/Mocha.Sagas.Tests/SagaRetryScopeWarmupTests.cs new file mode 100644 index 00000000000..62b30e40088 --- /dev/null +++ b/src/Mocha/test/Mocha.Sagas.Tests/SagaRetryScopeWarmupTests.cs @@ -0,0 +1,137 @@ +using System.Collections.Concurrent; +using Microsoft.Extensions.DependencyInjection; +using Mocha.Transport.InMemory; + +namespace Mocha.Sagas.Tests; + +/// +/// Same as , but a saga consume has already run on the bus, so +/// the receive context rented for the retried message is a pooled instance that served as a saga +/// attempt before and carries a restored . +/// +public class SagaRetryScopeWarmupTests +{ + private static readonly TimeSpan s_timeout = TimeSpan.FromSeconds(10); + + [Fact] + public async Task Saga_Should_UseFreshStore_When_SaveIsRetried_After_EarlierSagaConsume() + { + // arrange + var capture = new StoreCapture(); + var services = new ServiceCollection(); + services.AddSingleton(capture); + services.AddScoped(); + services.AddInMemorySagas(); + var builder = services.AddMessageBus() + .AddResilience(p => p.On().Retry(3, TimeSpan.FromMilliseconds(1), RetryBackoffType.Constant)) + .AddSaga(); + builder.AddInMemory(); + + await using var provider = services.BuildServiceProvider(); + var runtime = (MessagingRuntime)provider.GetRequiredService(); + await runtime.StartAsync(CancellationToken.None); + + using var scope = provider.CreateScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + var storage = provider.GetRequiredService(); + + // warm-up: one saga consume that succeeds + await bus.PublishAsync(new StartRetrySaga(), CancellationToken.None); + await WaitUntilAsync(() => storage.Count == 1); + + // act: a second saga instance whose first save fails once + capture.Arm(); + await bus.PublishAsync(new StartRetrySaga(), CancellationToken.None); + await capture.Saved.Task.WaitAsync(s_timeout, TestContext.Current.CancellationToken); + await WaitUntilAsync(() => storage.Count == 2); + + // assert - the failed save and the retried save must come from two different scoped stores + Assert.Equal(2, capture.ArmedStoreIds.Count); + Assert.Equal(2, capture.ArmedStoreIds.Distinct().Count()); + } + + private static async Task WaitUntilAsync(Func condition) + { + var deadline = DateTime.UtcNow + s_timeout; + while (!condition() && DateTime.UtcNow < deadline) + { + await Task.Delay(25, TestContext.Current.CancellationToken); + } + + Assert.True(condition(), "timed out waiting for the saga state to land"); + } + + public sealed class StoreCapture + { + private int _armed; + + public ConcurrentQueue ArmedStoreIds { get; } = new(); + + public TaskCompletionSource Saved { get; } = new(); + + public int Failures; + + public bool Armed => Volatile.Read(ref _armed) == 1; + + public void Arm() => Volatile.Write(ref _armed, 1); + } + + public sealed class ArmedThrowOnceSagaStore(InMemorySagaStateStorage storage, StoreCapture capture) : ISagaStore + { + private readonly InMemorySagaStore _inner = new(storage); + private readonly Guid _id = Guid.NewGuid(); + + public Task StartTransactionAsync(CancellationToken cancellationToken) + => _inner.StartTransactionAsync(cancellationToken); + + public async Task SaveAsync(Saga saga, T state, CancellationToken cancellationToken) + where T : SagaStateBase + { + if (capture.Armed) + { + capture.ArmedStoreIds.Enqueue(_id); + + if (Interlocked.Increment(ref capture.Failures) == 1) + { + throw new InvalidOperationException("transient"); + } + } + + await _inner.SaveAsync(saga, state, cancellationToken); + + if (capture.Armed) + { + capture.Saved.TrySetResult(); + } + } + + public Task DeleteAsync(Saga saga, Guid id, CancellationToken cancellationToken) + => _inner.DeleteAsync(saga, id, cancellationToken); + + public Task LoadAsync(Saga saga, Guid id, CancellationToken cancellationToken) + => _inner.LoadAsync(saga, id, cancellationToken); + } + + public sealed class RetrySagaState : SagaStateBase; + + public sealed class StartRetrySaga; + + public sealed class EndRetrySaga; + + public sealed class RetrySaga : Saga + { + protected override void Configure(ISagaDescriptor descriptor) + { + descriptor + .Initially() + .OnEvent() + .StateFactory(_ => new RetrySagaState()) + .TransitionTo("Started"); + + // Started is not final, so the state stays in storage and Count is observable. + descriptor.During("Started").OnEvent().TransitionTo("Ended"); + + descriptor.Finally("Ended"); + } + } +}