Ack only the completed delivery, not every lower tag (GH-3706)

docs/guide/messaging/transports/rabbitmq/performance.md

@@ -42,6 +42,18 @@ strict one-at-a-time ordering on that endpoint, which is the whole point — if
 use `PartitionProcessingByGroupId` or sharded queues instead of a single serialized consumer.
 :::
 
+## Acknowledgements are per message
+
+Wolverine acks each delivery individually — `basic.ack` with `multiple: false`. It never uses a
+cumulative ack, which would tell the broker "and every lower delivery tag on this channel too".
+
+That matters as soon as completions can finish out of delivery order, which is the normal case
+with `ConsumerDispatchConcurrency` above 1: a cumulative ack on the message that happens to finish
+first would also acknowledge deliveries whose handlers are still running, and a crash at that
+moment loses them silently. Per-message acks cost nothing to make up for it — `basic.ack` is a
+fire-and-forget frame rather than an RPC, so there is no round trip to amortize, and batching them
+was measured at roughly **10% slower** and rejected (GH-3492).
+
 ## Choosing the endpoint mode
 
 - **Inline** (default): ack after successful handling. Safest, slowest per listener; scale with

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

@@ -0,0 +1,132 @@
+using JasperFx.Core;
+using JasperFx.Resources;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using RabbitMQ.Client;
+using Shouldly;
+using Wolverine.RabbitMQ.Internal;
+using Wolverine.Runtime;
+using Wolverine.Tracking;
+using Xunit;
+
+namespace Wolverine.RabbitMQ.Tests.Bugs;
+
+/// <summary>
+/// GH-3706: every ack used to go out as <c>BasicAckAsync(tag, multiple: true)</c>, which tells the broker
+/// "and every lower delivery tag on this channel too". That is only correct when completions happen in
+/// delivery order, and they do not — out-of-order completion already exists today with
+/// <c>ConsumerDispatchConcurrency</c> above 1 — so acking tag N silently swept up deliveries whose handlers
+/// were still running. A crash at that moment is silent message loss.
+///
+/// The cumulative sweep was load-bearing, which is why flipping it to per-message in isolation during the
+/// GH-3492 perf wave leaked a message into the quorum queues and was reverted: two dead-letter paths posted
+/// a copy elsewhere and never settled the original delivery at all, and nothing but a later cumulative ack
+/// reclaimed them.
+///
+/// These tests pin the settle. A delivery that has gone through a dead-letter path must leave the source
+/// queue empty even after the channel closes — an unsettled delivery is requeued by the broker on channel
+/// close, which is exactly how the leak manifested.
+/// </summary>
+public class Bug_3706_dead_letter_paths_settle_the_original_delivery : IAsyncLifetime
+{
+    private readonly string _queueName = "gh3706_" + Guid.NewGuid().ToString("N")[..8];
+    private string DeadLetterQueueName => _queueName + "_DLQ";
+    private IHost _host = null!;
+
+    public ValueTask InitializeAsync() => ValueTask.CompletedTask;
+
+    public async ValueTask DisposeAsync()
+    {
+        if (_host != null!)
+        {
+            await _host.StopAsync(TestContext.Current.CancellationToken);
+            await _host.TeardownResources();
+            _host.Dispose();
+        }
+    }
+
+    private async Task<RabbitMqTransport> bootstrapAsync(DeadLetterQueueMode mode)
+    {
+        _host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.UseRabbitMq().AutoProvision().AutoPurgeOnStartup();
+
+                opts.PublishAllMessages().ToRabbitQueue(_queueName);
+                opts.ListenToRabbitQueue(_queueName)
+                    .DeadLetterQueueing(new DeadLetterQueue(DeadLetterQueueName, mode));
+
+                opts.LocalRoutingConventionDisabled = true;
+            }).StartAsync();
+
+        return _host.Services.GetRequiredService<IWolverineRuntime>()
+            .Options.Transports.GetOrCreate<RabbitMqTransport>();
+    }
+
+    private static async Task<uint> queueDepthAsync(string queueName)
+    {
+        await using var conn = await new ConnectionFactory { HostName = "localhost" }.CreateConnectionAsync();
+        await using var channel = await conn.CreateChannelAsync();
+        var result = await channel.QueueDeclarePassiveAsync(queueName);
+        return result.MessageCount;
+    }
+
+    private static async Task waitForDeadLetteredAsync(RabbitMqQueue deadLetterQueue)
+    {
+        var deadline = DateTimeOffset.UtcNow.Add(30.Seconds());
+        while (DateTimeOffset.UtcNow < deadline)
+        {
+            if (await deadLetterQueue.QueuedCountAsync() > 0) return;
+            await Task.Delay(250.Milliseconds());
+        }
+
+        throw new TimeoutException("Never got a message into the dead letter queue");
+    }
+
+    /// <summary>
+    /// <see cref="RabbitMqInteropFriendlyCallback.MoveToErrorsAsync"/> is the path that was never settling.
+    /// It posts an enriched copy to the dead letter queue and used to stop there.
+    /// </summary>
+    [Fact]
+    public async Task interop_friendly_dead_lettering_leaves_no_unacked_delivery_behind()
+    {
+        var transport = await bootstrapAsync(DeadLetterQueueMode.InteropFriendly);
+
+        await _host.TrackActivity().DoNotAssertOnExceptionsDetected()
+            .PublishMessageAndWaitAsync(new AlwaysErrors());
+
+        await waitForDeadLetteredAsync(transport.Queues[DeadLetterQueueName]);
+
+        // Closing the channel is the whole point: the broker requeues anything still unacked on it. Before
+        // the fix, and with per-message acks, this came back as 1.
+        await _host.StopAsync(TestContext.Current.CancellationToken);
+
+        (await queueDepthAsync(_queueName)).ShouldBe(0u);
+
+        // ...and exactly one copy was dead lettered, not two. The interop callback sends its own enriched
+        // copy, so settling with a nack instead of an ack would have added a second via the DLX under
+        // UseEnhancedDeadLettering.
+        (await queueDepthAsync(DeadLetterQueueName)).ShouldBe(1u);
+    }
+
+    /// <summary>
+    /// Native mode settles through <c>RabbitMqChannelCallback.moveToErrorQueueAsync</c>, which always nacked
+    /// correctly. Here as the control: the behaviour that was already right has to stay right, including
+    /// that the broker's own dead-letter-exchange still routes exactly one copy.
+    /// </summary>
+    [Fact]
+    public async Task native_dead_lettering_still_leaves_no_unacked_delivery_behind()
+    {
+        var transport = await bootstrapAsync(DeadLetterQueueMode.Native);
+
+        await _host.TrackActivity().DoNotAssertOnExceptionsDetected()
+            .PublishMessageAndWaitAsync(new AlwaysErrors());
+
+        await waitForDeadLetteredAsync(transport.Queues[DeadLetterQueueName]);
+
+        await _host.StopAsync(TestContext.Current.CancellationToken);
+
+        (await queueDepthAsync(_queueName)).ShouldBe(0u);
+        (await queueDepthAsync(DeadLetterQueueName)).ShouldBe(1u);
+    }
+}

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

@@ -2,6 +2,7 @@
 using JasperFx.Core.Reflection;
 using Microsoft.Extensions.Logging;
 using RabbitMQ.Client;
+using RabbitMQ.Client.Exceptions;
 using Wolverine.Configuration;
 using Wolverine.Runtime;
 using Wolverine.Transports;
@@ -11,14 +12,20 @@ namespace Wolverine.RabbitMQ.Internal;
 
 internal class RabbitMqInteropFriendlyCallback : IChannelCallback, ISupportDeadLetterQueue
 {
+    // Matched without the closing quote deliberately -- the tag number sits INSIDE the quotes in the
+    // broker's message. See the same constant on RabbitMqChannelCallback.
+    private const string UnknownDeliveryTag = "PRECONDITION_FAILED - unknown delivery tag";
+
     private readonly IChannelCallback _inner;
     private readonly RetryBlock<Envelope> _sendBlock;
+    private readonly ILogger _logger;
 
 
     public RabbitMqInteropFriendlyCallback(RabbitMqTransport transport, RabbitMqQueue deadLetterQueue,
         IWolverineRuntime runtime)
     {
         _inner = transport.Callback!;
+        _logger = runtime.Logger;
         var sender = deadLetterQueue.ResolveSender(runtime);
 
         _sendBlock =
@@ -41,6 +48,40 @@ public async Task MoveToErrorsAsync(Envelope envelope, Exception exception)
     {
         DeadLetterQueueConstants.StampFailureMetadata(envelope, exception);
         await _sendBlock.PostAsync(envelope);
+
+        // GH-3706: settle the ORIGINAL delivery. Unlike its sibling
+        // RabbitMqChannelCallback.moveToErrorQueueAsync, this method posts a *copy* to the dead letter queue
+        // and stops -- it does not ack or nack the delivery it came from.
+        //
+        // On the normal error-handling path that is survivable, because MoveToErrorQueue.ExecuteAsync calls
+        // lifecycle.CompleteAsync() immediately after MoveToDeadLetterQueueAsync, and so does
+        // NoHandlerContinuation. This is the belt to those braces: settle it here so the delivery is not
+        // relying on a *later* step in a different layer for the only thing that ever reclaims it. The
+        // Acknowledged flag makes the CompleteAsync that follows a no-op rather than a double settle.
+        //
+        // ACK, not nack. The sibling nacks with requeue: false precisely so the broker's own
+        // x-dead-letter-exchange routes the original -- that is native dead lettering, and it sends no copy
+        // of its own. This callback has already taken responsibility by sending an enriched copy, so a nack
+        // here would either be discarded (InteropFriendly mode removes the DLX argument from the queue) or,
+        // under UseEnhancedDeadLettering against a Native-mode queue that still has a DLX, put a SECOND copy
+        // in the dead letter queue. Acking settles it exactly once in both shapes.
+        if (envelope is RabbitMqEnvelope e && !e.Acknowledged && e.RabbitMqListener.CanSettle(e))
+        {
+            try
+            {
+                // Marked before the ack so a later CompleteAsync is a no-op rather than a double settle.
+                e.Acknowledged = true;
+                e.HasBeenAcked = true;
+                await e.DeliveredOn.BasicAckAsync(e.DeliveryTag, false, CancellationToken.None);
+            }
+            catch (AlreadyClosedException closed) when (closed.Message.Contains(UnknownDeliveryTag))
+            {
+                // Terminal -- the tag's channel is gone and no retry can succeed. The copy is already on its
+                // 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");
+            }
+        }
     }
 
     public bool NativeDeadLetterQueueEnabled => true;
@@ -312,9 +353,10 @@ public override string ToString()
     ///
     /// A bare null check is NOT sufficient here, and using one would silently lose messages. Delivery
     /// tags are scoped to a single channel and restart at 1 on every new one, so a stale tag replayed
-    /// against a rebuilt channel addresses a completely different delivery -- and because CompleteAsync
-    /// acks with multiple: true, that would ack every LOWER tag on the new channel as well, including
-    /// messages still in the handler pipeline.
+    /// against a rebuilt channel settles a completely different, unrelated delivery. The guard predates
+    /// GH-3706 and its original rationale was worse still -- acks were cumulative then, so one replayed
+    /// stale tag also swept every lower tag on the new channel. Per-message acks narrow the blast radius
+    /// to one wrong message; they do not make settling on a replaced channel correct, so the guard stays.
     ///
     /// When this returns false the correct behavior is to do nothing at all. The broker never saw an
     /// ack, so it requeues the delivery on channel close and redelivers it; the durable inbox then
@@ -354,16 +396,23 @@ public Task CompleteAsync(RabbitMqEnvelope envelope)
             return Task.CompletedTask;
         }
 
-        // NOTE: multiple: true also acks every LOWER delivery tag on this channel. That is wrong
-        // under a concurrent listener -- it acks messages still in the handler pipeline -- but it
-        // is currently load-bearing, because it sweeps up deliveries that some settle paths never
-        // acknowledge at all. Changing it in isolation leaks those deliveries. Tracked separately;
-        // see GH-3492 and the ack-semantics branch.
+        // GH-3706: multiple MUST stay false. `multiple: true` tells the broker "and every lower delivery
+        // tag on this channel too", which is only correct when completions happen in delivery order.
+        // They do not: out-of-order completion already happens today with ConsumerDispatchConcurrency
+        // above 1. Acking tag N cumulatively swept up every lower unacked tag, including deliveries whose
+        // handlers were still running -- and a crash at that moment is silent message loss.
+        //
+        // The GH-3492 attempt at this flip was reverted because it leaked a message into the quorum queues:
+        // the cumulative sweep was covering for settle paths that never acknowledged a delivery on their
+        // own. Those are settled at the source now -- WorkerQueueMessageConsumer's un-mappable-message
+        // branch, which dead lettered and returned without touching the delivery at all, and
+        // RabbitMqInteropFriendlyCallback.MoveToErrorsAsync, which posts a copy and leaves the settle to a
+        // later CompleteAsync in another layer.
         //
         // Coalescing these into cumulative acks behind a batching window was measured and REJECTED:
         // basic.ack is a fire-and-forget frame, not an RPC, so batching saves no round trips while
         // the extra channel hops cost ~10% of max inline throughput. See the ledger in
         // RABBITMQ-PERF-DEEP-DIVE-PLAN.md.
-        return envelope.DeliveredOn.BasicAckAsync(envelope.DeliveryTag, true, _cancellation).AsTask();
+        return envelope.DeliveredOn.BasicAckAsync(envelope.DeliveryTag, false, _cancellation).AsTask();
     }
 }

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

@@ -121,6 +121,26 @@ public override async Task HandleBasicDeliverAsync(string consumerTag, ulong del
                 if (_workerQueue is ISupportDeadLetterQueue dlq)
                 {
                     await dlq.MoveToErrorsAsync(envelope, e);
+
+                    // GH-3706: settle it. This is the receiver's dead letter path -- the message has been
+                    // written to Wolverine's dead letter storage (or handed to a dead letter sender), so
+                    // Wolverine owns it now and the broker's copy has to go. Until acks became per-message
+                    // this delivery was left permanently unacked and only got reclaimed by the cumulative
+                    // sweep from some later BasicAckAsync(tag, multiple: true) on the same channel. Ack
+                    // rather than nack, for the same reason as RabbitMqInteropFriendlyCallback: a nack with
+                    // requeue: false would ALSO route the original through the queue's
+                    // x-dead-letter-exchange and leave two copies dead lettered.
+                    try
+                    {
+                        await Channel.BasicAckAsync(deliveryTag, multiple: false, _cancellation);
+                    }
+                    catch (Exception ackEx)
+                    {
+                        _logger.LogError(ackEx,
+                            "Failed to ack a dead lettered, un-mappable RabbitMQ message {MessageId}",
+                            properties.MessageId);
+                    }
+
                     return;
                 }
             }