GH-3950: quiesce a Rabbit channel that the broker has already closed

src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/stale_delivery_tag_settling.cs

@@ -74,6 +74,65 @@ public async Task a_stale_delivery_tag_does_not_lose_the_message()
 
         await host.StopAsync(TestContext.Current.CancellationToken);
     }
+
+    /// <summary>
+    /// GH-3950. A rejected settle now proactively rebuilds the listener's channel rather than leaving
+    /// deliveries streaming into a teardown. This asserts the listener is still CONSUMING afterwards.
+    /// </summary>
+    /// <remarks>
+    /// This is the guard for the hazard that proactive rebuild introduces rather than for the bug it
+    /// mitigates. #3391 is the precedent: a rebuild that only swaps the channel leaves a listener sitting
+    /// on an open channel with ZERO consumers while still reporting Connected — silently dead, and no
+    /// existing assertion catches it because the poisoned message itself is redelivered by the broker
+    /// regardless. Publishing a fresh batch AFTER the rebuild is what distinguishes "recovered" from
+    /// "quietly stopped listening".
+    /// </remarks>
+    [Fact]
+    public async Task the_listener_keeps_consuming_after_a_rejected_settle_rebuilds_the_channel()
+    {
+        var queueName = RabbitTesting.NextQueueName();
+
+        StaleTagHandler.Reset(poisonAt: 2);
+
+        using var host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.UseRabbitMq().AutoProvision().AutoPurgeOnStartup();
+                opts.PublishAllMessages().ToRabbitQueue(queueName).SendInline();
+                opts.ListenToRabbitQueue(queueName).ProcessInline();
+            })
+            .StartAsync(cancellationToken: TestContext.Current.CancellationToken);
+
+        var bus = host.MessageBus();
+
+        var first = Enumerable.Range(0, 5).Select(_ => Guid.NewGuid()).ToArray();
+        foreach (var id in first)
+        {
+            await bus.PublishAsync(new StaleTagMessage(id));
+        }
+
+        await StaleTagHandler.WaitForAll(first.Length, TimeSpan.FromSeconds(60));
+
+        // Let the rejection, the proactive quiesce and the rebuild actually happen -- all of which land
+        // after the last message of the first batch has been handled.
+        await StaleTagHandler.WaitForRedelivery(first.Length, TimeSpan.FromSeconds(30));
+
+        // The assertion that matters: brand new messages published after the rebuild still arrive.
+        var second = Enumerable.Range(0, 5).Select(_ => Guid.NewGuid()).ToArray();
+        foreach (var id in second)
+        {
+            await bus.PublishAsync(new StaleTagMessage(id));
+        }
+
+        await StaleTagHandler.WaitForIds(second, TimeSpan.FromSeconds(60));
+
+        foreach (var id in second)
+        {
+            StaleTagHandler.HandledIds.ShouldContain(id);
+        }
+
+        await host.StopAsync(TestContext.Current.CancellationToken);
+    }
 }
 
 public record StaleTagMessage(Guid Id);
@@ -139,6 +198,28 @@ public static async Task WaitForRedelivery(int firstPassCount, TimeSpan timeout)
             $"covering what it claims to.");
     }
 
+    /// <summary>
+    /// Polls for a specific set of ids rather than a count. The _source TCS is completed exactly once,
+    /// when the FIRST batch reaches _expected, so a second WaitForAll in the same test returns
+    /// immediately on the already-completed task and asserts against nothing.
+    /// </summary>
+    public static async Task WaitForIds(IEnumerable<Guid> ids, TimeSpan timeout)
+    {
+        var wanted = ids.ToArray();
+        var deadline = DateTimeOffset.UtcNow + timeout;
+
+        while (DateTimeOffset.UtcNow < deadline)
+        {
+            if (wanted.All(_ids.ContainsKey)) return;
+            await Task.Delay(250);
+        }
+
+        var missing = wanted.Where(x => !_ids.ContainsKey(x)).ToArray();
+        throw new TimeoutException(
+            $"{missing.Length} of {wanted.Length} messages published AFTER the channel rebuild were never " +
+            $"handled within {timeout}. The listener stopped consuming rather than recovering.");
+    }
+
     public static void Handle(StaleTagMessage message, Envelope envelope)
     {
         // Poison exactly one delivery. The counter only reaches _poisonAt once, so a redelivery of

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

@@ -42,6 +42,9 @@ internal RabbitMqChannelCallback(ILogger logger, CancellationToken cancellationT
                 if (isUnknownDeliveryTag(exception))
                 {
                     logger.LogInformation("Encountered an unknown delivery tag, discarding the envelope");
+
+                    // GH-3950: the broker has already closed that channel. Stop feeding it.
+                    e.RabbitMqListener.QuiesceAfterRejectedSettle(e);
                 }
             }
         }, logger, cancellationToken);
@@ -129,6 +132,9 @@ private async Task moveToErrorQueueAsync(RabbitMqEnvelope envelope, Cancellation
             if (isUnknownDeliveryTag(exception))
             {
                 Logger.LogInformation("Encountered an unknown delivery tag, discarding the envelope");
+
+                // GH-3950: the broker has already closed that channel. Stop feeding it.
+                envelope.RabbitMqListener.QuiesceAfterRejectedSettle(envelope);
                 return;
             }
 

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

@@ -80,6 +80,9 @@ public async Task MoveToErrorsAsync(Envelope envelope, Exception exception)
                 // way to the dead letter queue, so there is nothing left to do.
                 _logger.LogInformation(
                     "Encountered an unknown delivery tag while settling a dead lettered message, discarding the envelope");
+
+                // GH-3950: the broker has already closed that channel. Stop feeding it.
+                e.RabbitMqListener.QuiesceAfterRejectedSettle(e);
             }
         }
     }
@@ -419,6 +422,68 @@ public override string ToString()
     /// `Channel!` threw a NullReferenceException mid-reconnect -- can never succeed on any later
     /// channel, and just burns the retry budget while the same message cycles round again.
     /// </summary>
+    // The channel generation we have already quiesced, so a burst of rejected settles on the same dead
+    // channel triggers exactly one rebuild rather than one per envelope.
+    private object? _quiescedChannel;
+
+    /// <summary>
+    /// GH-3950. Called when the broker rejects a settle with <c>PRECONDITION_FAILED - unknown delivery
+    /// tag</c>, which means the broker has ALREADY closed the channel the tag arrived on.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// The damage that motivates this is not the rejected tag itself, which Wolverine has always handled.
+    /// It is that RabbitMQ.Client races itself while a channel is being torn down with deliveries still in
+    /// flight on it: an inbound frame arrives for a channel number just removed from the session map,
+    /// <c>SessionManager.Lookup</c> does an indexer read and throws <c>KeyNotFoundException</c>, and the
+    /// client escalates that into a library-initiated close of the WHOLE connection (code=541) — every
+    /// listener and sender on it, not just this channel.
+    /// </para>
+    /// <para>
+    /// Wolverine cannot catch that; it is thrown on the client's own MainLoop after our ack has already
+    /// gone out. What we CAN do is stop feeding a channel we now know is dead: cancel its consumer, tear
+    /// it down and rebuild, rather than leaving deliveries streaming into a teardown. This narrows the
+    /// window for further frames on the dead channel and gets the listener back on a healthy one sooner.
+    /// It does not, and cannot, prevent the connection death that a single rejected tag has already set in
+    /// motion.
+    /// </para>
+    /// </remarks>
+    internal void QuiesceAfterRejectedSettle(RabbitMqEnvelope envelope)
+    {
+        var channel = envelope.DeliveredOn;
+        if (channel is null || IsDisposed || _cancellation.IsCancellationRequested)
+        {
+            return;
+        }
+
+        // One rebuild per channel generation. Exchange returns the PREVIOUS value, so the first caller for
+        // a given channel sees something else and proceeds; every later one sees its own channel back.
+        if (ReferenceEquals(Interlocked.Exchange(ref _quiescedChannel, channel), channel))
+        {
+            return;
+        }
+
+        Logger.LogWarning(
+            "A Rabbit MQ delivery tag was rejected as unknown at {Uri}, which means the broker has already closed that channel. Proactively rebuilding the listener's channel so that in flight deliveries are not left racing its teardown. See GH-3950.",
+            Address);
+
+        // Deliberately not awaited: this runs from a settle path (a RetryBlock, or the dead letter
+        // callback) and ReconnectedAsync takes _reconnectLock and does real broker work.
+        _ = Task.Run(async () =>
+        {
+            try
+            {
+                await ReconnectedAsync();
+            }
+            catch (Exception e)
+            {
+                Logger.LogError(e,
+                    "Error while proactively rebuilding the Rabbit MQ channel for {Uri} after a rejected delivery tag",
+                    Address);
+            }
+        });
+    }
+
     internal bool CanSettle(RabbitMqEnvelope envelope)
     {
         var channel = Channel;