[Mocha] Fix RabbitMQ recovery, publishing and outbox dispatch by tobias-tengler · Pull Request #10543 · ChilliCream/graphql-platform

tobias-tengler · 2026-10-14T09:14:22Z

RabbitMQ consumers could stop receiving after a connection or channel loss, outbound sends could complete without broker confirmation, and the PostgreSQL outbox could lose or duplicate messages around a broker outage.

This PR reworks how the RabbitMQ transport connects, publishes and consumes, and how the outbox reacts to a failed dispatch.

RabbitMQ transport

Connections

One connect routine serves startup and recovery. Each attempt is limited to 30 seconds, so an endpoint that accepts TCP and never answers no longer blocks recovery.
Startup retries within the new ConnectTimeout setting and fails immediately on rejected credentials, access or topology.
The connection is kept during a broker memory or disk alarm. Publishes wait for their confirmation and are not sent again.

Publishing

A send or publish completes when the broker confirms the message. A publish interrupted by a connection loss is retried, which can produce duplicate deliveries.
Every publish uses a channel of its own from a pool, with at most 100 publishes in flight per transport. A message the broker rejects, or a publish that is cancelled, no longer affects other publishes.
New setting PublishTimeout (default 2 minutes, infinite allowed). A publish the broker does not confirm in time fails with the new TransportUnavailableException.
A message the broker can never accept, for example one above the size limit, fails with the broker's reason and is not retried.
A missing destination queue is detected, and its configured bindings and upstream exchanges are restored before the retry.

Consuming

Every consumer recovers on its own. Concurrency limits are preserved when interrupted handlers ignore cancellation.
Cancelling or stopping one consumer no longer closes the connection it shares with the others.
A consumer that cannot attach again because handlers of the lost channel still run is logged as an error and reported unhealthy after 30 seconds, not on every channel loss.
Requests wait for the local reply consumer before they are sent, and the dispatch time is part of the request deadline. Temporary reply queues survive short outages. Receive endpoints no longer wait for the reply endpoint before they consume.
During shutdown, application handlers drain before reply consumers stop. Cancellation ends the drain while channel cleanup continues.

Logging

The log events are renamed and each fact is logged once. The RabbitMQ transport docs list the events for connection loss, recovery, blocked connections, publish retries and consumer recovery.

PostgreSQL outbox

Concurrent dispatch through MaxConcurrentMessages. The default remains 1. Higher values use additional database connections and allow completion out of order. A failing dispatch does not affect the other workers.
A row is removed only after the transport accepted the message.
An unavailable transport, a dispatch that takes longer than DispatchTimeout (default 30 seconds) and a host stop release the row without counting an attempt.
A message the broker or the transport refuses counts as an attempt and backs off exponentially, up to MaxRetryDelay (default 10 minutes). After MaxAttempts (default 10) the row stays in the table and a Critical entry is logged.
A failed dispatch is logged with the outbox id, message id, message type and destination.

Core, hosting and other transports

Disposing the runtime without stopping it now cleans up temporary resources in the Postgres and Azure Service Bus transports.
A transport whose start fails stops what it already started.
The message bus health check reports disconnected brokers and blocked consumers with their reasons. AddMessageBus takes name, failureStatus, tags and timeout, with the tag ready and a timeout of 10 seconds by default.
The transport circuit breaker logs under its own category, Mocha.TransportCircuitBreakerMiddleware.
ExponentialBackoff.CalculateDelay applies jitter at the cap as well.

Breaking changes

RabbitMQDispatcher, RabbitMQConsumerManager and RabbitMQConnectionManagerBase are internal now, as are the Dispatcher and ConsumerManager properties of RabbitMQMessagingTransport. IRabbitMQConnectionProvider stays public.
A send or publish fails with TransportUnavailableException after PublishTimeout.
The names and texts of the RabbitMQ log events changed, and so did the reason texts of the health check.
AddMessageBus(IHealthChecksBuilder, ...) has new optional parameters, which is binary breaking. The check now carries the tag ready.

Known limitations

When the broker closes a whole connection for a frame it cannot read, for example a header above the frame size, every publish in flight on that connection fails once with the broker's reason.
A publish the broker rejects with basic.nack is not retried. The outbox counts it as an attempt.
ScheduledMessageWorker.StopAsync does not wait for a dispatch the transport already accepted, so a scheduled message can be sent again after a restart.

github-actions · 2026-10-14T09:22:56Z

Patch coverage

90.4% of changed lines covered (2147/2376)

File	Covered	Changed	Patch %
…/Connection/ChannelOpenBlockedException.cs	0	1	0.0% 🔴
…/Descriptors/RabbitMQQueueDescriptor.cs	0	1	0.0% 🔴
…/Mocha/src/Mocha/Consumers/Implementations/ReplyConsumer.cs	1	2	50.0% 🔴
src/Mocha/src/Mocha.Transport.RabbitMQ/ThrowHelper.cs	13	24	54.2% 🔴
…/Mocha/Middlewares/Consume/Retry/ConsumerRetryMiddleware.cs	11	15	73.3% 🔴
…/AzureServiceBusReceiveEndpoint.cs	14	18	77.8% 🔴
…/Connection/RabbitMQConsumerManager.cs	64	80	80.0% 🟡
…/Connection/RabbitMQDispatcher.Publishing.cs	147	181	81.2% 🟡
…/Receive/Redelivery/ReceiveRedeliveryMiddleware.cs	19	23	82.6% 🟡
…/Mocha.Transport.RabbitMQ/Connection/RabbitMQDispatcher.cs	148	174	85.1% 🟡
…/Connection/RabbitMQConnectionManager.cs	404	459	88.0% 🟡
…/src/Mocha.Transport.RabbitMQ/RabbitMQReceiveEndpoint.cs	15	17	88.2% 🟡
…/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQConsumer.cs	408	461	88.5% 🟡
…/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQQueue.cs	16	18	88.9% 🟡
…/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs	108	116	93.1% 🟡
…/Mocha.EntityFrameworkCore.Postgres/Outbox/OutboxProcessor.cs	166	172	96.5% 🟡
…/Middlewares/Receive/RabbitMQAcknowledgementMiddleware.cs	43	44	97.7% 🟡
…/Outbox/MessageBusOutboxWorker.cs	10	10	100.0% 🟢
…/Outbox/OutboxServiceCollectionExtensions.cs	17	17	100.0% 🟢
…/Outbox/PostgresMessageOutboxOptions.cs	3	3	100.0% 🟢

+35 more changed files; see the JSON below for uncovered lines.

Uncovered changed lines (JSON)

{
  "sha": "b0de3bc8eeafc9171ab9fbb54a1f17885c7d895d",
  "files": [
    { "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/ChannelOpenBlockedException.cs", "ranges": [[8, 8]] },
    { "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Descriptors/RabbitMQQueueDescriptor.cs", "ranges": [[159, 159]] },
    { "path": "src/Mocha/src/Mocha/Consumers/Implementations/ReplyConsumer.cs", "ranges": [[69, 69]] },
    { "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/ThrowHelper.cs", "ranges": [[9, 9], [15, 15], [18, 20], [38, 38], [50, 50], [53, 53], [56, 58]] },
    { "path": "src/Mocha/src/Mocha/Middlewares/Consume/Retry/ConsumerRetryMiddleware.cs", "ranges": [[46, 49]] },
    { "path": "src/Mocha/src/Mocha.Transport.AzureServiceBus/AzureServiceBusReceiveEndpoint.cs", "ranges": [[300, 303]] },
    { "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQConsumerManager.cs", "ranges": [[140, 141], [168, 168], [200, 200], [203, 207], [209, 209], [218, 218], [220, 220], [224, 224], [257, 257], [259, 259], [261, 261]] },
    { "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQDispatcher.Publishing.cs", "ranges": [[47, 48], [70, 71], [170, 178], [193, 203], [213, 222]] },
    { "path": "src/Mocha/src/Mocha/Middlewares/Receive/Redelivery/ReceiveRedeliveryMiddleware.cs", "ranges": [[53, 56]] },
    { "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQDispatcher.cs", "ranges": [[86, 86], [110, 111], [149, 153], [177, 180], [182, 183], [185, 185], [197, 197], [199, 201], [203, 203], [225, 226], [301, 303], [378, 378]] },
    { "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQConnectionManager.cs", "ranges": [[115, 115], [137, 140], [143, 144], [160, 160], [162, 163], [175, 176], [200, 201], [215, 217], [241, 243], [286, 286], [316, 316], [360, 361], [370, 370], [398, 400], [473, 474], [496, 497], [515, 515], [517, 518], [520, 522], [524, 524], [565, 565], [568, 568], [588, 589], [650, 651], [653, 653], [674, 674], [686, 687], [689, 689], [750, 751], [758, 760]] },
    { "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQReceiveEndpoint.cs", "ranges": [[70, 71]] },
    { "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQConsumer.cs", "ranges": [[102, 102], [269, 280], [348, 350], [460, 463], [478, 491], [563, 563], [565, 566], [682, 682], [684, 684], [702, 704], [750, 760]] },
    { "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQQueue.cs", "ranges": [[122, 123]] },
    { "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs", "ranges": [[80, 80], [158, 163], [166, 166]] },
    { "path": "src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/OutboxProcessor.cs", "ranges": [[130, 130], [132, 133], [138, 140]] },
    { "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Middlewares/Receive/RabbitMQAcknowledgementMiddleware.cs", "ranges": [[35, 35]] }
  ]
}

Project coverage: 59.7% (324482/543839 lines)

github-actions · 2026-10-16T15:42:15Z

✅ Nitro CLI E2E snapshots: no flow recordings changed vs main.

Copilot

🔵 Needs a closer look

Recovery, shared-channel confirmations, and shutdown draining require RabbitMQ/Postgres integration validation that cannot be performed in this dependency-free snapshot.

No open issues 🎉

What changed in this PR

This PR strengthens Mocha’s RabbitMQ delivery and recovery behavior and adds configurable concurrent dispatch to the Postgres outbox.

Changes:

Restores consumers after connection or channel loss, preserves concurrency limits, and keeps reply queues available during recovery and graceful shutdown.
Adds confirmed publishing, connection-loss retries, missing-queue detection, and startup timeout configuration.
Adds concurrent outbox workers, expands recovery and lifecycle tests, and documents delivery guarantees and operational limits.

File	Description
`website/content/docs/mocha/transports/rabbitmq.md`	Documents recovery, confirmations, shutdown, and queue lifetimes.
`website/content/docs/mocha/reliability.md`	Documents outbox concurrency and connection-pool considerations.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Topology/RabbitMQReplyQueueTests.cs`	Tests reply queue expiry and defaults.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Topology/RabbitMQBusDefaultsTests.cs`	Tests defaults for non-durable queues.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Routing/RabbitMQReceiveTopologyConventionTests.cs`	Updates generated reply queue expectations.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQTransportTests.cs`	Tests timeout propagation to both connections.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQDispatchProvisioningTests.cs`	Removes obsolete mocked dispatch tests.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Middlewares/RabbitMQAcknowledgementMiddlewareTests.cs`	Tests acknowledgement races and handler failures.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Helpers/TestBus.cs`	Bounds shutdown draining and cleans up failed startup.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Descriptors/RabbitMQDescriptorTests.cs`	Tests connection timeout validation.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Connection/RabbitMQConsumerManagerTests.cs`	Tests disposal after channel-close failures.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Connection/RabbitMQConsumerCapacityTests.cs`	Tests concurrency budgets across channel losses.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Connection/RabbitMQConsumerAttachmentIntegrationTests.cs`	Tests independent attachment, dependencies, and cancellation.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Connection/RabbitMQConnectionManagerBaseTests.cs`	Tests setup synchronization, timeouts, and recovery callbacks.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/UnifiedQueueBehaviorTests.cs`	Moves tests to the topology collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/TransportMiddlewareTests.cs`	Moves tests to the topology collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/TopologyRecoveryTests.cs`	Tests topology restoration after publisher reconnection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/TemporaryEndpointCleanupTests.cs`	Moves tests to the topology collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ShutdownTests.cs`	Tests draining, cancellation, requeueing, and reply availability.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/SendTests.cs`	Moves tests to the messaging collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/SchedulingTests.cs`	Moves tests to the messaging collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/RoutingKeyTests.cs`	Moves tests to the topology collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/RequestReplyTests.cs`	Updates reply lifetime and reconnection scenarios.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ReplyRecoveryTests.cs`	Tests request cancellation and replies during disconnection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ReplyQueueCleanupTests.cs`	Tests ownership-aware reply queue cleanup.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/PublishSubscribeTests.cs`	Moves tests to the messaging collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/PublishingReliabilityTests.cs`	Tests broker rejections and publishing failure isolation.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/PublishingConcurrencyTests.cs`	Tests concurrent sends and cancellation recovery.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/PublisherRecoveryTests.cs`	Tests publishing through blocking, disconnection, and restart.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/MissingQueueRecoveryTests.cs`	Updates queue and binding recovery scenarios.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/InboxTests.cs`	Moves tests to the messaging collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/FaultHandlingTests.cs`	Moves tests to the topology collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ExternalDestinationTests.cs`	Tests foreign destinations and missing-queue errors.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ExplicitTopologyTests.cs`	Moves tests to the topology collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ExclusiveQueueTests.cs`	Tests exclusive queue ownership and recovery.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ErrorQueueTests.cs`	Moves tests to the topology collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/EndpointMiddlewareTests.cs`	Moves tests to the topology collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/CustomHeaderTests.cs`	Moves tests to the topology collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ConsumerLifecycleTests.cs`	Tests endpoint restart and uncooperative handlers.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ConnectionStartupTests.cs`	Tests startup retries, failures, and connection names.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ConnectionRecoveryTests.cs`	Verifies recovery without test-side publish retries.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ConcurrencyTests.cs`	Updates recovery synchronization for concurrency tests.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ConcurrencyLimiterTests.cs`	Moves tests to the messaging collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/BusDefaultsIntegrationTests.cs`	Moves tests to the topology collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/BrokerRecoveryTests.cs`	Tests broker outages, credential changes, and resource cleanup.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/BatchingTests.cs`	Moves tests to the messaging collection.
`src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/AutoProvisionIntegrationTests.cs`	Moves tests to the topology collection.
`src/Mocha/test/Mocha.Tests/Transport/MessagingTransportLifecycleTests.cs`	Tests shutdown ordering and cleanup after failures.
`src/Mocha/test/Mocha.Tests/Threading/ExponentialBackoffTests.cs`	Tests retry delay bounds and randomization.
`src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxConcurrencyTests.cs`	Tests concurrent dispatch, idle workers, and claim release.
`src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/OutboxServiceRegistrationTests.cs`	Tests rejection of invalid concurrency settings.
`src/Mocha/src/Mocha/Transport/MessagingTransport.cs`	Drains application endpoints concurrently before replies.
`src/Mocha/src/Mocha/Runtime/MessagingRuntime.cs`	Cancels remaining shutdown work during disposal.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQTopologyProvisioner.cs`	Keeps exclusive queue provisioning on consumer connections.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQRoutingStrategy.cs`	Gives generated reply queues expiry-based lifetimes.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQReceiveTopology.cs`	Removes the receive topology wrapper.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQQueue.cs`	Adds expiry without auto-delete.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQMessagingTopology.cs`	Supports bypassing bus-level queue defaults.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/Configurations/RabbitMQQueueConfiguration.cs`	Adds internal control over default application.
`src/Mocha/src/Mocha.Transport.RabbitMQ/ThrowHelper.cs`	Centralizes setup, timeout, and returned-message exceptions.
`src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQReceiveEndpoint.cs`	Adds health reporting, reply dependencies, and guarded cleanup.
`src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessagingTransport.cs`	Applies startup timeouts and role-based connection names.
`src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs`	Retries connection losses and handles mandatory queue returns.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Mocha.Transport.RabbitMQ.csproj`	Updates RabbitMQ.Client to 7.2.2.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Middlewares/Receive/RabbitMQAcknowledgementMiddleware.cs`	Handles channel closure during acknowledgement or rejection.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Descriptors/RabbitMQQueueDescriptor.cs`	Qualifies the shared middleware configuration exception helper.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Descriptors/RabbitMQMessagingTransportDescriptor.cs`	Validates and configures startup connection timeouts.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Descriptors/IRabbitMQMessagingTransportDescriptor.cs`	Exposes the connection timeout API.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQConsumerManager.ChannelConsumer.cs`	Queues deliveries and tracks handler cancellation across loss.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Configurations/RabbitMQTransportConfiguration.cs`	Stores startup timeout and supports named connections.
`src/Mocha/src/Mocha.Transport.RabbitMQ/Configurations/RabbitMQDefaultQueueOptions.cs`	Skips incompatible defaults for non-durable queues.
`src/Mocha/src/Mocha.Threading/ExponentialBackoff.cs`	Adds reusable randomized retry delay calculation.
`src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/PostgresOutboxOptions.cs`	Exposes maximum concurrent outbox dispatches.
`src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/PostgresMessageOutboxOptions.cs`	Carries concurrency into processor configuration.
`src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/OutboxServiceCollectionExtensions.cs`	Validates and propagates outbox concurrency.
`src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/OutboxProcessor.cs`	Runs independent dispatch workers with coordinated cancellation.

_{Review effort: Balanced}

…hem away The transport now owns connection recovery instead of the client's automatic recovery, and re-attaches a consumer with backoff when the broker closes its channel or cancels it while the connection stays up. - Re-attach consumers whose channel was closed or whose subscription was cancelled, and declare the receive topology of every consumer before any consumer resumes on a connection. - Dispose lost connections and reconnect in the background with backoff. - Wait for the broker at startup for up to `ConnectTimeout`, two minutes by default, and fail at once on refused credentials, access or topology. - Publish with confirms, return messages sent to a missing queue, retry publishes on connection loss, and cap the number of rented publish channels. - Ack the redelivery of a message whose handler already completed without running the handler again. - Let reply queues expire instead of being deleted with their last consumer, and keep bus queue defaults away from them. - Declare exclusive queues only on the consumer connection. - Name connections by role and require RabbitMQ.Client 7.2.2.

Enforce connection timeouts, cancel consumer attachment during shutdown, and retry losses that arrive as recovery completes. Reuse consumer workers across reconnects, cancel handlers on channel loss, and expose blocked recovery health. Remove the completed-delivery cache and cover recovery, capacity limits, redelivery, and shutdown with regression tests.

Preserve compatible queue defaults and explicit argument overrides while skipping quorum and stream defaults for non-durable queues. Cover both queue type configuration forms with regression tests. Clarify that shutdown cancellation ends handler draining while channel cleanup continues independently.

Share publisher channels per endpoint and bound outstanding publish attempts while waiting for individual broker confirmations. Add configurable Postgres outbox concurrency with independent workers so blocked dispatches do not stall idle capacity. Preserve cancellation behavior with explicit loop termination conditions. Add concurrency and cancellation coverage and document the publishing and outbox settings.

Wait for consumer capacity before reattaching and for the local reply consumer before dispatching requests. Cancel pending connection waiters and drain connection setup during disposal, with idempotent consumer shutdown and forced-stop escalation. Keep request deadlines active until dispatch completes, including when a response or fault arrives before dispatch finishes. Apply explicit quorum replication defaults to generated reply queues so requests can recover through surviving cluster nodes. Respect broker defaults for unspecified queue types and delivery limits. Honor explicitly configured limits with RabbitMQ 3 compatibility, and preserve existing queues and their messages during provisioning. Cover broker process crashes, repeated restarts, quorum leader failover, connection failover, cancellation, disposal races, request deadlines, and finite or unlimited redelivery with regression tests. Validation: 223 targeted tests passed across RabbitMQ 3 and 4 and core request deadlines. Documentation formatting and rendered pages checked.

Add correlated recovery, retry, fault, publish-wait, and drain diagnostics. Preserve connection and channel identity across recovery, clarify health failures, and correct conversation tracing. Remove log-capture assertions and retain behavioral coverage for delivery, recovery, cancellation, and faults. Document scope export configuration.

Configure the cluster fixture to return disconnected-consumer deliveries after one second and verify the broker is running after each restart. Add coverage for an unacknowledged event held by a killed consumer node.

RabbitMQ transport - Replace the connection manager base class with one internal RabbitMQConnectionManager: one connect routine with a deadline per attempt, a ready signal, and a recovery that only the first loss report starts. A connection attempt against an endpoint that never answers now ends. - Publish on an exclusive channel lease from a pool per connection, with at most 100 channels. A channel is reused only after the broker gave a verdict on it, so a rejected or cancelled publish no longer affects other publishes. - Add the PublishTimeout transport setting (default 2 minutes). A send or publish that the broker does not confirm in time fails with the new TransportUnavailableException. - Keep the connection during a broker memory or disk alarm and do not send waiting messages again. - Run each consumer as one lifecycle loop. Cancelling or stopping one consumer no longer aborts the shared connection, and a consumer whose handlers block reports an error only after a grace period. - Remove the wait of receive endpoints for the reply endpoint. - Classify failures as transient or permanent. A message the broker can never accept is no longer retried without end. - Rename the log events and log one event per fact. - RabbitMQDispatcher, RabbitMQConsumerManager and RabbitMQConnectionManagerBase are no longer public. Outbox - Dispatch with independent workers. A failing dispatch no longer cancels the others, which sent an already sent message again. - Add DispatchTimeout, MaxAttempts and MaxRetryDelay. An unavailable transport does not use up attempts, and an exhausted message is logged as Critical with its id, type and destination. Core and other transports - Stopping with a cancelled token still cleans up in the Postgres and Azure Service Bus transports. - A failed transport start stops what already started. - The message bus health check takes a name, failure status, tags and a timeout, and uses the ready tag by default. - The transport circuit breaker logs under its own category. - ExponentialBackoff applies jitter at the cap as well. Tests and docs - Recovery tests prove the loss they recover from, with shared broker fault helpers and one timeout policy. Mock-based tests are replaced by tests against a broker. - Document recovery, the publish timeout, broker alarms, the outbox options and the health check.

github-actions Bot added the 🌶️ mocha label Oct 14, 2026

tobias-tengler force-pushed the tte/rabbitmq-consumer-reattach branch from 33654db to aee642f Compare October 19, 2026 07:36

github-actions Bot added the 📚 documentation This issue is about working on our documentation. label Oct 23, 2026

tobias-tengler force-pushed the tte/rabbitmq-consumer-reattach branch from 60e1ed8 to 420a0ea Compare October 27, 2026 07:48

tobias-tengler force-pushed the tte/rabbitmq-consumer-reattach branch 2 times, most recently from 0ec47a6 to c7833e6 Compare November 4, 2026 08:23

tobias-tengler changed the base branch from main to tte/mocha-rabbitmq-recreate-missing-queues November 4, 2026 08:23

tobias-tengler added this pull request to stack #10794 November 4, 2026 09:16

tobias-tengler changed the title ~~[Mocha] Fix RabbitMQ consumers not being re-attached after the broker takes them away~~ [Mocha] Fix RabbitMQ recovery and bound consumer concurrency Nov 4, 2026

tobias-tengler force-pushed the tte/rabbitmq-consumer-reattach branch 3 times, most recently from 2d19eb0 to ed13ac3 Compare November 5, 2026 15:16

tobias-tengler changed the title ~~[Mocha] Fix RabbitMQ recovery and bound consumer concurrency~~ [Mocha] Fix RabbitMQ recovery and message delivery Nov 7, 2026

tobias-tengler marked this pull request as ready for review November 7, 2026 16:36

Copilot AI balanced review requested due to automatic review settings November 7, 2026 16:36

Copilot started reviewing on behalf of tobias-tengler November 7, 2026 16:37

View session

Copilot AI reviewed Nov 7, 2026

View reviewed changes

tobias-tengler requested a review from PascalSenn November 9, 2026 08:50

tobias-tengler added 10 commits November 10, 2026 16:13

Fix RabbitMQ shutdown deleting externally managed reply queues

62b5794

Speed up RabbitMQ integration tests

82bfde8

Fix RabbitMQ recovery blocking unrelated consumers

83559ea

Fix RabbitMQ publisher channel ownership and reply shutdown ordering

6b963c3

Fix RabbitMQ publisher channel failure isolation

e99b284

Fix RabbitMQ channel capacity and consumer recovery races

88deb61

Fix RabbitMQ publishes racing connection setup during recovery

8161fc3

tobias-tengler added 17 commits November 10, 2026 16:13

Improve tests

10efd02

Fix RabbitMQ reply queue expiry and channel cleanup

ab83717

Fix RabbitMQ consumer recovery after channel open failures

437ec63

Fix RabbitMQ publisher recovery and preserve recovery callbacks

4a2d2ec

Fix RabbitMQ batch recovery test concurrency

95040e0

Fix runtime disposal and RabbitMQ recovery after reply endpoint restarts

299d5d1

Polish docs

6c5874f

Integrate into health check

dde2fa3

Fix flaky test

ea2cf36

Fix message bus readiness and RabbitMQ recovery diagnostics

20be281

Fix RabbitMQ failover test rejecting valid redeliveries

38a56c0

Fix RabbitMQ failover tests timing out on disconnected consumers

d7329ab

Configure the cluster fixture to return disconnected-consumer deliveries after one second and verify the broker is running after each restart. Add coverage for an unacknowledged event held by a killed consumer node.

Cleanup

7f3324a

Cleanup

d55287e

tobias-tengler force-pushed the tte/rabbitmq-consumer-reattach branch from 60eae49 to d55287e Compare November 10, 2026 15:14

tobias-tengler added 2 commits November 10, 2026 19:56

Cleanup

5b471a3

tobias-tengler changed the title ~~[Mocha] Fix RabbitMQ recovery and message delivery~~ [Mocha] Fix RabbitMQ recovery, publishing and outbox dispatch Nov 11, 2026

Navigation Menu

Repository navigation

[Mocha] Fix RabbitMQ recovery, publishing and outbox dispatch - #10543

[Mocha] Fix RabbitMQ recovery, publishing and outbox dispatch#10543
tobias-tengler wants to merge 29 commits into
tte/mocha-rabbitmq-recreate-missing-queuesfrom
tte/rabbitmq-consumer-reattach

tobias-tengler commented Oct 14, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Oct 14, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Oct 16, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

tobias-tengler commented Oct 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

RabbitMQ transport

Connections

Publishing

Consuming

Logging

PostgreSQL outbox

Core, hosting and other transports

Breaking changes

Known limitations

Uh oh!

github-actions Bot commented Oct 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Patch coverage

Uh oh!

github-actions Bot commented Oct 16, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🔵 Needs a closer look

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

tobias-tengler commented Oct 14, 2026 •

edited

Loading

github-actions Bot commented Oct 14, 2026 •

edited

Loading