Handling outstanding Publisher Confirms and deliveryTag reset during channel recovery in RabbitMQ.Client 7.x (.NET 8)

[alisasantos]

🏷️ Discussion Type

Question

Body

We are upgrading our event publisher service to the new RabbitMQ.Client 7.x SDK in .NET 8, moving away from synchronous IModel usages to the new async IChannel interface.

Under high throughput, we implement publisher confirms using CreateChannelOptions with publisher confirmations enabled. We track pending confirmations using a concurrent dictionary of TaskCompletionSource indexed by the message's deliveryTag, resolving them inside IChannel.BasicAcksAsync and IChannel.BasicNacksAsync event handlers.

However, when a network partition or broker restart triggers automatic recovery (AutomaticRecoveryEnabled = true), the deliveryTag counter resets on the new channel. This creates a race condition where:

1. Pending TaskCompletionSource instances for the old channel remain unhandled and leak memory.
2. If new messages are published immediately after recovery, duplicate deliveryTag values overlap with lingering pending tasks from the pre-recovery channel state.

What is the recommended design pattern in RabbitMQ.Client 7.x to safely fault or drain pending publisher confirm tasks upon connection/channel recovery without relying on internal channel state?

Guidelines

• I have read and understood this category's guidelines before making this post.

[MatheusWylliam]

To safely handle pending publisher confirms during connection or channel recovery in RabbitMQ.Client 7.x, you should avoid sharing confirmation state across channel recycles. Because deliveryTag sequence numbers reset to 1 on a new channel session, lingering TaskCompletionSource instances must be faulted immediately when the underlying channel closes.

Here is the recommended pattern using the async IChannel lifecycle:

1. Fault Pending Confirms on Channel Shutdown

Attach a handler to IChannel.ChannelShutdownAsync so that any unexpected disconnection drains and faults all unconfirmed tasks with a explicit exception.

public class ConfirmedPublisher
{
    private readonly ConcurrentDictionary<ulong, TaskCompletionSource<bool>> _pendingConfirms = new();
    private IChannel _channel;

    public async Task InitializeAsync(IConnection connection)
    {
        var options = new CreateChannelOptions(
            publisherConfirmationsEnabled: true,
            publisherConfirmationTrackingEnabled: true
        );

        _channel = await connection.CreateChannelAsync(options);
        _channel.ChannelShutdownAsync += OnChannelShutdownAsync;
        _channel.BasicAcksAsync += OnBasicAckAsync;
        _channel.BasicNacksAsync += OnBasicNackAsync;
    }

    private Task OnChannelShutdownAsync(object sender, ShutdownEventArgs args)
    {
        // Fault all lingering tasks for the current channel instance fast
        foreach (var key in _pendingConfirms.Keys)
        {
            if (_pendingConfirms.TryRemove(key, out var tcs))
            {
                tcs.TrySetException(new ChannelClosedException(args));
            }
        }
        return Task.CompletedTask;
    }

    private Task OnBasicAckAsync(object sender, BasicAckEventArgs args)
    {
        HandleAckNack(args.DeliveryTag, args.Multiple, success: true);
        return Task.CompletedTask;
    }

    private Task OnBasicNackAsync(object sender, BasicNackEventArgs args)
    {
        HandleAckNack(args.DeliveryTag, args.Multiple, success: false);
        return Task.CompletedTask;
    }

    private void HandleAckNack(ulong sequenceNumber, bool multiple, bool success)
    {
        if (multiple)
        {
            var keysToClean = _pendingConfirms.Keys.Where(k => k <= sequenceNumber).ToList();
            foreach (var key in keysToClean)
            {
                if (_pendingConfirms.TryRemove(key, out var tcs))
                {
                    if (success) tcs.TrySetResult(true);
                    else tcs.TrySetException(new Exception("Message NACKed by broker"));
                }
            }
        }
        else
        {
            if (_pendingConfirms.TryRemove(sequenceNumber, out var tcs))
            {
                if (success) tcs.TrySetResult(true);
                else tcs.TrySetException(new Exception("Message NACKed by broker"));
            }
        }
    }
}