Re-consume on a successful eager RabbitMQ channel restart + pin the ConnectionMonitor tracking invariant (GH-3391)

[jeremydmiller]

Fixes #3391.

Two follow-ups from the review of #3370 (the RabbitMQ listener ghosting fix). One is a real behavioral bug; the other is test-only. They're separated below.

---

Part 1 (behavioral fix) — a successful eager restart of a listener never re-consumed

RabbitMqChannelAgent.HandleChannelExceptionAsync handled a channel callback exception by running teardownChannel() + startNewChannel(). That gives you a fresh channel and nothing else — no re-declare, no BasicConsumeAsync.

The #3171/#3187 shutdown push-heal does not cover this, because teardownChannel() unsubscribes the shutdown handler before closing the channel. So a callback-exception-only failure (the channel itself survives) left the listener in exactly the state the issue describes:

an open channel with ZERO consumers while State = Connected

A silently dead listener on an apparently healthy process. Nothing pulls from the queue and nothing reports a problem.

Reproduced

The new listener_still_consumes_after_a_successful_eager_restart test fails on the pre-fix code precisely as predicted — after the eager restart the channel is open, the agent reports Connected, and the broker reports 0 consumers on the queue:

Shouldly.ShouldAssertException : consumers
    should be
1u
    but was
0u

Additional Info:
    The restarted listener should be consuming from the queue again

The fix

The eager restart is now a protected virtual restartAfterCallbackExceptionAsync():

• RabbitMqChannelAgent keeps the existing sender-shaped behavior as the base implementation — a fresh channel is all a sender needs, and RabbitMqSender is the only other subclass. The failure path is unchanged: an exception still leaves the agent tracked and Disconnected for connection recovery to rebuild (that's Fix RabbitMQ listener ghosting after broker restart on channel callback exception #3370's fix).
• RabbitMqListener overrides it to defer to its existing ReconnectedAsync(), which already does the correct work — cancel any surviving consumer, tear the channel down, re-declare, re-consume. No duplicate consume path was added; ReconnectedAsync() is the right authority and every other listener-rebuild trigger already routes through it.
• RabbitMqListener.ReconnectedAsync() is now serialized behind a _reconnectLock. It has three triggers (connection recovery, unexpected channel-only shutdown, and now callback exception), and callback exceptions arrive in bursts — that's the shape reported in Fix RabbitMQ listener ghosting after broker restart on channel callback exception #3370 (N × "Callback error in Rabbit Mq agent"). Without the gate, two concurrent rebuilds can race onto the same channel and leave duplicate consumers behind.

Part 2 (test-only) — pinning the ConnectionMonitor tracking invariant

The invariant: an agent must remain tracked by ConnectionMonitor after a callback-exception-triggered restart, whether that restart succeeds or fails. Only tracked agents are iterated by connectionOnRecoverySucceededAsync, so an agent that falls out of _agents is never rebuilt on reconnect.

Before #3370, _monitor.Remove(this) had no compensating re-Track() anywhere — so even a successful eager restart left the agent permanently untracked and one connection drop away from ghosting. #3370 fixed it; nothing pinned it.

• New internal ConnectionMonitor.TrackedAgents accessor over _agents (returns a snapshot under the existing lock).
• New agent_stays_tracked_by_the_connection_monitor_across_a_callback_exception_restart test.

This part changes no runtime behavior.

How the tests drive it

Both tests provoke a genuine channel callback exception against the live broker rather than reaching in and invoking the handler: subscribe a throwing handler to BasicReturnAsync, then publish a mandatory message to a routing key that resolves to nothing. The broker returns it, the handler throws, and RabbitMQ.Client raises CallbackExceptionAsync on the channel — while leaving the channel open, which is exactly the scenario the push-heal misses.

Following the Bug_3171_channel_only_shutdown_recovery pattern: internal-state assertions against a live broker, poll-based waits, unique queue names, no shared fan_out/direct_exchange state.

Test results

Wolverine.RabbitMQ.Tests, full suite, live broker: 473 passed / 477, 13m16s. The 4 failures are pre-existing shared-broker flakes unrelated to this change — use_fan_out_exchange, multi_tenancy_through_virtual_hosts.send_message_to_a_specific_tenant, and Bug_1594_ReplayDeadLetterQueue (both modes). All 4 pass in isolation with this change applied, and the vhost test also passes on unmodified main only when run alone — i.e. ordering/shared-state, not a regression.

Full solution builds clean: dotnet build wolverine.slnx -c Release → 0 warnings, 0 errors.

#3137 assessment — leaving the circuit-breaking tests quarantined

CircuitBreakingTests passes 24/24 (3m59s) on this branch, but I am not un-quarantining it, for two reasons:

1. This fix doesn't touch the implicated code. RabbitMQ circuit-breaking tests quarantined in dedicated workflow pending transport-level fix #3137 localizes the remaining flakiness to the interplay between broker back-pressure and Wolverine's BufferedReceiver drain / Restarter timing (buffered_not_parallelized, buffered_and_parallel). This PR changes the channel callback-exception restart path — a different subsystem. There's no mechanism by which it would fix a pause/drain timing race.
2. One green run is not evidence against an intermittent flake. That's precisely the failure mode RabbitMQ circuit-breaking tests quarantined in dedicated workflow pending transport-level fix #3137 exists to track, and the issue is explicit that un-quarantining should wait on a root cause.

Un-quarantining here would be guesswork of exactly the kind #3137 warns against, so #3137 stays open and circuit-breaking.yml stays informational.

🤖 Generated with Claude Code