Merge pull request #284 from Vulthil/fix/messaging-transport-lifecycle

Vulthil · web-flow · commit 4b8833a56dc2 · 2026-07-04T13:15:55.000+02:00
Fix RabbitMQ transport startup and shutdown lifecycle bugs
diff --git a/src/Vulthil.Messaging.RabbitMq/Consumers/RabbitMqConsumerWorker.cs b/src/Vulthil.Messaging.RabbitMq/Consumers/RabbitMqConsumerWorker.cs
@@ -164,6 +164,11 @@ private async Task ProcessAsync(PreparedDelivery prepared, BasicDeliverEventArgs
         }
         catch (Exception ex)
         {
+            if (ex is OperationCanceledException && ea.CancellationToken.IsCancellationRequested)
+            {
+                return;
+            }
+
             activity?.SetStatus(ActivityStatusCode.Error, ex.Message);
             activity?.AddException(ex);
             await HandleFailureAsync(ex, ea, prepared.DiagnosticTypeName);
diff --git a/src/Vulthil.Messaging.RabbitMq/RabbitMqBus.cs b/src/Vulthil.Messaging.RabbitMq/RabbitMqBus.cs
@@ -16,7 +16,6 @@ internal sealed class RabbitMqBus : ITransport, IAsyncDisposable
     private readonly RabbitMqBusStartupStatus _startupStatus;
     private readonly ILogger<RabbitMqBus> _logger;
     private readonly ILoggerFactory _loggerFactory;
-    private readonly MessageTypeCache _typeCache;
     private readonly List<RabbitMqConsumerWorker> _workers = [];
 
     public RabbitMqBus(
@@ -33,23 +32,27 @@ public RabbitMqBus(
         _startupStatus = startupStatus;
         _logger = logger;
         _loggerFactory = loggerFactory;
-        _typeCache = new MessageTypeCache(messageConfigurationProvider);
     }
 
     /// <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.
+    /// 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.
     /// </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, cancellationToken);
-            await StartConsumersAsync(queues, cancellationToken);
+            await SetupTopology(queues, typeCache, cancellationToken);
+            await StartConsumersAsync(queues, typeCache, cancellationToken);
 
             MessagingLog.BusStarted(_logger);
             _startupStatus.MarkStarted();
@@ -69,15 +72,15 @@ 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, CancellationToken cancellationToken)
+    private async Task StartConsumersAsync(IReadOnlyCollection<QueueDefinition> queues, MessageTypeCache typeCache, CancellationToken cancellationToken)
     {
         var workerLogger = _loggerFactory.CreateLogger<RabbitMqConsumerWorker>();
 
         foreach (var queue in queues)
         {
-            _typeCache.RegisterQueue(queue);
+            typeCache.RegisterQueue(queue);
 
-            var partitioned = _typeCache.IsQueuePartitioned(queue);
+            var partitioned = typeCache.IsQueuePartitioned(queue);
             var channelCount = partitioned ? 1 : queue.ChannelCount;
             var dispatchConcurrency = partitioned ? (ushort)1 : queue.ConcurrencyLimit;
 
@@ -96,7 +99,7 @@ private async Task StartConsumersAsync(IReadOnlyCollection<QueueDefinition> queu
                     _serviceScopeFactory,
                     queue,
                     channel,
-                    _typeCache,
+                    typeCache,
                     _messageConfigurationProvider,
                     workerLogger,
                     i,
@@ -113,7 +116,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, CancellationToken cancellationToken)
+    private async Task SetupTopology(IReadOnlyCollection<QueueDefinition> queues, MessageTypeCache typeCache, CancellationToken cancellationToken)
     {
         using var channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);
 
@@ -126,7 +129,7 @@ await channel.ExchangeDeclareAsync(
 
         foreach (var queue in queues)
         {
-            await SetupQueueTopology(queue, channel, cancellationToken);
+            await SetupQueueTopology(queue, typeCache, channel, cancellationToken);
             MessagingLog.QueueDeclared(_logger, queue.Name, queue.Registrations.Count);
         }
     }
@@ -136,7 +139,7 @@ await channel.ExchangeDeclareAsync(
     /// active (others stand by for failover) so ordering survives across load-balanced consumers. Partitioned queues
     /// opt in automatically; any queue can request it explicitly.
     /// </remarks>
-    private async Task SetupQueueTopology(QueueDefinition queue, IChannel channel, CancellationToken cancellationToken)
+    private async Task SetupQueueTopology(QueueDefinition queue, MessageTypeCache typeCache, IChannel channel, CancellationToken cancellationToken)
     {
         await channel.ExchangeDeclareAsync(
             exchange: queue.Name,
@@ -152,7 +155,7 @@ await channel.ExchangeDeclareAsync(
             args.Add("x-queue-type", "quorum");
         }
 
-        if (queue.SingleActiveConsumer || _typeCache.IsQueuePartitioned(queue))
+        if (queue.SingleActiveConsumer || typeCache.IsQueuePartitioned(queue))
         {
             args.Add("x-single-active-consumer", true);
         }
diff --git a/src/Vulthil.Messaging/ConsumerHostedService.cs b/src/Vulthil.Messaging/ConsumerHostedService.cs
@@ -1,3 +1,4 @@
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 
@@ -12,25 +13,42 @@ namespace Vulthil.Messaging;
 /// that is still coming up is a transient infrastructure condition rather than a reason to fault the host. The host's
 /// <c>BackgroundServiceExceptionBehavior</c> is left untouched, so a genuine fault raised once the transport is running
 /// still surfaces and stops the host as usual.
+/// <para>
+/// The generic host swallows exceptions thrown from inside <see cref="BackgroundService.ExecuteAsync"/> — even ones
+/// thrown synchronously before the first await — logging them and stopping the host without ever propagating them out
+/// of <c>IHost.StartAsync</c>. A hosted service constructor throwing is not subject to that: it fails DI resolution
+/// directly, so <c>IHost.StartAsync</c> throws it as-is. The clear "no transport registered" error therefore still
+/// needs to fire from the constructor to surface synchronously; it does so via <see cref="IServiceProviderIsService"/>,
+/// which answers whether <see cref="ITransport"/> is registered without constructing one. The transport instance
+/// itself is resolved lazily, inside the retry loop, so a transport whose construction depends on an unreachable
+/// resource (such as a broker connection) is treated as a retryable startup failure instead.
+/// </para>
 /// </remarks>
 internal sealed class ConsumerHostedService : BackgroundService
 {
+    private const string NoTransportRegisteredMessage =
+        "No messaging transport is registered. Call a transport extension such as .UseRabbitMq(...) " +
+        "inside AddMessaging(...) (or .UseTestHarness() in a test).";
+
     private static readonly TimeSpan InitialRetryDelay = TimeSpan.FromSeconds(1);
     private static readonly TimeSpan MaxRetryDelay = TimeSpan.FromSeconds(30);
 
-    private readonly ITransport _transport;
+    private readonly IServiceProvider _serviceProvider;
     private readonly TimeProvider _timeProvider;
     private readonly ILogger<ConsumerHostedService> _logger;
 
     public ConsumerHostedService(
-        IEnumerable<ITransport> transports,
+        IServiceProvider serviceProvider,
+        IServiceProviderIsService serviceProviderIsService,
         TimeProvider timeProvider,
         ILogger<ConsumerHostedService> logger)
     {
-        _transport = transports.LastOrDefault()
-            ?? throw new InvalidOperationException(
-                "No messaging transport is registered. Call a transport extension such as .UseRabbitMq(...) " +
-                "inside AddMessaging(...) (or .UseTestHarness() in a test).");
+        if (!serviceProviderIsService.IsService(typeof(ITransport)))
+        {
+            throw new InvalidOperationException(NoTransportRegisteredMessage);
+        }
+
+        _serviceProvider = serviceProvider;
         _timeProvider = timeProvider;
         _logger = logger;
     }
@@ -43,7 +61,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
         {
             try
             {
-                await _transport.StartAsync(stoppingToken);
+                await ResolveTransport().StartAsync(stoppingToken);
                 return;
             }
             catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
@@ -64,6 +82,10 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
         }
     }
 
+    private ITransport ResolveTransport() =>
+        _serviceProvider.GetServices<ITransport>().LastOrDefault()
+            ?? throw new InvalidOperationException(NoTransportRegisteredMessage);
+
     private async Task<bool> TryDelayAsync(TimeSpan delay, CancellationToken stoppingToken)
     {
         try
diff --git a/tests/Vulthil.Messaging.RabbitMq.Tests/AutoMockerServiceScopeFactory.cs b/tests/Vulthil.Messaging.RabbitMq.Tests/AutoMockerServiceScopeFactory.cs
@@ -0,0 +1,23 @@
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Vulthil.Messaging.RabbitMq.Tests;
+
+/// <summary>
+/// Adapts an <see cref="IServiceProvider"/> — typically an <c>AutoMocker</c> instance, which itself implements
+/// <see cref="IServiceProvider"/> — into an <see cref="IServiceScopeFactory"/> whose scopes resolve directly
+/// from it. Lets a test exercise a transport's real create-scope-per-delivery code path without standing up a
+/// full DI container.
+/// </summary>
+internal sealed class AutoMockerServiceScopeFactory(IServiceProvider serviceProvider) : IServiceScopeFactory
+{
+    public IServiceScope CreateScope() => new PassthroughServiceScope(serviceProvider);
+
+    private sealed class PassthroughServiceScope(IServiceProvider serviceProvider) : IServiceScope
+    {
+        public IServiceProvider ServiceProvider { get; } = serviceProvider;
+
+        public void Dispose()
+        {
+        }
+    }
+}
diff --git a/tests/Vulthil.Messaging.RabbitMq.Tests/RabbitMqBusStartupRetryTests.cs b/tests/Vulthil.Messaging.RabbitMq.Tests/RabbitMqBusStartupRetryTests.cs
@@ -0,0 +1,99 @@
+using System.Text.Json;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+using Vulthil.Messaging.Abstractions.Consumers;
+using Vulthil.Messaging.RabbitMq.HealthChecks;
+using Vulthil.xUnit;
+
+namespace Vulthil.Messaging.RabbitMq.Tests;
+
+public sealed class RabbitMqBusStartupRetryTests : BaseUnitTestCase
+{
+    private readonly Mock<IChannel> _channel = new();
+    private readonly Lazy<RabbitMqBus> _lazyTarget;
+    private IAsyncBasicConsumer? _capturedConsumer;
+
+    private RabbitMqBus Target => _lazyTarget.Value;
+
+    public RabbitMqBusStartupRetryTests()
+    {
+        _channel
+            .Setup(c => c.BasicConsumeAsync(
+                It.IsAny<string>(), It.IsAny<bool>(), It.IsAny<string>(), It.IsAny<bool>(), It.IsAny<bool>(),
+                It.IsAny<IDictionary<string, object?>>(), It.IsAny<IAsyncBasicConsumer>(), It.IsAny<CancellationToken>()))
+            .Callback((string _, bool _, string _, bool _, bool _, IDictionary<string, object?> _, IAsyncBasicConsumer consumer, CancellationToken _) =>
+                _capturedConsumer = consumer)
+            .ReturnsAsync("consumer-tag");
+
+        Use(new RabbitMqBusStartupStatus());
+        Use<ILoggerFactory>(NullLoggerFactory.Instance);
+        Use<ILogger<RabbitMqBus>>(NullLogger<RabbitMqBus>.Instance);
+        Use<IServiceScopeFactory>(new AutoMockerServiceScopeFactory(AutoMocker));
+
+        _lazyTarget = new(CreateInstance<RabbitMqBus>);
+    }
+
+    protected override ValueTask Dispose() => _lazyTarget.IsValueCreated ? Target.DisposeAsync() : base.Dispose();
+
+    [Fact]
+    public async Task StartAsyncFailThenRetrySucceedsWithoutDoublingHandlersOrRejectingTheRpcConsumer()
+    {
+        // Arrange
+        var handlerHits = 0;
+        Use(new RecordingConsumer(() => handlerHits++));
+        Use<IEnumerable<IConsumeFilter<TestMessage>>>([]);
+
+        var provider = TestProviders.Build(cfg => cfg.ConfigureQueue("orders", queue =>
+        {
+            queue.AddConsumer<RecordingConsumer>();
+            queue.AddRequestConsumer<RecordingRequestConsumer>();
+        }));
+        Use(provider);
+
+        GetMock<IConnection>()
+            .SetupSequence(c => c.CreateChannelAsync(It.IsAny<CreateChannelOptions?>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(_channel.Object)
+            .ThrowsAsync(new InvalidOperationException("connection blip"))
+            .ReturnsAsync(_channel.Object)
+            .ReturnsAsync(_channel.Object);
+
+        // Act
+        await Should.ThrowAsync<InvalidOperationException>(() => Target.StartAsync(CancellationToken));
+        await Target.StartAsync(CancellationToken);
+
+        await _capturedConsumer!.HandleBasicDeliverAsync(
+            "consumer-tag",
+            1,
+            false,
+            "orders",
+            "orders",
+            new BasicProperties { Type = typeof(TestMessage).FullName, Headers = new Dictionary<string, object?>() },
+            JsonSerializer.SerializeToUtf8Bytes(new TestMessage("payload")),
+            CancellationToken.None);
+
+        // Assert
+        handlerHits.ShouldBe(1);
+    }
+
+    internal sealed record TestMessage(string Value);
+    internal sealed record TestRequest(string Query);
+    internal sealed record TestResponse(string Result);
+
+    private sealed class RecordingConsumer(Action onConsume) : IConsumer<TestMessage>
+    {
+        public Task ConsumeAsync(IMessageContext<TestMessage> messageContext, CancellationToken cancellationToken = default)
+        {
+            onConsume();
+            return Task.CompletedTask;
+        }
+    }
+
+    private sealed class RecordingRequestConsumer : IRequestConsumer<TestRequest, TestResponse>
+    {
+        public Task<TestResponse> ConsumeAsync(IMessageContext<TestRequest> messageContext, CancellationToken cancellationToken = default)
+            => Task.FromResult(new TestResponse($"Processed: {messageContext.Message.Query}"));
+    }
+}
diff --git a/tests/Vulthil.Messaging.RabbitMq.Tests/RabbitMqConsumerWorkerTests.cs b/tests/Vulthil.Messaging.RabbitMq.Tests/RabbitMqConsumerWorkerTests.cs
diff --git a/tests/Vulthil.Messaging.Tests/ConsumerHostedServiceTests.cs b/tests/Vulthil.Messaging.Tests/ConsumerHostedServiceTests.cs

Original file line number	Diff line number	Diff line change
`@@ -164,6 +164,11 @@ private async Task ProcessAsync(PreparedDelivery prepared, BasicDeliverEventArgs`
`164`	`164`	`}`
`165`	`165`	`catch (Exception ex)`
`166`	`166`	`{`
	`167`	`+ if (ex is OperationCanceledException && ea.CancellationToken.IsCancellationRequested)`
	`168`	`+ {`
	`169`	`+ return;`
	`170`	`+ }`
	`171`	`+`
`167`	`172`	`activity?.SetStatus(ActivityStatusCode.Error, ex.Message);`
`168`	`173`	`activity?.AddException(ex);`
`169`	`174`	`await HandleFailureAsync(ex, ea, prepared.DiagnosticTypeName);`