The transactional outbox pattern guarantees that domain events raised by aggregate roots are published even if the process crashes after the database commit. The engine lives in Vulthil.SharedKernel.Outbox; Vulthil.SharedKernel.Infrastructure references it and adds the DbContext base and the EnableOutboxProcessing wiring.
- During
SaveChanges/SaveChangesAsync(both the synchronous and asynchronous paths are intercepted), aSaveChangesInterceptorserialises every pending domain event into anOutboxMessagerow in the same database transaction. - The aggregate root's event collection is cleared.
- A background service (
OutboxBackgroundService) relays unprocessed outbox messages — woken immediately once the captured rows are durable (on transaction commit, or right after a non-transactionalSaveChangesthat captured domain events) for low latency, and polling on an interval as the backstop. - Each message is routed by its
OutboxDestinationto the registeredIOutboxDispatcherthat handles it (in-process domain events by default, or the broker — see below). - Successfully relayed messages are marked as processed; failures are retried up to the configured maximum, after which the message is dead-lettered — its
FailedOnUtctimestamp is set, it is no longer relayed, and an error is logged with the last failure (theErrorcolumn).
This guarantees at-least-once delivery because the event and the business data are committed atomically.
Relay order is best-effort, not a contract: messages are fetched oldest-first (OccurredOnUtc, then Id), but a
failed message does not block the rest of its batch — the batch keeps dispatching, so later messages overtake the
failure, which is retried on a later cycle (no head-of-line blocking). With EnableParallelPublishing, order
within a batch is not preserved at all. Consumers that need ordering must not rely on the outbox for it.
AddDbContext is an extension on IHostApplicationBuilder (not IServiceCollection). Provider extensions such as UseNpgsql are called directly on the configurator — they register the EF Core context for you, so no separate AddDbContext/options callback is needed.
builder.AddDbContext<AppDbContext>(config =>
{
config
.UseNpgsql(connectionStringKey)
.EnableOutboxProcessing(o =>
{
o.BatchSize = 10; // Messages fetched per poll cycle
o.MaxRetries = 3; // Retries before a message is dead-lettered
});
});Your context must derive from BaseDbContext, which already implements ISaveOutboxMessages and includes the OutboxMessages DbSet:
public sealed class AppDbContext(DbContextOptions<AppDbContext> options)
: BaseDbContext(options)
{
protected override Assembly? ConfigurationAssembly =>
typeof(AppDbContext).Assembly;
public DbSet<User> Users => Set<User>();
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
base.OnModelCreating(modelBuilder);
modelBuilder.ApplyNpgsqlOutbox();
}
}BaseDbContext owns the OutboxMessages DbSet; apply the provider-optimized mapping in OnModelCreating by calling your provider's extension — ApplyNpgsqlOutbox(), ApplyMySqlOutbox(), or ApplyCosmosOutbox() (as shown above). The agnostic ApplyOutbox() is available for custom providers.
Because the DbSet lives on the base class, EF Core's convention discovery puts the OutboxMessage entity into
every derived context's model — and its migrations — even when outbox processing is never enabled for that
context. A context that will never use the outbox can opt out with modelBuilder.Ignore<OutboxMessage>() in its
OnModelCreating; an outbox-enabled context must keep the mapping (the relay store refuses to run without it).
Only one outbox-enabled DbContext is supported per host: the relay and retention background services resolve a
single IOutboxStore, so calling EnableOutboxProcessing() for a second context throws an
InvalidOperationException at startup instead of silently leaving the first context's messages unrelayed. The
Npgsql and MySQL stores also require your context to derive from BaseDbContext (or otherwise implement
IUnitOfWork) — without a transaction their FOR UPDATE SKIP LOCKED fetch would release its locks immediately,
letting concurrent relay instances double-dispatch the same messages, so RelationalOutboxStore throws instead of
running unprotected.
| Property | Default | Description |
|---|---|---|
BatchSize |
10 | Number of messages fetched per poll cycle |
MaxRetries |
3 | Publish attempts before a message is dead-lettered (FailedOnUtc set, no longer relayed) |
EnableParallelPublishing |
false |
Publish messages in parallel within a batch (each dispatch runs in its own DI scope) |
MaxDegreeOfParallelism |
4 | Maximum concurrent dispatches when EnableParallelPublishing is enabled |
OutboxProcessingDelaySeconds |
2 | Base polling delay between processing cycles |
MaxDelaySeconds |
60 | Maximum back-off delay when a cycle relays nothing (no pending messages, or every fetched message failed) |
EnableTracing |
true |
Carry the originating trace identifier when publishing |
The relay emits an ActivitySource named "Vulthil.SharedKernel.Outbox" (exposed as Telemetry.ActivitySourceName). When EnableTracing is on (the default), each relayed message starts an OutboxPublishing span parented on the trace that captured the row — the originating trace is carried forward through the OutboxMessage.TraceParent/TraceState columns stamped at capture — so the relay, which runs later on its own background service, still correlates back to the request that produced the message.
AddOutboxEngine (called by EnableOutboxProcessing) registers the source with OpenTelemetry automatically when EnableTracing is on (the default), so the spans reach whatever tracer the application has configured without extra wiring. If you build a TracerProviderBuilder yourself, the same registration is available as tracing.AddVulthilOutboxInstrumentation() — sugar for AddSource(Telemetry.ActivitySourceName).
The relay also emits metrics on a Meter named Telemetry.MeterName ("Vulthil.SharedKernel.Outbox"): the counters vulthil.outbox.relayed and vulthil.outbox.failed. AddOutboxEngine auto-registers the meter with OpenTelemetry when EnableMetrics is on (the default); for a hand-built MeterProviderBuilder, use metrics.AddVulthilOutboxInstrumentation().
Processed and dead-lettered rows remain in the OutboxMessages table after relay, so the table grows unbounded unless they are pruned. Opt into a retention sweep — a background service that periodically deletes terminal rows older than a window — by enabling Retention on the outbox options:
.EnableOutboxProcessing(o =>
{
o.Retention.Enabled = true; // turn the sweep on
o.Retention.RetentionPeriod = TimeSpan.FromDays(7); // delete processed/dead-lettered rows older than this
o.Retention.SweepInterval = TimeSpan.FromHours(1);
o.Retention.BatchSize = 1000;
});AddOutboxEngine (called by EnableOutboxProcessing) registers the sweep only when Retention.Enabled is set, so it costs nothing when off.
Retention property |
Default | Description |
|---|---|---|
Enabled |
false |
Whether the retention sweep runs |
RetentionPeriod |
7 days | How long a processed or dead-lettered row is kept |
SweepInterval |
1 hour | Delay between sweeps |
BatchSize |
1000 | Rows deleted per batch within a sweep |
The sweep deletes rows whose ProcessedOnUtc or FailedOnUtc is older than RetentionPeriod; pending rows are never touched. It runs through the registered IOutboxStore when it implements IOutboxRetentionStore (the EF Core store does; a custom store that doesn't gets a one-time warning and the sweep is skipped) — every store deletes at most BatchSize rows per call (relational providers select the oldest eligible keys and delete them set-based with ExecuteDelete), so a large backlog is drained in bounded, short-lived batches, and the same sweep covers Cosmos (a Cosmos container TTL is not used, because it cannot tell a pending row from a relayed one and could expire an undelivered message).
The relay engine talks to the database through an EF-free IOutboxStore (in Vulthil.SharedKernel.Outbox). The EF
implementation lives in Vulthil.SharedKernel.Outbox.EntityFrameworkCore (EntityFrameworkOutboxStore<TContext>),
and each provider supplies a subclass with its row-locking fetch — RelationalOutboxStore<TContext> (the
ExecuteUpdate base), NpgsqlOutboxStore<TContext> / MySqlOutboxStore<TContext> (FOR UPDATE SKIP LOCKED), and
CosmosOutboxStore<TContext> (best-effort, no transaction). A provider's UseNpgsql/UseMySql/UseCosmosDb
selects the store; you can supply your own by implementing IOutboxStore (or deriving from the EF base) and
registering it with UseOutboxStore<TStore>():
config
.UseNpgsql(connectionStringKey)
.UseOutboxStore<CustomOutboxStore<AppDbContext>>()
.EnableOutboxProcessing(o =>
{
o.BatchSize = 50;
});The relay engine is sink-agnostic: each OutboxMessage carries an OutboxDestination discriminator, and the
OutboxProcessor routes it to the registered IOutboxDispatcher whose Handles(destination) is true. The
in-process domain-event dispatcher is registered by default; other sinks plug in and coexist in the same outbox
table and relay, so an application never carries more than one outbox table regardless of how many sinks it uses.
Vulthil.Messaging.Outbox adds a sink for the message broker. A publish/send filter captures any message published
while a database transaction is open into the same outbox table (atomically with the business changes); the relay
sends it to the broker after the transaction commits, carrying a stable message id so a redelivered relay is
deduplicated by the receiving inbox — end-to-end effectively-once. A publish issued with no
active transaction is sent directly.
builder.AddDbContext<AppDbContext>(config => config.UseNpgsql("Default").EnableOutboxProcessing());
builder.AddMessaging(messaging =>
{
messaging.UseRabbitMq();
messaging.AddTransactionalOutbox();
});Capture is relational-only (it enlists in the ambient transaction); the relay works on any provider. It is built on
the general publish/send filter pipeline (IPublishFilter, registered via AddPublishFilter<T>()), which is the
publish-side counterpart to consume filters and can host other cross-cutting concerns.
Two pipeline consequences to know: capturing short-circuits the pipeline, so publish filters registered after
AddTransactionalOutbox() never see a captured message; and the relay publishes captured messages straight through
the transport, bypassing the publish-filter pipeline entirely (nothing runs twice, but nothing else observes the
relayed publish either). Register any filter that must observe every outgoing message before
AddTransactionalOutbox(). Request/reply is unaffected — requests never flow through the publish filters and are
never captured (a request awaiting a synchronous reply cannot be deferred to a relay).
The capture preserves the publish faithfully: the message id, correlation id, routing key, send destination, custom
headers, and the current trace context (TraceParent/TraceState, when tracing is enabled) are stored with the row
and re-applied on relay — the relayed publish runs under an activity linked to the capturing trace. Headers persist
as JSON, so a relayed message and a directly-published one carry identical header bytes on the wire and surface
identically to consumers; only CLR types with no JSON primitive form (e.g. a Guid header value) are re-published
in their JSON-serialized form rather than as the original CLR type.
On startup the relay waits for the broker transport to finish declaring its subscriber topology (exchanges, queues,
and bindings) before its first publish — otherwise the commit-time trigger could relay a message before the
subscriber queues exist, and a pub/sub message with no bound queue is silently dropped. This is wired automatically
by AddTransactionalOutbox via an IOutboxRelayGate that awaits ITransport.WaitUntilReadyAsync; the relay starts
immediately when no gate is registered.
Capture only happens when a database transaction is open around the publish — otherwise the message is sent directly. The transaction is established by one of:
- Commands — mark them
ITransactionalCommand<T>and registerAddTransactionalPipelineBehavior(); the behavior runs the command in a transaction. - Consumers — the inbox opens one, or call
messaging.AddTransactionalConsumer<TMessage>()to run a consumer in a transaction without the inbox. The two compose: if the inbox is also enabled it opens the transaction and the consume filter joins it rather than nesting. - Anything else — wrap the work in
IUnitOfWork.ExecuteInTransactionAsync(...).
Ambient System.Transactions.TransactionScope transactions are not supported: the capture gate checks for an
Entity Framework Core transaction specifically, and EF Core does not surface an ambient scope as one. Publishing
inside a TransactionScope with no EF Core transaction throws NotSupportedException rather than silently
publishing directly while the scope is still uncommitted (a ghost message on rollback) — establish the transaction
with one of the options above instead. Full ambient-scope support may be added in the future; it is deferred over
provider enlistment edge cases.
Aggregate.Raise(event)
↓
SaveChangesAsync → OutboxMessage row inserted (same transaction)
↓
OutboxBackgroundService polls
↓
OutboxProcessor deserialises & publishes via IDomainEventPublisher
↓
Message marked as processed (or retried on failure, then dead-lettered after MaxRetries)
- You need reliable event delivery across service boundaries.
- You want to avoid dual-write problems (writing to the database and a message broker in separate transactions).
- Your domain events trigger side-effects in other bounded contexts or external systems.