Merge pull request #288 from Vulthil/fix/rabbitmq-dispatch-scoping

fix: scope dispatch plans per queue and serialize RPC replies through the consumer channel gate

Author: Vulthil

docs/articles/messaging.md

@@ -859,8 +859,13 @@ internal sealed class MyHandlerFactory : IMessageHandlerFactory<Dispatch>
 ```
 
 Let `MessageExecutionRegistry<THandler>` assemble the per-message-type plans from the configured
-queues — it handles URN keying, polymorphic fan-out, deduplication, request-consumer uniqueness,
-and partition attachment:
+queues — it handles URN keying, polymorphic fan-out, deduplication, request-consumer uniqueness
+(at most one per message type per queue), and partition attachment. Plans are keyed by URN within
+a registry instance, so queues registered into the same instance accumulate their handlers into
+one plan per message type. Register every queue in a single instance only when your transport
+dispatches each produced message exactly once (as the in-memory test harness does); a transport
+that receives a distinct delivery per queue (as the RabbitMQ transport does) must build one
+registry per queue so a delivery dispatches only the handlers its own queue registered:
 
 ```csharp
 var registry = new MessageExecutionRegistry<Dispatch>(provider, new MyHandlerFactory());

src/Vulthil.Messaging.RabbitMq/Consumers/GatedPublisher.cs

@@ -0,0 +1,17 @@
+using RabbitMQ.Client;
+
+namespace Vulthil.Messaging.RabbitMq.Consumers;
+
+/// <summary>
+/// Publishes a message on a consumer worker's shared channel, serialized through the worker's channel gate
+/// alongside the delivery settles (acks, nacks, retry republishes and fault publishes) — RabbitMQ channels must
+/// not be used concurrently. Handed to <see cref="MessageHandler.DispatchAsync"/> in place of the raw channel so
+/// a request/reply handler's response publish cannot interleave frames with a parallel settle on the same channel.
+/// </summary>
+/// <param name="exchange">The exchange to publish to; an empty string targets the broker's default exchange.</param>
+/// <param name="routingKey">The routing key (the destination queue name when using the default exchange).</param>
+/// <param name="mandatory">Whether the broker should return the message when it cannot be routed.</param>
+/// <param name="basicProperties">The AMQP properties to publish with.</param>
+/// <param name="body">The serialized message body.</param>
+/// <returns>A task that completes once the publish has been written to the channel.</returns>
+internal delegate Task GatedPublisher(string exchange, string routingKey, bool mandatory, BasicProperties basicProperties, ReadOnlyMemory<byte> body);

src/Vulthil.Messaging.RabbitMq/Consumers/MessageHandler.cs

@@ -1,4 +1,3 @@
-using RabbitMQ.Client;
 using RabbitMQ.Client.Events;
 using Vulthil.Messaging.Queues;
 using Vulthil.Messaging.Transport;
@@ -20,10 +19,11 @@ internal sealed record MessageHandler
     public required HandlerKind Kind { get; init; }
 
     /// <summary>
-    /// Dispatches a deserialized message through the consume pipeline and (for RPC) publishes the response on the supplied channel.
-    /// Consumer-kind handlers ignore the channel parameter. The envelope is non-null on the standard receive path
-    /// (Vulthil-produced messages) and null on the bare-JSON compat path (external producers); the closure picks the
-    /// appropriate <c>MessageContextFactory.CreateContext</c> overload.
+    /// Dispatches a deserialized message through the consume pipeline and (for RPC) publishes the response through
+    /// the supplied <see cref="GatedPublisher"/>, which serializes the write with the worker's other channel
+    /// operations. Consumer-kind handlers ignore the publisher parameter. The envelope is non-null on the standard
+    /// receive path (Vulthil-produced messages) and null on the bare-JSON compat path (external producers); the
+    /// closure picks the appropriate <c>MessageContextFactory.CreateContext</c> overload.
     /// </summary>
-    public required Func<IServiceProvider, object, BasicDeliverEventArgs, MessageEnvelope?, IChannel, CancellationToken, Task> DispatchAsync { get; init; }
+    public required Func<IServiceProvider, object, BasicDeliverEventArgs, MessageEnvelope?, GatedPublisher, CancellationToken, Task> DispatchAsync { get; init; }
 }

src/Vulthil.Messaging.RabbitMq/Consumers/MessageHandlerFactory.cs

@@ -54,7 +54,7 @@ public static MessageHandler ForRequestConsumer<TConsumer, TRequest, TResponse>(
         {
             RetryPolicy = retryPolicy,
             Kind = HandlerKind.RequestConsumer,
-            DispatchAsync = async (sp, message, ea, envelope, channel, ct) =>
+            DispatchAsync = async (sp, message, ea, envelope, publishAsync, ct) =>
             {
                 var consumer = sp.GetRequiredService<TConsumer>();
                 var publisher = sp.GetRequiredService<IPublisher>();
@@ -98,7 +98,7 @@ public static MessageHandler ForRequestConsumer<TConsumer, TRequest, TResponse>(
                     reply = BuildFaultReply(exception.Message, exception.GetType().FullName ?? "Unknown", exception.StackTrace, jsonOptions, ea, envelope);
                 }
 
-                await SendResponseAsync(ea, reply, channel, jsonOptions);
+                await SendResponseAsync(ea, reply, publishAsync, jsonOptions);
             }
         };
 
@@ -131,7 +131,7 @@ private static MessageEnvelope BuildFaultReply(
         return BuildReply(RpcFault.UrnUri, JsonSerializer.SerializeToElement(fault, jsonOptions), ea, requestEnvelope);
     }
 
-    private static async Task SendResponseAsync(BasicDeliverEventArgs ea, MessageEnvelope reply, IChannel channel, JsonSerializerOptions jsonOptions)
+    private static async Task SendResponseAsync(BasicDeliverEventArgs ea, MessageEnvelope reply, GatedPublisher publishAsync, JsonSerializerOptions jsonOptions)
     {
         if (string.IsNullOrEmpty(ea.BasicProperties.ReplyTo))
         {
@@ -146,6 +146,6 @@ private static async Task SendResponseAsync(BasicDeliverEventArgs ea, MessageEnv
             ContentType = RabbitMqConstants.ContentType,
         };
 
-        await channel.BasicPublishAsync(string.Empty, ea.BasicProperties.ReplyTo, true, replyProps, body);
+        await publishAsync(string.Empty, ea.BasicProperties.ReplyTo, true, replyProps, body);
     }
 }

src/Vulthil.Messaging.RabbitMq/Consumers/MessageTypeCache.cs

@@ -7,7 +7,8 @@ namespace Vulthil.Messaging.RabbitMq.Consumers;
 /// RabbitMQ adapter over <see cref="MessageExecutionRegistry{THandler}"/>. Delegates plan assembly to the core
 /// registry and decorates each resolved plan with a <see cref="RabbitMqPlan"/> carrying the AMQP partition key
 /// extractor. Wrappers are built once per plan (keyed by URN) during registration, so delivery-time lookups stay
-/// read-only.
+/// read-only. The bus builds one cache per queue, so a worker's plan lookups resolve only the handlers its own
+/// queue registered even when several queues consume the same message type.
 /// </summary>
 internal sealed class MessageTypeCache
 {

src/Vulthil.Messaging.RabbitMq/Consumers/RabbitMqConsumerWorker.cs

@@ -26,11 +26,12 @@ internal sealed class RabbitMqConsumerWorker : IAsyncDisposable
     private readonly bool _partitioned;
     private readonly ConcurrentDictionary<ulong, Task> _inFlight = new();
 
-    // RabbitMQ channels must not be used concurrently. On a partitioned queue the lanes complete in parallel
-    // and each settles its delivery (ack/nack/retry-republish/fault) on this shared channel, so every channel
-    // write is serialized through this gate to avoid interleaved frames. Message processing stays parallel;
-    // only the brief settle/publish frames are serialized.
+    // RabbitMQ channels must not be used concurrently. Concurrent dispatch (and, on a partitioned queue, the
+    // lanes completing in parallel) settles deliveries (ack/nack/retry-republish/fault) and publishes RPC
+    // replies on this shared channel, so every channel write is serialized through this gate to avoid
+    // interleaved frames. Message processing stays parallel; only the brief settle/publish frames are serialized.
     private readonly SemaphoreSlim _channelGate = new(1, 1);
+    private readonly GatedPublisher _gatedPublisher;
 
     private JsonSerializerOptions _jsonOptions => _messageConfigurationProvider.JsonSerializerOptions;
 
@@ -54,6 +55,7 @@ public RabbitMqConsumerWorker(
         _logger = logger;
         _channelIndex = channelIndex;
         _partitioned = partitioned;
+        _gatedPublisher = PublishThroughGateAsync;
     }
 
     /// <summary>
@@ -77,6 +79,14 @@ private async Task OnChannelAsync(Func<ValueTask> channelOperation)
 
     private Task NackAsync(BasicDeliverEventArgs ea) => OnChannelAsync(() => _channel.BasicNackAsync(ea.DeliveryTag, false, requeue: false));
 
+    /// <summary>
+    /// Publishes on the shared consumer channel through the channel gate. Handed to handler dispatch as the
+    /// <see cref="GatedPublisher"/> so a request/reply response serializes with the acks, nacks and republishes
+    /// settling other deliveries on this channel.
+    /// </summary>
+    private Task PublishThroughGateAsync(string exchange, string routingKey, bool mandatory, BasicProperties basicProperties, ReadOnlyMemory<byte> body)
+        => OnChannelAsync(() => _channel.BasicPublishAsync(exchange, routingKey, mandatory, basicProperties, body));
+
     public async Task StartAsync(CancellationToken cancellationToken)
     {
         var consumer = new AsyncEventingBasicConsumer(_channel);
@@ -280,7 +290,7 @@ private async Task HandleFailureAsync(Exception ex, BasicDeliverEventArgs ea, st
 
             props.Expiration = delay.TotalMilliseconds.ToString(CultureInfo.InvariantCulture);
 
-            await OnChannelAsync(() => _channel.BasicPublishAsync($"{_queueDefinition.Name}.Retry", ea.RoutingKey, true, props, ea.Body));
+            await PublishThroughGateAsync($"{_queueDefinition.Name}.Retry", ea.RoutingKey, true, props, ea.Body);
             await AckAsync(ea);
             return;
         }
@@ -323,7 +333,7 @@ private async Task PublishFaultAsync(Exception ex, BasicDeliverEventArgs ea, IDi
                 Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds())
             };
 
-            await OnChannelAsync(() => _channel.BasicPublishAsync(exchange, routingKey, false, faultProps, faultBody));
+            await PublishThroughGateAsync(exchange, routingKey, false, faultProps, faultBody);
         }
         catch (Exception faultEx)
         {
@@ -429,7 +439,7 @@ private async Task DispatchHandlersAsync(RabbitMqPlan plan, object message, Basi
 
         foreach (var handler in plan.Handlers)
         {
-            await handler.DispatchAsync(scope.ServiceProvider, message, ea, envelope, _channel, ea.CancellationToken);
+            await handler.DispatchAsync(scope.ServiceProvider, message, ea, envelope, _gatedPublisher, ea.CancellationToken);
         }
     }
 

src/Vulthil.Messaging.RabbitMq/RabbitMqBus.cs

@@ -37,22 +37,21 @@ public RabbitMqBus(
     /// <remarks>
     /// A failed start disposes any partially-created consumer workers and rethrows without faulting the readiness
     /// signal, so the hosting consumer service can retry a transient failure (such as a broker that is still coming
-    /// up) while <see cref="RabbitMqBusStartupStatus.Ready"/> stays pending until a start attempt succeeds. A fresh
-    /// type cache is built for every attempt (rather than kept as instance state) so a retry after a partial failure
-    /// re-registers queues against an empty registry instead of appending to handlers (or request-consumer
-    /// bookkeeping) left over from the attempt that failed.
+    /// up) while <see cref="RabbitMqBusStartupStatus.Ready"/> stays pending until a start attempt succeeds. Fresh
+    /// per-queue type caches are built for every attempt (rather than kept as instance state) so a retry after a
+    /// partial failure re-registers each queue against an empty registry instead of appending to handlers (or
+    /// request-consumer bookkeeping) left over from the attempt that failed.
     /// </remarks>
     public async Task StartAsync(CancellationToken cancellationToken = default)
     {
-        var typeCache = new MessageTypeCache(_messageConfigurationProvider);
-
         try
         {
             var queues = _messageConfigurationProvider.QueueDefinitions;
             MessagingLog.BusStarting(_logger, queues.Count);
 
-            await SetupTopology(queues, typeCache, cancellationToken);
-            await StartConsumersAsync(queues, typeCache, cancellationToken);
+            var typeCaches = BuildTypeCaches(queues);
+            await SetupTopology(queues, typeCaches, cancellationToken);
+            await StartConsumersAsync(queues, typeCaches, cancellationToken);
 
             MessagingLog.BusStarted(_logger);
             _startupStatus.MarkStarted();
@@ -64,6 +63,25 @@ public async Task StartAsync(CancellationToken cancellationToken = default)
         }
     }
 
+    /// <summary>
+    /// Builds one <see cref="MessageTypeCache"/> per queue, each registering only that queue. Several queues may
+    /// consume the same message type, and the broker delivers a distinct copy to each of them; scoping the plan
+    /// cache to its queue keeps a delivery from also dispatching the consumers every other queue registered for
+    /// that type.
+    /// </summary>
+    internal Dictionary<string, MessageTypeCache> BuildTypeCaches(IReadOnlyCollection<QueueDefinition> queues)
+    {
+        var typeCaches = new Dictionary<string, MessageTypeCache>(StringComparer.OrdinalIgnoreCase);
+        foreach (var queue in queues)
+        {
+            var typeCache = new MessageTypeCache(_messageConfigurationProvider);
+            typeCache.RegisterQueue(queue);
+            typeCaches[queue.Name] = typeCache;
+        }
+
+        return typeCaches;
+    }
+
     public Task WaitUntilReadyAsync(CancellationToken cancellationToken = default) =>
         _startupStatus.Ready.WaitAsync(cancellationToken);
 
@@ -72,14 +90,13 @@ public Task WaitUntilReadyAsync(CancellationToken cancellationToken = default) =
     /// partition lanes in arrival order; parallelism comes from the lanes (bounded by <c>PrefetchCount</c>) rather
     /// than concurrent dispatch.
     /// </remarks>
-    private async Task StartConsumersAsync(IReadOnlyCollection<QueueDefinition> queues, MessageTypeCache typeCache, CancellationToken cancellationToken)
+    private async Task StartConsumersAsync(IReadOnlyCollection<QueueDefinition> queues, Dictionary<string, MessageTypeCache> typeCaches, CancellationToken cancellationToken)
     {
         var workerLogger = _loggerFactory.CreateLogger<RabbitMqConsumerWorker>();
 
         foreach (var queue in queues)
         {
-            typeCache.RegisterQueue(queue);
-
+            var typeCache = typeCaches[queue.Name];
             var partitioned = typeCache.IsQueuePartitioned(queue);
             var channelCount = partitioned ? 1 : queue.ChannelCount;
             var dispatchConcurrency = partitioned ? (ushort)1 : queue.ConcurrencyLimit;
@@ -116,7 +133,7 @@ private async Task StartConsumersAsync(IReadOnlyCollection<QueueDefinition> queu
     /// here by convention with the faulted message's URN as the routing key, so a subscriber binds its queue with
     /// <c>"#"</c> to observe all faults or with a specific URN to filter by faulted message type.
     /// </remarks>
-    private async Task SetupTopology(IReadOnlyCollection<QueueDefinition> queues, MessageTypeCache typeCache, CancellationToken cancellationToken)
+    private async Task SetupTopology(IReadOnlyCollection<QueueDefinition> queues, Dictionary<string, MessageTypeCache> typeCaches, CancellationToken cancellationToken)
     {
         using var channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);
 
@@ -129,7 +146,7 @@ await channel.ExchangeDeclareAsync(
 
         foreach (var queue in queues)
         {
-            await SetupQueueTopology(queue, typeCache, channel, cancellationToken);
+            await SetupQueueTopology(queue, typeCaches[queue.Name], channel, cancellationToken);
             MessagingLog.QueueDeclared(_logger, queue.Name, queue.Registrations.Count);
         }
     }