feat: Allow waiting for prefetched messages on drain

[benjamin-alexander-simplisafe]

Adds an option to wait for prefetched messages in rabbitmq to be passed to the receiver before continuing with the shutdown process. Disabled by default to maintain parity in existing code, especially with the default 100 message prefetch (where this could be harmful).

There is a caviat where this won't necessarily handle parallel dispatch cleanly, as the cancel is handled in parallel with real messages, which has been documented.

[Copilot]

Pull request overview

Adds an opt-in RabbitMQ listener behavior to wait for prefetched messages already sitting in the RabbitMQ client’s internal dispatcher buffer to be delivered to the consumer during shutdown (“drain”), reducing “silent” redeliveries when a channel is closed while messages are prefetched but not yet handed to the receiver.

Changes:

• Introduces a per-listener configuration toggle (DrainWaitForPrefetch) and fluent config API to enable it.
• Tracks client-dispatched basic.cancel-ok in the RabbitMQ consumer to support “wait for dispatcher to drain” semantics.
• Updates listener shutdown to conditionally use BasicCancelAsync(..., noWait: false) and await the cancel-ok dispatch signal.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.

File	Description
src/Transports/RabbitMQ/Wolverine.RabbitMQ/RabbitMqListenerConfiguration.cs	Adds fluent configuration method to enable waiting for prefetched-message drain during shutdown.
src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/WorkerQueueMessageConsumer.cs	Adds a completion signal that fires when basic.cancel-ok is dispatched by the client.
src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqQueue.cs	Adds queue-level boolean option (DrainWaitForPrefetch) with documentation.
src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqListener.cs	Implements conditional no-wait cancel + optional wait-for-cancel-ok during stop.

Suppressed comments (1)

src/Transports/RabbitMQ/Wolverine.RabbitMQ/RabbitMqListenerConfiguration.cs:121

• The XML doc comment for ConsumerDispatchConcurrency is malformed (missing the opening
  tag), which can break generated docs and may trigger XML doc warnings/errors.

    ///     Override the RabbitMQ client's consumer dispatch concurrency for just this endpoint's
    ///     listening channels. This governs how many deliveries the client hands to the consumer
    ///     at once: at the default of 1, an <c>Inline</c> listener runs strictly one message at a
    ///     time through the handler no matter how MaxDegreeOfParallelism is set. Raising it is the
    ///     per-endpoint alternative to the transport-wide
    ///     <c>ConfigureChannelCreation(o =&gt; o.ConsumerDispatchConcurrency = n)</c>. See GH-3492.
    /// </summary>

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.

Suppressed comments (2)

src/Transports/RabbitMQ/Wolverine.RabbitMQ/RabbitMqListenerConfiguration.cs:119

• The XML doc comment for ConsumerDispatchConcurrency is missing its opening
  tag, which makes the documentation malformed (the block ends with
  but never starts). Add the missing opening tag so tooling generates correct docs.

    ///     Override the RabbitMQ client's consumer dispatch concurrency for just this endpoint's
    ///     listening channels. This governs how many deliveries the client hands to the consumer
    ///     at once: at the default of 1, an <c>Inline</c> listener runs strictly one message at a
    ///     time through the handler no matter how MaxDegreeOfParallelism is set. Raising it is the
    ///     per-endpoint alternative to the transport-wide

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

• The new DrainWaitForPrefetch shutdown behavior is subtle (timing-sensitive and dependent on RabbitMQ dispatch semantics), but there is no integration test coverage for it. Adding a focused RabbitMQ test around StopAndDrainAsync with prefetched messages would help prevent regressions.

            using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
            try
            {
                foreach (var consumerTag in consumer.ConsumerTags)
                {
                    await channel.BasicCancelAsync(consumerTag, false, cts.Token);
                }

                await consumer.CancelOkReceived.WaitAsync(cts.Token);

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.

Suppressed comments (2)

src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/drain_wait_for_prefetch.cs:126

• This test waits exactly 30 seconds for StopAndDrainAsync, but the implementation bounds the new drain wait by DurabilitySettings.DrainTimeout (default 30s) and StopAndDrain can legitimately take that long (or longer with additional drain steps). Using an equal timeout makes this test more likely to be flaky under load/slow CI.

        await stopTask.WaitAsync(30.Seconds(), TestContext.Current.CancellationToken);

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

• XML doc comment has a grammatical error (“so once all are in the prefetch backlog has drained”) and is hard to parse. This is public-facing documentation for a shutdown guarantee, so it should read cleanly.

    /// Wait until the broker has replied cancel-ok for <paramref name="count"/> cancelled consumer
    /// tags. Each cancel-ok is dispatched by the client only after every prefetched delivery ahead of
    /// it in the FIFO dispatcher queue has been processed, so once all are in the prefetch backlog has
    /// drained (to the batching channel, in durable micro-batching mode).

[jeremydmiller]

@benjamin-alexander-simplisafe Are you ready for me to review and maybe get this in today? There's yet another release brewing up this morning