Skip to main content
Table of Contents

Outbox + RabbitMQ Setup

Derived page. The behaviour described here is specified by the outbox-and-messaging, projections and sagas capabilities under openspec/specs/. Those specifications are the source; this page explains and illustrates them. Where the two disagree, the specification is right and this page is a bug.

Stratara.Outbox.RabbitMQ provides the IMessageBus implementation backed by a RabbitMQ broker. It uses publisher confirms + automatic reconnect + mandatory routing — failed-to-deliver messages are caught + retried from the outbox table.

Add the package

dotnet add package Stratara.Outbox.RabbitMQ

Configure

The bus reads its connection from one of two places, and there is no RabbitMq section — a section by that name is silently ignored. Either a connection string named rabbitmq:

// appsettings.json
{
  "ConnectionStrings": {
    "rabbitmq": "amqp://guest:guest@localhost:5672/"
  }
}

or four flat configuration keys, Kubernetes-style, which take precedence when RABBITMQ_HOST is set:

RABBITMQ_HOST=rabbitmq
RABBITMQ_PORT=5672            # default 5672
RABBITMQ_USERNAME=app         # outside Development both credentials are mandatory
RABBITMQ_PASSWORD=…           # in Development only, guest/guest is the fallback

Fail-fast outside Development (v3.4.0; v3.0.14+ for Production only): if RABBITMQ_USERNAME / RABBITMQ_PASSWORD are missing on any host that is not in Development, publishing throws InvalidOperationException naming the environment. The guest/guest fallback is Development-only — same pattern as the key-store guard.

Changed in 3.4.0. The check used to be IsProduction(), which recognises exactly one name: Staging, QA, UAT, Preview, anything self-named, and even Production-EU and prod all fell through to guest. RabbitMQ restricts guest to localhost by default, so a remote broker refused the connection anyway — but a broker in the same container or network running a default configuration accepted it. If you deliberately want the default account outside Development, set RABBITMQ_USERNAME=guest and RABBITMQ_PASSWORD=guest explicitly; the configuration is the opt-in.

Wire the worker

A typical worker host wires both the outbox-drainer and the command consumer:

var builder = Host.CreateApplicationBuilder(args);

builder.AddOutboxWorkerServices();   // drains outbox_entry → publishes to bus
builder.AddCommandWorkerServices();  // subscribes to bus → dispatches commands

builder.Services.AddCommandHandlersFromAssemblyContaining<MyCommandMarker>();

await builder.Build().RunAsync();

Routing model

Topic and subscription names come from the Messaging configuration section. Every one has a default, so a host that configures nothing still works:

Topic Default name Default subscription(s) Who publishes / consumes
Command command command-subscription Your outbox (and other apps) publish; the command worker consumes
HeavyCommand heavy-command heavy-command-subscription IHeavyCommand commands; the heavy-command worker consumes
EventBundle event-bundle event-bundle-subscription, event-bundle-saga-subscription The write-store publishes; projection and saga workers consume
Notification notifications — Consumer-defined notification fan-out

Override any of them by name:

{
  "Messaging": {
    "Topics": [
      {
        "Name": "Command",
        "Value": "myapp.command",
        "Subscriptions": [ { "Name": "CommandSubscription", "Value": "myapp.command.worker" } ]
      }
    ]
  }
}

Topics are fanout exchanges + per-subscription queues. Multiple worker hosts can scale out by sharing a queue — RabbitMQ does the work-stealing.

Establish subscriptions before the first publish

A queue exists from the moment something binds it, and RabbitMQ delivers only to queues that already exist. It does not hold a message for a queue that shows up later.

That is fine while one subscription binds late — the publish reaches nobody, the broker returns it, and the outbox retries. It is not fine when a topic carries more than one subscription, which event-bundle does: projections and sagas share it. If the saga queue is bound and the projection queue is not, the publish is confirmed, no outbox row is written, nothing retries, and nothing logs. The projection simply never sees that event.

The window is real. Two workers that start twenty seconds apart — separate deployments, a shared migration lock, a cold environment — are enough.

Close it from whichever process publishes first, before it publishes anything:

var ids = app.Services.GetRequiredService<IMessagingIdentifier>();
var bus = app.Services.GetRequiredService<IMessageBus>();

await bus.EnsureSubscriptionAsync(ids.EventBundleTopic, ids.EventBundleSubscription);
await bus.EnsureSubscriptionAsync(ids.EventBundleTopic, ids.EventBundleSagaSubscription);

EnsureSubscriptionAsync creates the queue and its binding without consuming from it. That is the whole difference from SubscribeAsync, which can only create a queue by also starting to read it — and therefore only when the worker that reads is ready. Anyone can establish anyone's subscription, so the publishing host can create a queue for a worker that has not started, or will not start for another hour.

Three things worth knowing:

  • Only the first time matters. Worker queues are durable and are not auto-deleted, so once they exist they survive restarts of the app and of the broker. This is a cold-environment problem: a new stage, a rebuilt host, a CI run against a fresh broker. (The queue behind a worker subscription is named <subscription>.v2 — see When a handler cannot take a message.)
  • Take the names from IMessagingIdentifier, not from strings. A subscription you forget is a subscription that keeps losing messages, silently, and a typo produces a second queue nobody reads.
  • Nothing calls this for you. The framework cannot know your start-up order or which processes publish, so it makes no attempt to guess. If your host runs every worker in one process and nothing publishes before the host is serving, you may not need it at all.

What it costs. Once a subscription is established, messages published to it are kept until something consumes them — where before they were dropped. If you establish a subscription for a worker you then never deploy, or later retire, its queue grows. That is the trade this makes deliberately: a fact kept somewhere you must clear is better than a fact silently gone.

So you hear about it: when EnsureSubscriptionAsync finds that the subscription already holds at least Messaging:UnconsumedSubscriptionWarningThreshold messages (default 10000) and no consumer is attached, it logs 108_114 at Warning, naming the subscription, the topic and the count. Every host that establishes the subscription says so at its next start. 0 turns the warning off; a negative value is refused when the host starts. Subscribing reports nothing — a backlog met by an attaching handler is a worker catching up.

The count covers the whole queue: a subscription established for several topics is reported once for each, with the same count.

The framework does not cap or expire the queue, and a broker policy is no way to do it either. The worker queue is declared with x-overflow: reject-publish, which a policy cannot override, so a length limit makes the broker refuse publications to the full queue — on a shared topic they go back through the outbox and are delivered again to every other subscription. A time-to-live only moves the growth to the dead-letter queue. When a worker is retired for good, stop establishing its subscription in the publishing hosts and then delete <subscription>.v2 and <subscription>.dead-letter; deleting the queue alone does not last, because the next EnsureSubscriptionAsync declares it again.

How many consumers a worker opens

The projection worker and the saga worker each open several consumers on their subscription — one per processor by default — and the broker deals consecutive bundles to different ones, so bundles are not processed in publication order. What the workers guarantee is that bundles about one aggregate are applied one at a time within the process; see Write a projection for what a handler may rely on and how it says "not yet".

Both counts are configuration:

{
  "Projections": { "DegreeOfParallelism": 1 },
  "Sagas": { "DegreeOfParallelism": 1 }
}

A value that is not a positive number — including leaving the key out — means one consumer per processor. 1 gives a worker that applies every bundle in the order the transport delivers it, at the cost of the parallelism; it is the right setting for a host that needs strict order, and the wrong one for a host that only needs the per-aggregate guarantee, which it already has.

Client subscriptions (default-*) are refused: they are declared exclusive and auto-deleting, so a queue established ahead of its consumer would be removed the moment the declaring channel closed. Establishing one would look like it worked and retain nothing.

Durable bundles: closing the commit-to-publish window

By default a save commits its events and then tries the bus; the bundle reaches the outbox table only if the bus refuses it. Between the commit and that publish there is a window in which the process may end — and a bundle lost there is lost for every subscription: the events are in the store, and no projection and no saga is told. Projections recover by a replay; a saga has no such repair.

Outbox:DurableBundles closes the window. The bundle is written to the outbox table in the same transaction that commits its events, published after the commit, and removed once the bus has accepted it. A process that ends anywhere after the commit leaves a row the outbox worker delivers.

{
  "Outbox": {
    "DurableBundles": true,       // default false: bus-first, window open
    "PollingIntervalSeconds": 5   // how soon the worker picks up what a crash left behind
  }
}

What it costs: one insert per save inside a transaction that already exists, and one delete per accepted bundle afterwards. Measured on PostgreSQL against a bus that accepts instantly — an upper bound on the relative cost, since a real broker slows both paths alike — the option takes 28–30 % off the appends per second at 1, 8 and 32 writers alike (2 650 → 1 860 at 32 writers on the reference machine). On a healthy bus the table stays near empty; a spike in it now means the bus is down, which it meant before too. Because the worker may drain a stored row before the save's own removal reaches it, a bundle can occasionally be delivered twice — which at-least-once already requires every handler to tolerate. Commands dispatched through ICommandOutboxDispatcher are unaffected — a command has no commit to be atomic with, and bus-first stays right for it. A host running sagas on the bus path, or one that cannot afford a replay, should switch it on; the outbox worker must run somewhere in the deployment for the rows a crash leaves to be delivered.

When a handler cannot take a message

A worker subscription is a quorum queue with a dead-letter queue beside it. A message whose handler throws is redelivered a bounded number of times and then moved to <subscription>.dead-letter, where you can inspect it and return it once the cause is fixed. Nothing is dropped by the framework. The bounds are the same on Azure Service Bus and come from one section:

{
  "MessageRetry": {
    "MaxDeliveryAttempts": 3,    // handler failures: delivered 3 times, then dead-lettered
    "MaxConflictRequeues": 100   // concurrency conflicts: requeued 100 times, then dead-lettered
  }
}

A concurrency conflict is a retry, not a failure — the next delivery usually sees the new aggregate version and passes — so it has the larger bound. Redelivery is immediate on both transports; there is no back-off. The framework reads the delivery count the broker stamps on every redelivery (x-acquired-count from RabbitMQ 4.3, x-delivery-count before) and decides itself; the queue's x-delivery-limit sits one above the larger bound as a backstop for redeliveries nobody asked for — a consumer that died mid-message — which is the only kind RabbitMQ 4.3+ counts against it. A message the broker dead-letters by that backstop is not in the framework's log or counter.

Changing the bounds on a running deployment. RabbitMQ compares x-delivery-limit whenever a queue is declared again and cannot change it on an existing quorum queue. A worker queue that already exists therefore keeps the limit it was first declared with: the subscription opens and uses it, and the bus logs a warning (108_112) naming the queue and the broker's reply. The new bounds still decide every redelivery the consumer asks for. From RabbitMQ 4.3 the old limit counts no other kind, so nothing more is needed. Before 4.3 it counts every redelivery, so a bound raised above the old limit is cut short by the broker; delete the drained queue and let the next start declare it again. The worker queue dead-letters through the default exchange straight to <subscription>.dead-letter with the at-least-once strategy, so the move itself cannot lose the message.

Every dead-lettering is logged (108_110) and counted on messaging.dead_lettered, tagged with messaging.topic, messaging.subscription and reason (conflict or failure). A rising conflict count is contention to look at or a bound to raise; a rising failure count is a handler to fix and a queue to replay.

Returning a message. Move it from <subscription>.dead-letter back to the worker queue with the shovel plugin or the management UI. It arrives as a fresh delivery, with its count starting over.

The queue names changed with this feature. A worker subscription named event-bundle-subscription now consumes event-bundle-subscription.v2; the old classic queue cannot be redeclared as a quorum queue, and the broker refuses the attempt. The rollout is two steps:

  1. Deploy. New consumers bind <subscription>.v2 to the same exchange; old consumers keep draining the old queue, and every bundle reaches both — which at-least-once already requires handlers to tolerate.
  2. When the old queue is empty and no old consumer is left, delete it. It is still bound to the exchange and would otherwise fill with every message forever.

Quorum queues exist since RabbitMQ 3.8; the framework's floor is 3.13 (see Prerequisites) and its integration tests run against 4.x.

Client subscriptions (default- prefix) are exclusive, auto-deleting queues for a process that is listening right now; they have no dead-letter queue. A conflict is requeued, any other failure is dropped — there is nobody to return a message to.

When the host stops

A subscription stops with the token it was opened with — for a worker, the host's stopping token. It takes no further message, waits for the handler it is running to settle, and then closes its channel. From the moment the application starts stopping — ApplicationStopping included — until the host has stopped, it waits as long as the host's shutdown timeout allows (HostOptions.ShutdownTimeout, 30 seconds by default; an infinite timeout waits as long as the handler runs); a subscription whose own token was cancelled at any other time waits twenty seconds. The host counts as stopped only after the subscriptions its hosted services stopped have closed, so a handler that is still running finishes while the services it uses exist, and its message is acknowledged rather than delivered again. A handler's outcome is settled whatever the subscription's token says, and a handler that gives up on that token — an OperationCanceledException while the subscription stops — has its message requeued rather than dead-lettered by the framework. The broker counts that delivery like any other: it uses up one of MaxDeliveryAttempts, and before RabbitMQ 4.3 it counts against the queue's x-delivery-limit too.

A handler that takes longer than the wait does not hold the stop up. The subscription closes its channel; the broker acknowledges the close and puts the handler's message back at once, so another consumer may take it while the first handler is still running — the handler must tolerate that, as at-least-once delivery already requires. The subscription counts as stopped at most five seconds later, whether or not the handler has returned.

What the subscription had fetched but not yet handed to its handler goes back to the queue unhandled. The broker counts that as a delivery, so each such stop brings those messages one delivery closer to their bound. The number a subscription holds at once, the running handler's message included, is therefore bounded:

{
  "Messaging": {
    "PrefetchCount": 16   // 1 to 65535; a value outside is refused at start-up
  }
}

The handler still runs one message at a time; a higher bound only hides more network latency.

Backpressure

The OutboxWorker polls the outbox table every OutboxOptions.PollingIntervalSeconds (default 30) and publishes pending rows. If the broker is unreachable, rows sit in the table — at-least-once delivery preserved. The next poll-cycle retries.

A cycle takes one batch of each kind and ends. Rows the broker would not accept stay in the table and are retried on the next interval; a cycle never re-reads what it has just failed to publish. That bounds the work a cycle can do, and it is what stops an unreachable broker — or a suppressed drain during a projection replay — from turning a cycle into a loop over the same rows. The practical consequence: a large accumulated backlog drains at one batch per interval rather than in a single pass. With the defaults that is 20 000 rows a minute, and both knobs below are yours.

OutboxOptions.BatchSize (default 10_000) caps how many rows the worker claims per cycle, and LockLeaseSeconds (default 60) is how long a claimed batch stays leased to one worker. Bind them under the Outbox configuration section.

Running more than one outbox worker

Every poll cycle runs under a lease-based lock (IOutboxLock), so two replicas never drain the same rows at the same time. A replica that does not get the lock logs it (106_005, debug level) and skips the cycle entirely, then tries again at the next interval.

The lock registered by default always grants. That is correct for exactly one outbox worker and unsafe for several. Before you scale out, replace it with the Redis-leased lock:

builder.AddCaching();                    // IConnectionMultiplexer from ConnectionStrings:redis
builder.Services.AddRedisOutboxLock();   // replaces the always-granting default

The lease is LockLeaseSeconds from the Outbox section. Choose one comfortably longer than your slowest drain. A holder that stalls past its lease can overlap with a peer, and at-least-once delivery is what absorbs the resulting duplicate publish.

The lock is released only by its holder

A release affects only the lease the releasing instance acquired. Each acquisition stores a token that only that instance holds, and the release removes the lease only while it still carries that token.

That matters for a replica whose lease ran out mid-cycle, whether from a long garbage-collection pause, a stalled disk or a network hiccup. By the time it finishes and releases, another replica may have taken the lock. The late release leaves that replica's lease alone and the lock stays held.

An unavailable lock service counts as "not acquired", and the worker keeps running. If Redis cannot be reached when a replica tries to acquire, the failure is logged as a warning (106_106) and the cycle is skipped, exactly as if another replica held the lock. The rows stay in the outbox table and the next interval tries again. If Redis cannot be reached at release, that is logged as well (106_107) and the lease expires on its own once LockLeaseSeconds have passed. Neither failure propagates out of the worker.

Connection health

Stratara.Outbox.RabbitMQ uses RabbitMQ.Client's automatic recovery + topology recovery. NetworkRecoveryInterval is set to a small default; consumers re-subscribe automatically after a reconnect.

On startup the bus fails fast in Production if the broker connection can't be established, rather than starting a worker that silently publishes nowhere.

Observability

The outbox plane records the outbox.published counter (on the Stratara.Service meter), tagged by entry kind — command or event — so you can watch command-dispatch and event-bundle throughput separately. It counts what the broker accepted, not what was read from the table: a row that could not be published is not counted, so the counter going flat while the table stays full is the signal that dispatch is stuck rather than busy. The Stratara.ServiceDefaults OpenTelemetry config wires the Stratara.Service meter and the Stratara.Application activity source automatically.

Failure paths (PublishReturnException on no-binding, broker-disconnect, …) emit warning-level log events from Stratara.Shared.Diagnostics.Extensions.LoggerOutboxExtensions — see the LogEvents Schema.

Edit this page