PublishAsync silently fails during connection recovery due to retry/recovery timing mismatch

[niemyjski]

Problem

When a RabbitMQ connection drops while PublishAsync is in-flight (or called during the recovery window), the publish fails permanently even though the connection recovers successfully seconds later.

What happens today:

T+0.0s  Connection drops
T+0.0s  Attempt 1: BasicPublishAsync → AlreadyClosedException
T+1.0s  Attempt 2: (still recovering) → AlreadyClosedException
T+3.0s  Attempt 3: (still recovering) → AlreadyClosedException
T+3.0s  ALL RETRIES EXHAUSTED → MessageBusException propagates to caller
T+5.0s  Recovery completes → channel is healthy again (too late)

What should happen:

T+0.0s  Connection drops
T+0.0s  PublishAsync detects recovery in progress → suspends (awaits gate)
T+5.0s  Recovery completes → gate opens → publish succeeds transparently

Why it matters: Any code calling PublishAsync during a routine network blip (NAT timeout, broker rolling restart, brief network partition) gets a permanent failure for what is actually a transient ~5s disruption. The library should bridge this gap transparently.

Root Cause

It's NOT a "dead channel reference"

_publisherChannel holds an AutorecoveringChannel -- a wrapper from RabbitMQ .NET 7.x. Calling channel.BasicPublishAsync() delegates to InnerChannel which reads _innerChannel at call time. After recovery, the wrapper swaps its inner channel to a fresh one (source). The same reference becomes usable again post-recovery. Moving the capture inside the retry loop has zero effect.

It IS a timing mismatch

• Recovery time: ~5s (default NetworkRecoveryInterval)
• Retry budget: ~3s (3 attempts, exponential from 1s)
• Retries exhaust before recovery completes

Increasing retry count is inadequate because:

• Wastes CPU retrying against a known-dead connection
• Recovery time is variable
• Still timing-sensitive (just a bigger guess)

Why other transports don't have this

• Redis: StackExchange.Redis multiplexer blocks/queues internally during reconnect
• Azure Service Bus: SDK has built-in retry + reconnection; exceptions are terminal by the time they reach us
• Kafka: Producer has internal retry/buffer with its own broker reconnect
• RabbitMQ .NET client: Throws AlreadyClosedException immediately during recovery -- no internal waiting, no buffering

Proposed Solution: Gate Pattern with Configurable Timeout

Industry standard. MassTransit's documented expected behavior: "All calls to Publish should hang waiting for RabbitMQ to come online." EasyNetQ enters a "wait for connection" loop. RabbitMQ team: "Applications should wait for recovery to fully complete through recovery event notifications."

How it works

Single AsyncManualResetEvent (already in Foundatio.AsyncEx):

• Starts open (publishes flow through immediately)
• Closes when connection drops unexpectedly
• Opens when recovery succeeds (or fails -- so waiters get a clean error)
• PublishMessageAsync awaits the gate before attempting the publish

Behavior from the caller's perspective

Scenario	Today	After fix
Normal publish (connection healthy)	Succeeds immediately	Succeeds immediately (gate is open, zero overhead)
Publish during 5s recovery	Fails with MessageBusException after ~3s of useless retries	Suspends ~5s, then succeeds transparently
Publish during permanent failure	Fails after ~3s	Fails after timeout (configurable, default 10s)
Publish with CancellationToken during recovery	Fails after ~3s	Respects token -- caller controls their own timeout
Graceful shutdown (application-initiated)	Fails immediately	Fails immediately (gate stays open for app-initiated shutdown)

Configuration

/// <summary>
/// Maximum time PublishAsync will wait for connection recovery before failing.
/// During a transient connection drop, publishes suspend (rather than failing immediately)
/// and resume automatically when the connection recovers.
/// Set to TimeSpan.Zero to disable (fail immediately on connection drop, like pre-fix behavior).
/// Default: 10 seconds (covers one full NetworkRecoveryInterval cycle with margin).
/// </summary>
public TimeSpan PublishRecoveryTimeout { get; set; } = TimeSpan.FromSeconds(10);

Edge Cases

Edge case	Behavior
Graceful shutdown (Initiator == Application)	Gate stays open; publish fails immediately with "channel was closed"
Recovery never succeeds	Gate opens on recovery error; callers time out cleanly or get error on next attempt
Multiple rapid drops	Reset() is idempotent; gate stays closed until a Set()
Race: recovery completes between gate and BasicPublishAsync	Fine -- channel is healthy, publish succeeds
Race: drop AFTER gate check, BEFORE BasicPublishAsync	AlreadyClosedException → resilience policy retries → fails after 3 attempts. Acceptable: sub-millisecond race window.
Disposal during wait	cancellationToken is linked to DisposedCancellationToken in base class → OperationCanceledException
Gate + lock ordering	Gate is OUTSIDE lock. Multiple callers wait concurrently (no lock contention during wait). After gate opens, they serialize through lock (correct for channel thread safety).
PublishRecoveryTimeout = TimeSpan.Zero	Disables the gate entirely. Pre-fix behavior (fail immediately). Opt-out for callers who prefer fail-fast.

What This Does NOT Fix

1. Messages in-flight at moment of disconnection -- lost without publisher confirms, possibly duplicated with confirms. Fundamental RabbitMQ limitation.
2. Application code that ignores exceptions -- if the gate times out, PublishAsync still throws. Non-critical publishes should be wrapped in try/catch by the caller.
3. Broker resource alarms -- handled separately by existing _isPublisherBlocked mechanism.
4. Repeated rapid disconnections beyond the timeout -- if recovery takes >10s (or whatever timeout is configured), publish still fails. This is by design -- it's no longer "transient."