Stop dereferencing a null channel when a RabbitMQ listener is disposed mid-startup (GH-3842)

src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/Bugs/Bug_3842_listener_create_with_no_channel.cs

@@ -0,0 +1,103 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using NSubstitute;
+using Shouldly;
+using Wolverine.RabbitMQ.Internal;
+using Wolverine.Runtime;
+using Wolverine.Transports;
+using Xunit;
+
+namespace Wolverine.RabbitMQ.Tests.Bugs;
+
+/// <summary>
+/// GH-3842. RabbitMqChannelAgent.EnsureInitiated() is best-effort: it returns without a channel when the
+/// agent has been disposed, and it logs-and-swallows a failure to open one. CreateAsync() used to call
+/// `Queue.DeclareAsync(Channel!, Logger)` straight afterwards, so both outcomes surfaced as a bare
+/// NullReferenceException from RabbitMqQueue.DeclareAsync -- six frames from the actual cause.
+///
+/// This was the mechanism behind the intermittent failure of
+/// Bug_189_fails_if_there_are_many_messages_in_queue_on_startup under concurrent host lifecycle.
+/// </summary>
+public class Bug_3842_listener_create_with_no_channel : IAsyncLifetime
+{
+    private IHost _host = null!;
+    private RabbitMqTransport _transport = null!;
+    private RabbitMqQueue _queue = null!;
+
+    public async ValueTask InitializeAsync()
+    {
+        var queueName = "gh3842-" + Guid.NewGuid().ToString("n").Substring(0, 8);
+
+        _host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.UseRabbitMq().AutoProvision();
+                opts.ListenToRabbitQueue(queueName);
+            }).StartAsync();
+
+        var runtime = _host.Services.GetRequiredService<IWolverineRuntime>();
+        _transport = runtime.Options.Transports.GetOrCreate<RabbitMqTransport>();
+        _queue = _transport.Queues[queueName];
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        try
+        {
+            await _host.StopAsync();
+        }
+        catch (Exception)
+        {
+            // One of these tests deliberately disposes the transport's listening connection, so an
+            // orderly shutdown is not always possible afterwards. The assertion has already run.
+        }
+
+        _host.Dispose();
+    }
+
+    private RabbitMqListener buildListener()
+    {
+        var runtime = _host.Services.GetRequiredService<IWolverineRuntime>();
+        return new RabbitMqListener(runtime, _queue, _transport, Substitute.For<IReceiver>());
+    }
+
+    [Fact]
+    public async Task disposed_during_startup_abandons_creation_quietly()
+    {
+        var listener = buildListener();
+
+        // The race as it happens in the field: the host stops while this listener is still coming up,
+        // so EnsureInitiated() returns early and never opens a channel.
+        await listener.DisposeAsync();
+        listener.Channel.ShouldBeNull();
+
+        // Previously threw NullReferenceException from RabbitMqQueue.DeclareAsync.
+        await Should.NotThrowAsync(() => listener.CreateAsync());
+    }
+
+    [Fact]
+    public async Task a_live_agent_that_cannot_open_a_channel_throws_something_diagnosable()
+    {
+        // Take the connection out from under the transport so that startNewChannel() genuinely fails.
+        // EnsureInitiated() logs and swallows that, then returns with Channel still null and the agent
+        // very much alive -- the second of its two no-channel exits, and the one that is an error.
+        //
+        // Note it is not enough to just null out Channel: EnsureInitiated() would simply open a fresh
+        // one and the branch under test would never be reached.
+        await _transport.ListeningConnection.DisposeAsync();
+
+        var listener = buildListener();
+        await listener.EnsureInitiated();
+        listener.Channel.ShouldBeNull();
+        listener.IsDisposed.ShouldBeFalse();
+
+        var ex = await Should.ThrowAsync<InvalidOperationException>(() => listener.CreateAsync());
+
+        // The message has to name the endpoint and the queue. The whole complaint in GH-3842 is that a
+        // bare NullReferenceException six frames deep told you neither.
+        ex.Message.ShouldContain(_queue.QueueName);
+        ex.Message.ShouldContain("Unable to open a Rabbit MQ channel");
+
+        await listener.DisposeAsync();
+    }
+}

src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqChannelAgent.cs

@@ -34,6 +34,13 @@ protected RabbitMqChannelAgent(ConnectionMonitor monitor,
 
     internal IChannel? Channel { get; set; }
 
+    /// <summary>
+    /// True once DisposeAsync has run. Callers of <see cref="EnsureInitiated"/> need this to tell the two
+    /// ways it can return without a channel apart: disposal is a legitimate outcome that should be handled
+    /// quietly, whereas a swallowed channel-creation failure is not.
+    /// </summary>
+    internal bool IsDisposed => _disposed;
+
     public virtual async ValueTask DisposeAsync()
     {
         if (_disposed)
@@ -49,6 +56,13 @@ public virtual async ValueTask DisposeAsync()
         // SemaphoreSlim finalizer. See #3132.
     }
 
+    /// <summary>
+    /// Best-effort: brings <see cref="Channel"/> up if it is missing or dead. This method deliberately does
+    /// NOT guarantee a channel on return -- it returns without one when the agent has been disposed, and it
+    /// logs-and-swallows a failure to open one. Callers must therefore null-check <see cref="Channel"/>
+    /// rather than assume success; see GH-3842, where a `Channel!` in RabbitMqListener.CreateAsync turned
+    /// both of those outcomes into a bare NullReferenceException six frames away in queue declaration.
+    /// </summary>
     internal async Task EnsureInitiated()
     {
         if (_disposed)

src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqListener.cs

@@ -225,20 +225,42 @@ public async Task CreateAsync()
     {
         await EnsureInitiated();
 
+        // EnsureInitiated is best-effort and can return without a channel two different ways, so the
+        // channel is captured once and checked rather than dereferenced through `Channel!` (GH-3842).
+        // Reading the property repeatedly would also race a concurrent rebuild replacing it mid-method.
+        var channel = Channel;
+        if (channel is null)
+        {
+            // Disposal during startup is legitimate -- a host that stops while its listeners are still
+            // coming up hits this routinely, and there is nothing left to build against.
+            if (IsDisposed)
+            {
+                Logger.LogDebug(
+                    "Rabbit MQ listener at {Uri} was disposed while starting up; abandoning listener creation.",
+                    Address);
+                return;
+            }
+
+            // Otherwise EnsureInitiated logged and swallowed a channel-creation failure. Say so here,
+            // instead of letting the null surface as a NullReferenceException inside queue declaration.
+            throw new InvalidOperationException(
+                $"Unable to open a Rabbit MQ channel for listener {Address} (queue '{Queue.QueueName}'). The underlying failure was logged by the channel agent.");
+        }
+
         if (Queue.AutoDelete || _transport.AutoProvision)
         {
-            await Queue.DeclareAsync(Channel!, Logger);
+            await Queue.DeclareAsync(channel, Logger);
 
             if (Queue.DeadLetterQueue != null && Queue.DeadLetterQueue.Mode != DeadLetterQueueMode.WolverineStorage)
             {
                 var dlq = _transport.Queues[Queue.DeadLetterQueue.QueueName];
-                await dlq.DeclareAsync(Channel!, Logger);
+                await dlq.DeclareAsync(channel, Logger);
             }
         }
 
         try
         {
-            var result = await Channel!.QueueDeclarePassiveAsync(Queue.QueueName, _cancellation);
+            var result = await channel.QueueDeclarePassiveAsync(Queue.QueueName, _cancellation);
             if (Queue.Role == EndpointRole.Application)
             {
                 Logger.LogInformation("{Count} messages in queue {QueueName} at listening start up time",
@@ -252,10 +274,10 @@ public async Task CreateAsync()
 
         var mapper = Queue.BuildMapper(_runtime);
 
-        _consumer = new WorkerQueueMessageConsumer(Channel!, _receiver, Logger, this, mapper, Address, _cancellation);
+        _consumer = new WorkerQueueMessageConsumer(channel, _receiver, Logger, this, mapper, Address, _cancellation);
 
-        await Channel!.BasicQosAsync(0, Queue.PreFetchCount, false, _cancellation);
-        await Channel.BasicConsumeAsync(Queue.QueueName, false,
+        await channel.BasicQosAsync(0, Queue.PreFetchCount, false, _cancellation);
+        await channel.BasicConsumeAsync(Queue.QueueName, false,
             _transport.ConnectionFactory?.ClientProvidedName ?? _runtime.Options.ServiceName, Queue.ConsumerArguments, _consumer,
             _runtime.Cancellation);