diff --git a/docs/guide/messaging/transports/rabbitmq/performance.md b/docs/guide/messaging/transports/rabbitmq/performance.md index d67d0e892..8a64f03e7 100644 --- a/docs/guide/messaging/transports/rabbitmq/performance.md +++ b/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 diff --git a/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/Bugs/Bug_3706_dead_letter_paths_settle_the_original_delivery.cs b/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/Bugs/Bug_3706_dead_letter_paths_settle_the_original_delivery.cs new file mode 100644 index 000000000..5ff653801 --- /dev/null +++ b/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; + +/// +/// GH-3706: every ack used to go out as BasicAckAsync(tag, multiple: true), 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 +/// ConsumerDispatchConcurrency 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. +/// +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 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() + .Options.Transports.GetOrCreate(); + } + + private static async Task 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"); + } + + /// + /// is the path that was never settling. + /// It posts an enriched copy to the dead letter queue and used to stop there. + /// + [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); + } + + /// + /// Native mode settles through RabbitMqChannelCallback.moveToErrorQueueAsync, 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. + /// + [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); + } +} diff --git a/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqListener.cs b/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqListener.cs index 622490d30..96b354b6a 100644 --- a/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqListener.cs +++ b/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 _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(); } } diff --git a/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/WorkerQueueMessageConsumer.cs b/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/WorkerQueueMessageConsumer.cs index 8af19f734..fd29a6e10 100644 --- a/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/WorkerQueueMessageConsumer.cs +++ b/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; } }