Cover the command and publish legs of saga fault routing

OnReplyFault was only exercised through a request carrying a response type. Add
the command variant, handled by SendConsumer rather than the request consumer,
which both faults and acknowledges through the same reply endpoint.

Characterise the publish leg as well, which is not covered. A published event
carries no reply address, so a failing subscriber is routed to the error
endpoint and the saga is never told, holding its state until it times out. The
saga header does survive onto the error endpoint, which is what any correlated
workaround, or a later fix, has to build on.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: alisan3

src/Mocha/test/Mocha.Sagas.Tests/SagaPublishFaultTests.cs

@@ -0,0 +1,196 @@
+using Microsoft.Extensions.DependencyInjection;
+using Mocha.Events;
+using Mocha.Features;
+using Mocha.Middlewares;
+using Mocha.Transport.InMemory;
+
+namespace Mocha.Sagas.Tests;
+
+/// <summary>
+/// Characterizes the publish leg of saga fault handling. A published event carries no reply address,
+/// so a failing subscriber is routed to the error endpoint rather than back to the saga.
+/// </summary>
+public class SagaPublishFaultTests
+{
+    private static readonly TimeSpan s_timeout = TimeSpan.FromSeconds(10);
+
+    private static readonly TaskCompletionSource<NotAcknowledgedEvent> s_faultObserved = new();
+
+    [Fact]
+    public async Task Saga_Should_NotReceiveFault_When_PublishedEventSubscriberFaults()
+    {
+        // arrange
+        var handlerRan = new TaskCompletionSource();
+        var services = new ServiceCollection();
+        services.AddInMemorySagas();
+        var builder = services.AddMessageBus();
+        builder.Services.AddSingleton(handlerRan);
+        builder.AddEventHandler<FaultingSubscriber>();
+        builder.AddSaga<PublishingSaga>();
+        builder.AddInMemory();
+
+        await using var provider = services.BuildServiceProvider();
+        var runtime = (MessagingRuntime)provider.GetRequiredService<IMessagingRuntime>();
+        await runtime.StartAsync(CancellationToken.None);
+
+        using var scope = provider.CreateScope();
+        var bus = scope.ServiceProvider.GetRequiredService<IMessageBus>();
+        var storage = provider.GetRequiredService<InMemorySagaStateStorage>();
+
+        // act
+        await bus.PublishAsync(new StartPublishEvent(), CancellationToken.None);
+
+        // assert - the subscriber ran and threw, proving the published event was delivered
+        await handlerRan.Task.WaitAsync(s_timeout, TestContext.Current.CancellationToken);
+
+        // assert - the saga is never told, so it stays in its waiting state holding its state
+        var completed = await Task.WhenAny(s_faultObserved.Task, Task.Delay(2000, TestContext.Current.CancellationToken));
+        Assert.NotSame(s_faultObserved.Task, completed);
+        Assert.Equal(1, storage.Count);
+    }
+
+    [Fact]
+    public async Task ErrorEndpoint_Should_ReceiveSagaId_When_PublishedEventSubscriberFaults()
+    {
+        // The fault carries the original envelope, so the saga header survives onto the error
+        // endpoint even though the saga itself is never notified.
+
+        // arrange
+        var handlerRan = new TaskCompletionSource();
+        var services = new ServiceCollection();
+        services.AddInMemorySagas();
+        var builder = services.AddMessageBus();
+        builder.Services.AddSingleton(handlerRan);
+        builder.AddEventHandler<ErrorQueueSubscriber>();
+        builder.AddSaga<ErrorQueuePublishingSaga>();
+        builder.AddInMemory(d => d.AddConvention(new TestErrorEndpointConvention()));
+
+        await using var provider = services.BuildServiceProvider();
+        var runtime = (MessagingRuntime)provider.GetRequiredService<IMessagingRuntime>();
+        await runtime.StartAsync(CancellationToken.None);
+
+        using var scope = provider.CreateScope();
+        var bus = scope.ServiceProvider.GetRequiredService<IMessageBus>();
+
+        // act
+        await bus.PublishAsync(new StartErrorQueueEvent(), CancellationToken.None);
+        await handlerRan.Task.WaitAsync(s_timeout, TestContext.Current.CancellationToken);
+
+        // assert
+        var envelope = await ReadFirstErrorEnvelopeAsync(runtime);
+        Assert.Equal(MessageKind.Fault, envelope.Headers!.Get(MessageHeaders.MessageKind));
+        Assert.True(Guid.TryParse(envelope.Headers!.Get(SagaContextData.SagaId), out _));
+    }
+
+    /// <summary>
+    /// Drains whichever error queue receives a message first, since every default endpoint gets one.
+    /// </summary>
+    private static async Task<MessageEnvelope> ReadFirstErrorEnvelopeAsync(MessagingRuntime runtime)
+    {
+        var transport = runtime.Transports.OfType<InMemoryMessagingTransport>().Single();
+        var topology = (InMemoryMessagingTopology)transport.Topology;
+        var errorQueues = topology.Queues.Where(q => q.Name.EndsWith("_error")).ToArray();
+
+        using var cts = new CancellationTokenSource(s_timeout);
+
+        var reads = errorQueues
+            .Select(async queue =>
+            {
+                await foreach (var item in queue.ConsumeAsync(cts.Token))
+                {
+                    var envelope = new MessageEnvelope(item.Envelope);
+                    item.Dispose();
+                    return envelope;
+                }
+
+                return null;
+            })
+            .ToArray();
+
+        var completed = await Task.WhenAny(reads);
+        await cts.CancelAsync();
+
+        return await completed ?? throw new InvalidOperationException("no error envelope was received");
+    }
+
+    private sealed class TestErrorEndpointConvention : IInMemoryReceiveEndpointConfigurationConvention
+    {
+        public void Configure(
+            IMessagingConfigurationContext context,
+            InMemoryMessagingTransport transport,
+            InMemoryReceiveEndpointConfiguration configuration)
+        {
+            if (configuration is { Kind: ReceiveEndpointKind.Default, QueueName: { } queueName })
+            {
+                var feature = configuration.Features.GetOrSet<ReceiveFaultEndpointFeature>();
+                feature.Address ??= new Uri($"{transport.Schema}:q/{queueName}_error");
+            }
+        }
+    }
+
+    public sealed class StartErrorQueueEvent;
+
+    public sealed record ErrorQueueEvent(Guid Id);
+
+    public sealed class ErrorQueueSubscriber(TaskCompletionSource handlerRan) : IEventHandler<ErrorQueueEvent>
+    {
+        public ValueTask HandleAsync(ErrorQueueEvent message, CancellationToken cancellationToken)
+        {
+            handlerRan.TrySetResult();
+            throw new InvalidOperationException("terminal failure");
+        }
+    }
+
+    public sealed class ErrorQueuePublishingSaga : Saga<PublishState>
+    {
+        protected override void Configure(ISagaDescriptor<PublishState> descriptor)
+        {
+            descriptor
+                .Initially()
+                .OnEvent<StartErrorQueueEvent>()
+                .StateFactory(_ => new PublishState())
+                .Publish((_, state) => new ErrorQueueEvent(state.Id))
+                .TransitionTo("Awaiting");
+
+            descriptor.During("Awaiting").OnReplyFault().TransitionTo("Failed");
+
+            descriptor.Finally("Failed");
+        }
+    }
+
+    public sealed class PublishState : SagaStateBase;
+
+    public sealed class StartPublishEvent;
+
+    public sealed record PublishedEvent(Guid Id);
+
+    public sealed class FaultingSubscriber(TaskCompletionSource handlerRan) : IEventHandler<PublishedEvent>
+    {
+        public ValueTask HandleAsync(PublishedEvent message, CancellationToken cancellationToken)
+        {
+            handlerRan.TrySetResult();
+            throw new InvalidOperationException("terminal failure");
+        }
+    }
+
+    public sealed class PublishingSaga : Saga<PublishState>
+    {
+        protected override void Configure(ISagaDescriptor<PublishState> descriptor)
+        {
+            descriptor
+                .Initially()
+                .OnEvent<StartPublishEvent>()
+                .StateFactory(_ => new PublishState())
+                .Publish((_, state) => new PublishedEvent(state.Id))
+                .TransitionTo("Awaiting");
+
+            descriptor
+                .During("Awaiting")
+                .OnReplyFault()
+                .Then((_, fault) => s_faultObserved.TrySetResult(fault))
+                .TransitionTo("Failed");
+
+            descriptor.Finally("Failed");
+        }
+    }
+}

src/Mocha/test/Mocha.Sagas.Tests/SagaSendCommandTests.cs

@@ -0,0 +1,161 @@
+using Microsoft.Extensions.DependencyInjection;
+using Mocha.Events;
+using Mocha.Transport.InMemory;
+
+namespace Mocha.Sagas.Tests;
+
+/// <summary>
+/// Tests the reply leg of a saga send whose command has no response type, so it is handled by
+/// <c>SendConsumer</c> rather than the request consumer.
+/// </summary>
+public class SagaSendCommandTests
+{
+    private static readonly TimeSpan s_timeout = TimeSpan.FromSeconds(10);
+
+    private static readonly TaskCompletionSource<NotAcknowledgedEvent> s_faultObserved = new();
+
+    private static readonly TaskCompletionSource<object> s_replyObserved = new();
+
+    private static async Task<ServiceProvider> CreateBusAsync(Action<IMessageBusHostBuilder> configure)
+    {
+        var services = new ServiceCollection();
+        services.AddInMemorySagas();
+        var builder = services.AddMessageBus();
+        configure(builder);
+        builder.AddInMemory();
+
+        var provider = services.BuildServiceProvider();
+        var runtime = (MessagingRuntime)provider.GetRequiredService<IMessagingRuntime>();
+        await runtime.StartAsync(CancellationToken.None);
+        return provider;
+    }
+
+    [Fact]
+    public async Task Saga_Should_ReceiveFault_When_CommandWithoutResponseFaults()
+    {
+        // arrange
+        var handlerRan = new TaskCompletionSource();
+        await using var provider = await CreateBusAsync(b =>
+        {
+            b.Services.AddSingleton(handlerRan);
+            b.AddRequestHandler<FaultingCommandHandler>();
+            b.AddSaga<FaultingCommandSaga>();
+        });
+
+        using var scope = provider.CreateScope();
+        var bus = scope.ServiceProvider.GetRequiredService<IMessageBus>();
+
+        // act
+        await bus.PublishAsync(new StartFaultingCommandEvent(), CancellationToken.None);
+
+        // assert - the handler ran and threw, proving the command was delivered
+        await handlerRan.Task.WaitAsync(s_timeout, TestContext.Current.CancellationToken);
+
+        // assert - the fault reply routed back to the saga
+        var fault = await s_faultObserved.Task.WaitAsync(s_timeout, TestContext.Current.CancellationToken);
+        Assert.Equal(ErrorCodes.Exception, fault.ErrorCode);
+    }
+
+    [Fact]
+    public async Task Saga_Should_ReceiveAcknowledgement_When_CommandWithoutResponseSucceeds()
+    {
+        // A command with no response type still acknowledges, so the saga's reply route sees an
+        // AcknowledgedEvent rather than nothing at all.
+
+        // arrange
+        var handlerRan = new TaskCompletionSource();
+        await using var provider = await CreateBusAsync(b =>
+        {
+            b.Services.AddSingleton(handlerRan);
+            b.AddRequestHandler<SucceedingCommandHandler>();
+            b.AddSaga<SucceedingCommandSaga>();
+        });
+
+        using var scope = provider.CreateScope();
+        var bus = scope.ServiceProvider.GetRequiredService<IMessageBus>();
+
+        // act
+        await bus.PublishAsync(new StartSucceedingCommandEvent(), CancellationToken.None);
+
+        // assert - the handler ran and returned
+        await handlerRan.Task.WaitAsync(s_timeout, TestContext.Current.CancellationToken);
+
+        // assert - the acknowledgement routed back to the saga
+        var reply = await s_replyObserved.Task.WaitAsync(s_timeout, TestContext.Current.CancellationToken);
+        Assert.IsType<AcknowledgedEvent>(reply);
+    }
+
+    public sealed class CommandState : SagaStateBase;
+
+    public sealed class StartFaultingCommandEvent;
+
+    public sealed class StartSucceedingCommandEvent;
+
+    public sealed record FaultingCommand;
+
+    public sealed record SucceedingCommand;
+
+    public sealed class FaultingCommandHandler(TaskCompletionSource handlerRan)
+        : IEventRequestHandler<FaultingCommand>
+    {
+        public ValueTask HandleAsync(FaultingCommand request, CancellationToken cancellationToken)
+        {
+            handlerRan.TrySetResult();
+            throw new InvalidOperationException("terminal failure");
+        }
+    }
+
+    public sealed class SucceedingCommandHandler(TaskCompletionSource handlerRan)
+        : IEventRequestHandler<SucceedingCommand>
+    {
+        public ValueTask HandleAsync(SucceedingCommand request, CancellationToken cancellationToken)
+        {
+            handlerRan.TrySetResult();
+            return default;
+        }
+    }
+
+    public sealed class FaultingCommandSaga : Saga<CommandState>
+    {
+        protected override void Configure(ISagaDescriptor<CommandState> descriptor)
+        {
+            descriptor
+                .Initially()
+                .OnEvent<StartFaultingCommandEvent>()
+                .StateFactory(_ => new CommandState())
+                .Send((_, _) => new FaultingCommand())
+                .TransitionTo("Awaiting");
+
+            descriptor
+                .During("Awaiting")
+                .OnReplyFault()
+                .Then((_, fault) => s_faultObserved.TrySetResult(fault))
+                .TransitionTo("Failed");
+
+            descriptor.Finally("Failed");
+        }
+    }
+
+    public sealed class SucceedingCommandSaga : Saga<CommandState>
+    {
+        protected override void Configure(ISagaDescriptor<CommandState> descriptor)
+        {
+            descriptor
+                .Initially()
+                .OnEvent<StartSucceedingCommandEvent>()
+                .StateFactory(_ => new CommandState())
+                .Send((_, _) => new SucceedingCommand())
+                .TransitionTo("Awaiting");
+
+            descriptor
+                .During("Awaiting")
+                .OnAnyReply()
+                .Then((_, reply) => s_replyObserved.TrySetResult(reply))
+                .TransitionTo("Done");
+
+            descriptor.During("Awaiting").OnReplyFault().TransitionTo("Done");
+
+            descriptor.Finally("Done");
+        }
+    }
+}