Fix RabbitMQ agent permanently latched Disconnected after a channel-only shutdown (#3171)

[jeremydmiller]

Fixes #3171.

Problem

A RabbitMQ channel can shut down while the underlying connection stays alive — and without surfacing a callback exception (a channel-only shutdown). When that happens the agent latches into AgentState.Disconnected and never recovers until the process restarts:

• RabbitMqSender.SendAsync throws InvalidOperationException: The RabbitMQ agent for ... is disconnected on every send; the durable outbox retries forever.
• Listeners silently stop consuming.

Root cause (confirmed against main): the two channel-failure handlers were asymmetric.

• HandleChannelExceptionAsync (callback exception) tears down, nulls Channel, rebuilds, sets Connected. ✅
• HandleChannelShutdownAsync (channel shutdown) only set State = Disconnected and returned without nulling Channel or rebuilding.

EnsureInitiated() then short-circuited on the stale, non-null channel (if (Channel is not null) return;), so it never rebuilt. The only recovery paths both miss this case: callback-exception recovery needs an exception, and ConnectionMonitor/ReconnectedAsync needs the whole connection to drop.

Fix

• EnsureInitiated() now treats a non-null-but-closed channel as dead: it rebuilds whenever Channel is missing or not IsOpen, tearing down the stale channel under the lock first. The under-lock IsOpen re-check avoids racing the existing callback-exception rebuild and double-restarts, and the Application-initiator close is left alone. This heals senders lazily on the next SendAsync — no fire-and-forget tasks.
• A new HandleUnexpectedChannelShutdown() hook is invoked from HandleChannelShutdownAsync for non-Application shutdowns. Listeners sit blocked on the broker and never call EnsureInitiated() again on their own, so RabbitMqListener overrides it to eagerly rebuild and re-consume via ReconnectedAsync(). It's guarded on ConnectionIsLive so a full connection drop is left to the ConnectionMonitor recovery path (no double restart).
• RabbitMqListener.ReconnectedAsync() is hardened to tolerate cancelling consumers on an already-dead channel before tearing down and rebuilding.

Tests

Bug_3171_channel_only_shutdown_recovery covers both paths against a live broker:

• sender_recovers_after_a_channel_only_shutdown — closes only the channel, asserts the agent latches Disconnected with a stale closed channel, then recovers on EnsureInitiated() and a real send succeeds.
• listener_resumes_consuming_after_an_unexpected_channel_shutdown — forces a broker-initiated (non-Application) channel close and asserts the listener rebuilds on a fresh channel and resumes consuming end-to-end.

Both tests fail without the fix (the sender throws immediately; the listener times out at 30s, never recovering) and pass with it. The surrounding RabbitMQ lifecycle suite (end_to_end, exclusive_listeners, rate_limiting_end_to_end) passes; the only other failures observed were the pre-existing shared-broker flakes (fan_out/direct_exchange) which pass in isolation.

🤖 Generated with Claude Code