Fix RabbitMQ listener ghosting after broker restart on channel callback exception

[kconfesor]

Problem

A RabbitMQ listener can be left permanently ghosted (0 consumers on a healthy process) after a broker restart, when a channel callback exception coincides with the broker being unavailable — e.g. a broker/node roll that hits a listener while it has prefetched, unacked messages in flight. Messages then pile up undelivered until the process is restarted.

Idle listeners recover fine; only a listener that raises a callback exception at the moment the broker becomes unreachable is affected, so the blast radius is variable (one roll may ghost 1 listener, another several) and it reads as intermittent.

Root cause

RabbitMqChannelAgent.HandleChannelExceptionAsync:

Task.Run(async () =>
{
    await Locker.WaitAsync();
    try
    {
        _monitor.Remove(this);          // (1) agent removed from the ConnectionMonitor
        try { await teardownChannel(); }
        catch (Exception e) { Logger.LogError(e, "Error when trying to tear down a blocked channel"); }

        Channel = null;
        await startNewChannel();         // (2) throws if the broker is unreachable
        State = AgentState.Connected;
        Logger.LogInformation("Restarted the Rabbit MQ channel");
    }
    finally
    {
        Locker.Release();               // (3) no catch: exception escapes, agent NOT re-added
    }
});

1. A channel callback exception fires (broker dropping mid-I/O) and _monitor.Remove(this) removes the agent from ConnectionMonitor._agents.
2. startNewChannel() throws because the broker is now unreachable.
3. The exception escapes the fire-and-forget Task.Run (there is only a finally, no catch), so the agent is not re-added and State is left Disconnected.
4. When the client's automatic recovery reconnects, ConnectionMonitor.connectionOnRecoverySucceededAsync rebuilds every tracked agent — but this one is no longer in _agents, so it is skipped and never rebuilt.

ConnectionMonitor.Track() is only ever called from the RabbitMqChannelAgent constructor, so once Remove() runs there is no path that re-adds the agent — the removal is unsafe in any failure case, not just this one.

Fix

Do not de-register the agent on a failed eager restart. Keep it tracked (marked Disconnected) so the connection-level recovery path rebuilds it, and observe the restart exception instead of letting it escape:

• Removed the _monitor.Remove(this) call.
• Wrapped the restart in try/catch; on failure the agent stays registered and Disconnected, so RecoverySucceededAsync → ReconnectedAsync (StopAsync + teardownChannel + CreateAsync) cleanly rebuilds it once the connection returns. StopAsync cancels any partial consumer first, so there is no double-consumer risk.

Reproduction

With AutomaticRecoveryEnabled = true and TopologyRecoveryEnabled = true:

1. A durable-inbox listener whose handler does channel I/O (ack/nack/reschedule).
2. Publish several persistent messages so the consumer has prefetched, unacked messages.
3. Kill the broker while those messages are in flight, then bring it back.

Before the fix, at the instant of the kill the affected listener logs N × Callback error in Rabbit Mq agent. Attempting to restart the channel and zero Restarted the Rabbit MQ channel (every startNewChannel() threw); after recovery every other listener logs Opened a new channel …, but the affected queue ends with 0 consumers and messages stuck, while RabbitMQ connection is recovered successfully is logged (the recovery loop completed — the agent was simply absent from it). Rolling the broker with no in-flight messages recovers 100% of listeners every time.

After the fix, the same scenario logs Could not eagerly restart the Rabbit MQ channel; leaving it for connection recovery and the listener is rebuilt by the recovery path — 0 ghosts across repeated runs where the callback path fired.

[jeremydmiller]

Thorough review done against the #3171/#3187 recovery work — this composes cleanly with both orderings (recovery firing before or after the failed eager restart), and you actually undersold the bug: _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. Double-consumer risk checked: the eager restart never issues BasicConsume, and ReconnectedAsync cancels consumer tags before re-consuming, so removing the de-registration is safe.

Verified locally: build clean + both Bug_3171 regression tests green against a live broker with this diff applied.

Two non-blocking follow-ups we'll track separately:

1. A cheap regression test for the tracking invariant (agent remains tracked after a callback-exception restart) — needs an internal accessor on ConnectionMonitor._agents; the Bug_3171 test file shows the pattern.
2. Pre-existing gap spotted in passing: a successful eager restart of a listener only runs startNewChannel() — no re-declare/re-consume — so a callback-exception-only failure can leave an open channel with zero consumers. Not this PR's problem.

Merging when CI completes. Thanks for the excellent diagnosis and minimal diff.