feat: add publish resilience gate for transient RabbitMQ connection drops

[niemyjski]

Summary

Fixes publish resilience during transient RabbitMQ connection drops. Previously, PublishAsync would fail immediately when the broker connection was momentarily lost (e.g., network blip, broker restart). Now publishes suspend transparently during recovery and resume automatically.

Problem

When the RabbitMQ connection drops, the .NET client's auto-recovery kicks in (default: 5s NetworkRecoveryInterval). During this window, any PublishAsync call would throw AlreadyClosedException or similar failures. The resilience policy's retry budget (3 attempts, 1s+2s backoff = ~3s total) was often exhausted before recovery completed (~5s), causing unnecessary message loss.

Solution: Recovery Gate Pattern

An AsyncManualResetEvent ("gate") suspends publishes during connection recovery and releases them when the connection is restored. The gate is checked inside the resilience policy's retry loop, meaning:

• Each retry attempt re-evaluates gate status (connection flaps don't burn the retry budget)
• Disposal cancellation (from base class DisposedCancellationToken) cleanly unblocks waiting publishers
• The IsOpen check on the channel provides defense-in-depth for edge cases

Architecture

PublishMessageAsync
  -> _resiliencePolicy.ExecuteAsync (retry loop)
       -> Check gate (wait if connection recovering)
       -> Acquire lock
       -> Verify channel.IsOpen
       -> BasicPublishAsync

Key Design Decisions

Decision	Rationale
Gate inside retry loop	Prevents instant retry-budget exhaustion during connection flaps
Gate opened after connection close on disposal	CleanupAsync closes connections first (nulling channel), then opens gate; waiters resume into null-channel check and fail cleanly. Base class DisposedCancellationToken also fires, cancelling linked tokens used by WaitAsync
Single _isPublisherBlocked check (inside lock)	Outer check was stale/racy; inner lock-protected check is authoritative
PublishRecoveryTimeout option (default: 10s)	Configurable per-publisher; zero disables waiting (fail-fast for legacy behavior)
IsOpen check per attempt	Defense against race between recovery signal and channel ready state

Configuration

var bus = new RabbitMQMessageBus(o => o
    .ConnectionString("amqp://localhost")
    .PublishRecoveryTimeout(TimeSpan.FromSeconds(15))); // wait up to 15s for recovery

Set to TimeSpan.Zero for pre-fix behavior (immediate failure on connection drop).

Changes

• RabbitMQMessageBus.cs: Added AsyncManualResetEvent gate, restructured PublishMessageAsync with gate inside retry loop, removed redundant outer _isPublisherBlocked check, leveraged disposal cancellation
• RabbitMQMessageBusOptions.cs: Added PublishRecoveryTimeout option with builder and validation
• Properties/AssemblyInfo.cs: Added InternalsVisibleTo for test project (simulation methods)
• RabbitMqPublishResilienceTests.cs: 7 tests covering: timeout-disabled, recovery wait, timeout expiry, cancellation, healthy publish, disposal, recovery-error

Test Plan

• All tests pass (dotnet test)
• Build with 0 warnings, 0 errors
• Timeout=zero fails immediately (after retry exhaustion)
• Gate blocks during recovery, unblocks on success
• Configurable timeout throws MessageBusException on expiry
• Caller cancellation respected during gate wait
• Disposal cancels waiting publishers cleanly (OperationCanceledException)
• Recovery errors keep gate closed (no premature unblock)

[Copilot]

Pull request overview

This PR addresses transient RabbitMQ connection drops causing PublishAsync to fail before auto-recovery completes by introducing a publish “readiness gate” that pauses publishing during recovery (with a configurable timeout), then resumes once recovery succeeds.

Changes:

• Added an AsyncManualResetEvent gate in RabbitMQMessageBus to suspend publishes during connection recovery and time out with a MessageBusException.
• Introduced PublishRecoveryTimeout option (+ builder method) to configure/disable the recovery wait behavior.
• Added resilience-focused tests and InternalsVisibleTo to enable test-only simulation hooks.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.

File	Description
tests/Foundatio.RabbitMQ.Tests/Messaging/RabbitMqPublishResilienceTests.cs	Adds tests for publish behavior during simulated recovery/timeout/cancellation/disposal scenarios.
src/Foundatio.RabbitMQ/Properties/AssemblyInfo.cs	Grants test assembly access to internals used for recovery simulations.
src/Foundatio.RabbitMQ/Messaging/RabbitMQMessageBusOptions.cs	Adds PublishRecoveryTimeout option and fluent builder configuration.
src/Foundatio.RabbitMQ/Messaging/RabbitMQMessageBus.cs	Implements publish recovery gate, wires it into connection shutdown/recovery events, and adds simulation helpers.

Comments suppressed due to low confidence (2)

src/Foundatio.RabbitMQ/Messaging/RabbitMQMessageBus.cs:563

• This warning always says publishes will "wait up to {Timeout}" when the connection drops, but when PublishRecoveryTimeout is TimeSpan.Zero the gate is disabled and publishes will not wait. Consider adjusting the log message (or branching) to reflect the configured behavior (wait vs fail-fast) to avoid misleading operational output.

        if (e.Initiator != ShutdownInitiator.Application)
        {
            _publisherReady.Reset();
            _logger.LogWarning("Publisher connection lost (Reply Code: {ReplyCode}, Reason: {ReplyText}). Publishes will wait up to {Timeout:g} for recovery.",
                e.ReplyCode, e.ReplyText, _options.PublishRecoveryTimeout);
        }

tests/Foundatio.RabbitMQ.Tests/Messaging/RabbitMqPublishResilienceTests.cs:152

• The name PublishAsync_RecoveryError_StaysBlocked suggests a recovery-error event is simulated, but the test never triggers ConnectionRecoveryError (it only leaves the gate closed until the timeout). Consider either renaming this test to reflect what it actually verifies (timeout while recovery never completes) or adding a simulation that invokes the recovery-error path and asserts the gate remains closed until RecoverySucceeded.

    [Fact]
    public async Task PublishAsync_RecoveryError_StaysBlocked()
    {
        await using var messageBus = new RabbitMQMessageBus(o => o
            .ConnectionString(ConnectionString)
            .LoggerFactory(Log)
            .PublishRecoveryTimeout(TimeSpan.FromMilliseconds(500)));

        await messageBus.PublishAsync(new SimpleMessageA { Data = "warmup" }, cancellationToken: TestCancellationToken);

        messageBus.SimulatePublisherConnectionLost();

        var publishTask = messageBus.PublishAsync(new SimpleMessageA { Data = "should stay blocked" },
            cancellationToken: new CancellationTokenSource(TimeSpan.FromSeconds(5)).Token);

        await Task.Delay(100, TestCancellationToken);
        Assert.False(publishTask.IsCompleted);

        await Task.Delay(100, TestCancellationToken);
        Assert.False(publishTask.IsCompleted);

        var ex = await Assert.ThrowsAsync<MessageBusException>(() => publishTask);
        Assert.Contains("connection recovery did not complete within", ex.Message);

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (1)

src/Foundatio.RabbitMQ/Messaging/RabbitMQMessageBus.cs:875

• In ParseHostEndpoint, when the string contains a colon but the port portion is not a valid int (e.g., "host:abc"), the fallback currently creates an endpoint with HostName set to the entire trimmed input (including ":abc"). This yields an invalid host and breaks failover parsing. Consider falling back to the parsed hostname with defaultPort (similar to the successful parse path), rather than using the original trimmed string.

            : new AmqpTcpEndpoint(trimmed, defaultPort);
    }

    private void RegisterPublisherConnectionEventHandlers()
    {
        if (_publisherConnection is null)
            throw new MessageBusException("Publisher connection must be initialized before registering event handlers.");

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.