fix: RMQ.Async producer moves confirm handlers to a replacement channel

[iancooper]

Symptom

The RMQ.Async RmqMessageProducer stopped raising publish confirmations once its channel was replaced. A channel gets replaced after an IOException reset, a broker-side channel close, or another gateway dropping the shared pooled connection. After that:

• OnMessagePublished never fired again, so outbox entries were never marked dispatched and were resent as duplicates;
• pending confirmations piled up, and dispose waited out the whole WaitForConfirmsTimeOutInMilliseconds.

Root cause

The producer attached OnPublishSucceeded/OnPublishFailed to BasicAcksAsync/BasicNacksAsync only when Channel was null before EnsureBrokerAsync. RmqMessageGateway.ConnectToBrokerAsync replaces a closed channel in place, and Channel only goes back to null on dispose. So a replacement channel never got the handlers.

The channel is created with publisher-confirmation tracking on, so publishes on the new channel still succeeded. Their pending entries were simply never settled.

Evidence:

• A code trace, with its v7 claims checked against the RabbitMQ.Client v7.2.2 source.
• A broker-backed regression test that times out waiting for the confirmation without the fix.

The full record is in bugfixes/0055-rmq-async-confirm-resubscribe/bugfix.md.

Fix

This follows #4567's fix for RMQ.Sync, adapted for v7:

• The producer records the channel its handlers are attached to (_confirmChannel) and compares it by reference on each send. When the channel has changed, it detaches the handlers from the old channel and attaches them to the new one.
• Dispose detaches from the tracked channel instead of whatever Channel currently is.
• v7 turns confirms on when it creates the channel, so no ConfirmSelect is needed.

Detaching after the IOException reset is safe in v7, so it is unchanged. Disposing an AutorecoveringConnection clears its channel list but does not dispose the channels, so removing handlers does not throw. That is unlike the v6 ObjectDisposedException fixed in #4567.

Tests

• New: MessagingGateway/Proactor/When_the_channel_is_replaced_should_confirm_messages_sent_on_the_new_channel.cs, run against a real broker.
• New doubles:
  • ConfirmCountingRmqChannel, a forwarding DispatchProxy over IChannel that counts confirm subscribers;
  • ConfirmCountingRmqProducer, which keeps every channel the producer opens.
• Paramore.Brighter.RMQ.Async.Tests with CI's filter: 257 passed, 0 failed, on both net9.0 and net10.0.
• Category=RMQNativeDelay: 103 passed, 0 failed.

Scope

Confirm first planned a wider fix: clear pending confirmations on a channel switch, guard the dispose wait, and lock the attach. Writing tests for those showed none of them can fail on a single-threaded send path once this fix is in. In tracking mode each send settles its own entry before BasicPublishAsync returns, and a new channel's tags overwrite older entries with the same tag.

The real remaining risk is concurrent sends on one producer. It shows up three ways: duplicate delivery tags, a failed send's finally removing another channel's entry, and a double attach. That is one design problem, recorded on #4577: #4577 (comment)

Also found during verification: #4581, where the RMQ.Sync and RMQ.Async dispatcher tests share queue names with different durability, so the two suites conflict when they run on one broker. This does not affect CI.

Fixes #4578

🤖 Generated with Claude Code

https://claude.ai/code/session_01SpjnqGWbARr3Z3ofGc6C2x

[iancooper]

@claude Let's review the following. Please pay attention to the following in the comments when scoping the review: "Confirm first planned a wider fix: clear pending confirmations on a channel switch, guard the dispose wait, and lock the attach. Writing tests for those showed none of them can fail on a single-threaded send path once this fix is in. In tracking mode each send settles its own entry before BasicPublishAsync returns, and a new channel's tags overwrite older entries with the same tag.

The real remaining risk is concurrent sends on one producer. It shows up three ways: duplicate delivery tags, a failed send's finally removing another channel's entry, and a double attach. That is one design problem, recorded on #4577: #4577 (comment)

Also found during verification: #4581, where the RMQ.Sync and RMQ.Async dispatcher tests share queue names with different durability, so the two suites conflict when they run on one broker. This does not affect CI."

[claude]

Claude finished @iancooper's task in 1m 32s —— View job

---

PR review: RMQ.Async confirm handlers move to replacement channel

• Read diff and scope note
• Review the RmqMessageProducer change against the gateway's channel lifecycle
• Review tests and test doubles
• Check the scope claims (single-threaded safety, deferral to RMQ producers: silent channel replacement leaves stale pending confirmations, so delivery tags collide #4577)
• Post review

Verdict: The fix is correct and small. It matches the root cause, and the regression test pins the behaviour against a real broker. I found one small consistency problem worth fixing (below) and a few nits. Nothing blocks the merge.

---

Scope check

I reviewed this against the scope you gave: a single-threaded send path, with concurrency left to #4577 and the test-queue clash left to #4581.

• "Each send settles its own entry before BasicPublishAsync returns": this holds. The channel is created with publisherConfirmationTrackingEnabled: true (RmqMessageGateway.cs:164-168). In v7 tracking mode, BasicPublishAsync awaits the broker confirm. The ack/nack event handlers run before that awaited task completes. So when PublishMessageAsync returns (RmqMessageProducer.cs:206), SettleConfirmationsAsync has already removed the entry. If the channel closes mid-publish, the publish throws and the finally removes the entry by tag (:247-248). On a single-threaded path, then, no stale entry from the old channel survives into the new channel's tag space. A multiple=true ack on the new channel therefore can't falsely settle an old message. Clearing pending entries on a switch and guarding the dispose wait would add code that no test can make fail, so leaving them out is right.
• Concurrency is correctly left out. ListenForConfirmsOnChannel runs outside _stateLock, so two concurrent sends that both see a new channel can each attach (a double attach). Reading the delivery tag (:203) separately from the publish is the duplicate-tag race. Both already happened before this PR on the first channel. They are the same design problem recorded on RMQ producers: silent channel replacement leaves stale pending confirmations, so delivery tags collide #4577, so they don't belong here.
• RMQ.Sync and RMQ.Async dispatcher tests share queue names with different durability, so the suites conflict on one broker #4581 doesn't apply. The new test uses a random routing key and queue name, so it can't hit the durability clash.

Finding: the IOException catch leaves _confirmChannel out of step

RmqMessageProducer.cs:233-239

var failedChannel = Channel;
await ResetConnectionToBrokerAsync(cancellationToken);
if (failedChannel is not null)
{
    failedChannel.BasicAcksAsync -= OnPublishSucceeded;
    failedChannel.BasicNacksAsync -= OnPublishFailed;
}

The new field's comment says _confirmChannel is "the channel our confirm handlers are attached to". After this block runs, that is false: the handlers have been removed from failedChannel, but _confirmChannel still points to it.

• Today the damage is limited. The reset disposes the pooled connection, so the channel reports IsClosed, the next send gets a new instance, and the reference check moves the handlers. Removing an absent handler from the old channel a second time does nothing.

• The risk is the early return. The new design depends on _confirmChannel being accurate, because ListenForConfirmsOnChannel returns early on ReferenceEquals. Suppose any path ever hands back the same IChannel instance after this catch: an AutorecoveringChannel recovering in place, a reset that fails with BrokerUnreachableException before the old connection is disposed, or a later refactor of the pool. Then the early return fires and the producer has no handlers attached. That is the original bug again, without a test to catch it.

• The fix is one line. Use the helper the PR already adds. It detaches from the tracked channel and clears the field, so the invariant holds on every path:

  await ResetConnectionToBrokerAsync(cancellationToken);
  DetachPublisherConfirmHandlers();
  throw new ChannelFailureException(...);

  The failedChannel capture and its comment can go too. _confirmChannel already records the right channel, which is the point of this change. Your v7 finding still holds: a detach after the reset doesn't throw.

Fix this →

Production code: what looks right

• Dispose detaches from the right channel. DetachPublisherConfirmHandlers now uses _confirmChannel instead of Channel (:537-544). Before, a dispose after a channel switch would have removed handlers from a channel that never had them.
• The dispose order is still right. Dispose waits for active sends, then waits for confirmations, and only then detaches (:260-262, :277-279). So late acks still reach the handlers during the wait.
• No ConfirmSelect is needed. That's correct for v7, where confirms are a CreateChannelOptions setting.

Tests and test doubles

• The regression test is well aimed. It fails without the fix: the second message's confirmation times out. It also asserts the subscriber counts on both channels, which covers the "detach from the old channel" half of the fix as well as the "attach to the new one" half.
• ConfirmCountingRmqProducer wraps every channel the gateway creates. Its is ConfirmCountingRmqChannel check means a channel is wrapped once, and replacement channels are still caught.

Nits, all optional:

• ConfirmCountingRmqChannel.Invoke changes a count before it forwards the call. If the forwarded call throws, the count is already wrong. That's harmless for this test, but incrementing after a successful method.Invoke would be more honest.
• The test constructor blocks on QueueFactory.CreateAsync().GetAwaiter().GetResult(). IAsyncLifetime would avoid that. Ignore this if blocking matches the neighbouring Proactor tests, since consistency matters more.
• Channel! in ListenForConfirmsOnChannel relies on a null check its caller makes. Passing the channel in as a parameter, e.g. ListenForConfirmsOn(Channel), would drop the ! and make the dependency explicit.

Housekeeping

• The bugfixes/0055-* artifacts are fine. .confirm-approved and .issue-number match the existing bugfix folders.
• I didn't run the tests. This job has no RabbitMQ broker, so I'm relying on the results in the PR description.