Write business state and integration events in one database transaction, publish them safely after commit, and make duplicate delivery harmless.
Build an Orders API that commits an order and an integration event together, publishes the event later through a relay, and safely ignores duplicate deliveries in the consumer.
POST /api/orders — create an order and its outbox messageGET /api/orders/{id} — read an orderGET /api/outbox — inspect pending and processed messages in developmentGET /health — application and database healthOrders API
│
├── INSERT Order ─────────────┐
└── INSERT OutboxMessage ─────┴── one database transaction
│
▼
Outbox dispatcher
│
▼
RabbitMQ
│
▼
Idempotent consumer + inboxA normal request often needs to change the database and notify another service. Those are two independent systems. They cannot be committed with one ordinary local transaction.
app.MapPost("/api/orders", async (
CreateOrderRequest request,
OrdersDbContext db,
IMessagePublisher publisher,
CancellationToken cancellationToken) =>
{
var order = Order.Create(request.CustomerId, request.Total);
db.Orders.Add(order);
await db.SaveChangesAsync(cancellationToken);
// Failure window: the order is committed, but publishing may fail.
await publisher.PublishAsync(
new OrderPlaced(order.Id, order.CustomerId, order.Total),
cancellationToken);
return Results.Created($"/api/orders/{order.Id}", order);
});A try/catch cannot make two independent systems atomic. The transactional outbox changes the problem: write the event to the same database transaction as the business state, then publish it asynchronously.
BackgroundService and cancellation tokens.mkdir OutboxOrders
cd OutboxOrders
dotnet new sln -n OutboxOrders
dotnet new webapi -n OutboxOrders.Api -o src/OutboxOrders.Api
dotnet new classlib -n OutboxOrders.Contracts -o src/OutboxOrders.Contracts
dotnet new worker -n OutboxOrders.Consumer -o src/OutboxOrders.Consumer
dotnet new xunit -n OutboxOrders.Tests -o tests/OutboxOrders.Tests
dotnet sln add src/OutboxOrders.Api
dotnet sln add src/OutboxOrders.Contracts
dotnet sln add src/OutboxOrders.Consumer
dotnet sln add tests/OutboxOrders.Tests
dotnet add src/OutboxOrders.Api reference src/OutboxOrders.Contracts
dotnet add src/OutboxOrders.Consumer reference src/OutboxOrders.Contracts<Project>
<PropertyGroup>
<ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
</PropertyGroup>
<ItemGroup>
<!-- Replace placeholders with the versions restored and tested before publishing. -->
<PackageVersion Include="Microsoft.EntityFrameworkCore.Design" Version="10.x.y" />
<PackageVersion Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="10.x.y" />
<PackageVersion Include="RabbitMQ.Client" Version="7.x.y" />
<PackageVersion Include="Microsoft.AspNetCore.OpenApi" Version="10.x.y" />
<PackageVersion Include="Testcontainers.PostgreSql" Version="4.x.y" />
<PackageVersion Include="Testcontainers.RabbitMq" Version="4.x.y" />
</ItemGroup>
</Project>services:
postgres:
image: postgres:17
environment:
POSTGRES_DB: orders
POSTGRES_USER: app
POSTGRES_PASSWORD: dev-password
ports:
- "5432:5432"
healthcheck:
test: ["CMD-SHELL", "pg_isready -U app -d orders"]
interval: 5s
timeout: 5s
retries: 10
rabbitmq:
image: rabbitmq:4-management
ports:
- "5672:5672"
- "15672:15672"
healthcheck:
test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"]
interval: 5s
timeout: 5s
retries: 10public sealed class OutboxMessage
{
public Guid Id { get; init; }
public string Type { get; init; } = string.Empty;
public string Payload { get; init; } = string.Empty;
public DateTimeOffset OccurredOnUtc { get; init; }
public DateTimeOffset? ProcessedOnUtc { get; set; }
public int Attempts { get; set; }
public string? Error { get; set; }
public string? TraceParent { get; init; }
public DateTimeOffset? NextAttemptOnUtc { get; set; }
public static OutboxMessage From<T>(
T message,
string? traceParent,
JsonSerializerOptions jsonOptions)
{
ArgumentNullException.ThrowIfNull(message);
return new OutboxMessage
{
Id = Guid.NewGuid(),
Type = typeof(T).FullName
?? throw new InvalidOperationException("Message type has no full name."),
Payload = JsonSerializer.Serialize(message, jsonOptions),
OccurredOnUtc = DateTimeOffset.UtcNow,
TraceParent = traceParent
};
}
}protected override void OnModelCreating(ModelBuilder modelBuilder)
{
base.OnModelCreating(modelBuilder);
modelBuilder.Entity<OutboxMessage>(entity =>
{
entity.ToTable("OutboxMessages");
entity.HasKey(x => x.Id);
entity.Property(x => x.Type).HasMaxLength(500).IsRequired();
entity.Property(x => x.Payload).HasColumnType("jsonb").IsRequired();
entity.Property(x => x.Error).HasMaxLength(4000);
entity.Property(x => x.TraceParent).HasMaxLength(256);
entity.HasIndex(x => new
{
x.ProcessedOnUtc,
x.NextAttemptOnUtc,
x.OccurredOnUtc
});
});
}The row id is also the message id sent to the broker. That stable id gives the consumer a reliable deduplication key.
public sealed record OrderPlaced(
Guid OrderId,
Guid CustomerId,
decimal Total,
DateTimeOffset OccurredOnUtc);
app.MapPost("/api/orders", async (
CreateOrderRequest request,
OrdersDbContext db,
JsonSerializerOptions jsonOptions,
CancellationToken cancellationToken) =>
{
var order = Order.Create(request.CustomerId, request.Total);
var integrationEvent = new OrderPlaced(
order.Id,
order.CustomerId,
order.Total,
DateTimeOffset.UtcNow);
db.Orders.Add(order);
db.OutboxMessages.Add(OutboxMessage.From(
integrationEvent,
Activity.Current?.Id,
jsonOptions));
// One SaveChanges call means EF Core commits both rows together.
await db.SaveChangesAsync(cancellationToken);
return Results.Created($"/api/orders/{order.Id}", new { order.Id });
});A single SaveChangesAsync call is already transactional for the changes it sends. The explicit transaction in the next section becomes useful when the operation includes multiple saves or other database work that must be retried as one unit.
builder.Services.AddDbContext<OrdersDbContext>(options =>
options.UseNpgsql(
builder.Configuration.GetConnectionString("Orders"),
npgsql => npgsql.EnableRetryOnFailure()));var strategy = db.Database.CreateExecutionStrategy();
await strategy.ExecuteAsync(async () =>
{
await using var transaction =
await db.Database.BeginTransactionAsync(cancellationToken);
db.Orders.Add(order);
db.OutboxMessages.Add(outboxMessage);
await db.SaveChangesAsync(cancellationToken);
await transaction.CommitAsync(cancellationToken);
});The execution strategy may run the delegate again after a transient failure. Keep the work deterministic and avoid external side effects, such as publishing to RabbitMQ, inside this delegate.
public interface IDomainEvent
{
DateTimeOffset OccurredOnUtc { get; }
}
public abstract class AggregateRoot
{
private readonly List<IDomainEvent> _domainEvents = [];
public IReadOnlyCollection<IDomainEvent> DomainEvents => _domainEvents;
protected void Raise(IDomainEvent domainEvent) =>
_domainEvents.Add(domainEvent);
public IReadOnlyCollection<IDomainEvent> DequeueDomainEvents()
{
IDomainEvent[] events = [.. _domainEvents];
_domainEvents.Clear();
return events;
}
}public sealed class OutboxInterceptor(
JsonSerializerOptions jsonOptions)
: SaveChangesInterceptor
{
public override ValueTask<InterceptionResult<int>> SavingChangesAsync(
DbContextEventData eventData,
InterceptionResult<int> result,
CancellationToken cancellationToken = default)
{
if (eventData.Context is not OrdersDbContext db)
{
return base.SavingChangesAsync(
eventData,
result,
cancellationToken);
}
var events = db.ChangeTracker
.Entries<AggregateRoot>()
.SelectMany(entry => entry.Entity.DequeueDomainEvents())
.ToArray();
foreach (IDomainEvent domainEvent in events)
{
db.OutboxMessages.Add(OutboxMessage.From(
domainEvent,
Activity.Current?.Id,
jsonOptions));
}
return base.SavingChangesAsync(
eventData,
result,
cancellationToken);
}
}builder.Services.AddSingleton<OutboxInterceptor>();
builder.Services.AddDbContext<OrdersDbContext>((services, options) =>
{
options.UseNpgsql(
builder.Configuration.GetConnectionString("Orders"),
npgsql => npgsql.EnableRetryOnFailure());
options.AddInterceptors(
services.GetRequiredService<OutboxInterceptor>());
});The interceptor centralizes capture. Business code raises a domain event; the interceptor translates it into an outbox row immediately before EF Core saves the transaction.
public sealed class OutboxDispatcher(
IServiceScopeFactory scopeFactory,
TimeProvider timeProvider,
IOptions<OutboxOptions> options,
ILogger<OutboxDispatcher> logger)
: BackgroundService
{
protected override async Task ExecuteAsync(
CancellationToken stoppingToken)
{
using var timer = new PeriodicTimer(
TimeSpan.FromSeconds(options.Value.PollingIntervalSeconds),
timeProvider);
do
{
try
{
await DispatchBatchAsync(stoppingToken);
}
catch (OperationCanceledException)
when (stoppingToken.IsCancellationRequested)
{
break;
}
catch (Exception exception)
{
logger.LogError(exception, "Outbox dispatch failed.");
}
}
while (await timer.WaitForNextTickAsync(stoppingToken));
}
private async Task DispatchBatchAsync(CancellationToken cancellationToken)
{
await using AsyncServiceScope scope = scopeFactory.CreateAsyncScope();
var service = scope.ServiceProvider
.GetRequiredService<OutboxBatchProcessor>();
await service.ProcessAsync(cancellationToken);
}
}The dispatcher creates a new dependency-injection scope for each batch. That gives each batch a fresh DbContext and prevents a long-running worker from holding stale tracked entities.
public interface IMessagePublisher
{
Task PublishAsync(
PublishedMessage message,
CancellationToken cancellationToken);
}
public sealed record PublishedMessage(
Guid MessageId,
string Type,
ReadOnlyMemory<byte> Body,
string? TraceParent);public sealed class RabbitMqPublisher(
IConnection connection,
IOptions<RabbitMqOptions> options)
: IMessagePublisher
{
public async Task PublishAsync(
PublishedMessage message,
CancellationToken cancellationToken)
{
await using IChannel channel =
await connection.CreateChannelAsync(cancellationToken: cancellationToken);
var properties = new BasicProperties
{
MessageId = message.MessageId.ToString("D"),
Type = message.Type,
Persistent = true,
Headers = message.TraceParent is null
? null
: new Dictionary<string, object?>
{
["traceparent"] = message.TraceParent
}
};
await channel.BasicPublishAsync(
exchange: options.Value.Exchange,
routingKey: message.Type,
mandatory: true,
basicProperties: properties,
body: message.Body,
cancellationToken: cancellationToken);
}
}The application and relay depend on IMessagePublisher, not RabbitMQ SDK types. Azure Service Bus, Kafka, or another transport can be introduced behind the same application-facing contract.
SELECT "Id", "Type", "Payload", "OccurredOnUtc", "TraceParent"
FROM "OutboxMessages"
WHERE "ProcessedOnUtc" IS NULL
AND ("NextAttemptOnUtc" IS NULL OR "NextAttemptOnUtc" <= now())
ORDER BY "OccurredOnUtc"
FOR UPDATE SKIP LOCKED
LIMIT @batch_size;public async Task ProcessAsync(CancellationToken cancellationToken)
{
var strategy = db.Database.CreateExecutionStrategy();
await strategy.ExecuteAsync(async () =>
{
await using var transaction =
await db.Database.BeginTransactionAsync(cancellationToken);
List<OutboxMessage> messages = await db.OutboxMessages
.FromSqlInterpolated($"""
SELECT *
FROM "OutboxMessages"
WHERE "ProcessedOnUtc" IS NULL
AND ("NextAttemptOnUtc" IS NULL
OR "NextAttemptOnUtc" <= now())
ORDER BY "OccurredOnUtc"
FOR UPDATE SKIP LOCKED
LIMIT {options.Value.BatchSize}
""")
.ToListAsync(cancellationToken);
foreach (OutboxMessage message in messages)
{
await PublishOneAsync(message, cancellationToken);
}
await db.SaveChangesAsync(cancellationToken);
await transaction.CommitAsync(cancellationToken);
});
}SKIP LOCKED lets several relay instances select different rows without waiting for one another. PostgreSQL documents this as suitable for queue-like access, not for a general consistent view.
UPDLOCK, READPAST, and usually ROWLOCK within a transaction. Test the exact query under your isolation level.ProcessedOnUtc. The row will be published again after restart.public sealed class ProcessedMessage
{
public Guid MessageId { get; init; }
public DateTimeOffset ProcessedOnUtc { get; init; }
}
modelBuilder.Entity<ProcessedMessage>(entity =>
{
entity.HasKey(x => x.MessageId);
});public async Task HandleAsync(
Guid messageId,
OrderPlaced message,
CancellationToken cancellationToken)
{
await using var transaction =
await db.Database.BeginTransactionAsync(cancellationToken);
bool alreadyProcessed = await db.ProcessedMessages
.AnyAsync(x => x.MessageId == messageId, cancellationToken);
if (alreadyProcessed)
{
return;
}
db.OrderNotifications.Add(new OrderNotification
{
Id = Guid.NewGuid(),
OrderId = message.OrderId,
CustomerId = message.CustomerId,
CreatedOnUtc = DateTimeOffset.UtcNow
});
db.ProcessedMessages.Add(new ProcessedMessage
{
MessageId = messageId,
ProcessedOnUtc = DateTimeOffset.UtcNow
});
await db.SaveChangesAsync(cancellationToken);
await transaction.CommitAsync(cancellationToken);
}The unique primary key on MessageId is the final protection against two consumer instances racing. Catch the provider-specific unique-constraint exception and treat it as “already processed.”
private async Task PublishOneAsync(
OutboxMessage message,
CancellationToken cancellationToken)
{
try
{
await publisher.PublishAsync(
new PublishedMessage(
message.Id,
message.Type,
Encoding.UTF8.GetBytes(message.Payload),
message.TraceParent),
cancellationToken);
message.ProcessedOnUtc = timeProvider.GetUtcNow();
message.Error = null;
}
catch (Exception exception)
{
message.Attempts++;
message.Error = exception.Message[..Math.Min(exception.Message.Length, 4000)];
if (message.Attempts >= options.Value.MaxAttempts)
{
message.NextAttemptOnUtc = null;
message.ProcessedOnUtc = timeProvider.GetUtcNow();
deadLetterWriter.Write(message, exception);
return;
}
TimeSpan delay = TimeSpan.FromSeconds(
Math.Min(Math.Pow(2, message.Attempts), 300));
message.NextAttemptOnUtc = timeProvider.GetUtcNow() + delay;
}
}A poison message is one that will never publish successfully without intervention—for example, an invalid contract or oversized payload. Do not let it block every newer message.
private static readonly ActivitySource ActivitySource =
new("OutboxOrders.Outbox");
private async Task PublishWithTraceAsync(
OutboxMessage message,
CancellationToken cancellationToken)
{
ActivityContext parentContext = default;
if (message.TraceParent is not null)
{
ActivityContext.TryParse(
message.TraceParent,
traceState: null,
isRemote: true,
out parentContext);
}
using Activity? activity = ActivitySource.StartActivity(
"outbox publish",
ActivityKind.Producer,
parentContext);
activity?.SetTag("messaging.message.id", message.Id);
activity?.SetTag("messaging.destination.name", options.Value.Exchange);
activity?.SetTag("outbox.attempt", message.Attempts + 1);
await publisher.PublishAsync(
new PublishedMessage(
message.Id,
message.Type,
Encoding.UTF8.GetBytes(message.Payload),
activity?.Id ?? message.TraceParent),
cancellationToken);
}Monitor more than error logs. The most useful outbox signals are:
A five-second polling interval is simple and reliable, but it adds latency. A useful optimization is push-then-poll:
Channel<bool>.public sealed class OutboxSignal
{
private readonly Channel<bool> _channel = Channel.CreateBounded<bool>(
new BoundedChannelOptions(1)
{
FullMode = BoundedChannelFullMode.DropWrite,
SingleReader = true,
SingleWriter = false
});
public void Notify() => _channel.Writer.TryWrite(true);
public ValueTask<bool> WaitAsync(CancellationToken cancellationToken) =>
_channel.Reader.ReadAsync(cancellationToken);
}For very high throughput, Change Data Capture tools such as Debezium can read committed outbox rows from the database log and publish them without application polling. That is a different operational model and is intentionally outside this hands-on implementation.
[Fact]
public async Task Creating_order_writes_order_and_outbox_together()
{
await using OrdersDbContext db = CreateDbContext();
var order = Order.Create(Guid.NewGuid(), 125m);
db.Orders.Add(order);
db.OutboxMessages.Add(CreateOutboxMessage(order));
await db.SaveChangesAsync();
Assert.True(await db.Orders.AnyAsync(x => x.Id == order.Id));
Assert.True(await db.OutboxMessages.AnyAsync());
}
[Fact]
public async Task Duplicate_message_is_applied_once()
{
Guid messageId = Guid.NewGuid();
var message = new OrderPlaced(
Guid.NewGuid(), Guid.NewGuid(), 125m, DateTimeOffset.UtcNow);
await handler.HandleAsync(messageId, message, CancellationToken.None);
await handler.HandleAsync(messageId, message, CancellationToken.None);
Assert.Equal(1, await db.OrderNotifications.CountAsync());
Assert.Equal(1, await db.ProcessedMessages.CountAsync());
}Use real PostgreSQL and RabbitMQ containers for integration tests. EF Core InMemory cannot reproduce transactions, relational locks, provider errors, or SKIP LOCKED.
docker compose up -d
dotnet ef database update --project src/OutboxOrders.Api --startup-project src/OutboxOrders.Api
dotnet run --project src/OutboxOrders.Api
dotnet run --project src/OutboxOrders.Consumer# Create an order while RabbitMQ is running.
curl -k -X POST https://localhost:7001/api/orders -H "Content-Type: application/json" -d '{"customerId":"11111111-1111-1111-1111-111111111111","total":125.00}'
# Stop the broker, then create another order.
docker compose stop rabbitmq
curl -k -X POST https://localhost:7001/api/orders -H "Content-Type: application/json" -d '{"customerId":"22222222-2222-2222-2222-222222222222","total":75.00}'
# The API succeeds because the database transaction does not depend on RabbitMQ.
# Restart RabbitMQ and watch the relay publish the backlog.
docker compose start rabbitmqThe code in this tutorial is intentionally hand-rolled so every reliability rule is visible. Production teams may prefer a library that also supplies broker integrations, retries, sagas, dashboards, and operational tooling.
| Option | License | Outbox support | Use when |
|---|---|---|---|
| Hand-rolled | Your code | You implement it | The needs are focused and the team wants full control. |
| MassTransit v8 | Apache 2.0 | Built in | You already use v8 and accept its support horizon. |
| MassTransit v9+ | Commercial | Built in | The commercial terms and support model fit the organization. |
| Wolverine | MIT | Durable outbox/inbox | You want messaging plus durable local queues and EF Core/Marten integration. |
| DotNetCore.CAP | MIT | Core design | You want an outbox-oriented event bus with common brokers. |
| Brighter | MIT | Available | You use command processing and messaging patterns. |
| Rebus | MIT | Available | You prefer a mature, lightweight service bus. |
| NServiceBus | Commercial | Built in | You need enterprise support and a broad messaging platform. |
Directory.Packages.props, not in the article prose.This implementation solves the producer-side dual write. Natural follow-up tutorials are:
No. It provides eventual at-least-once delivery. If the relay publishes and crashes before marking the row processed, it publishes the same message again. Consumers must deduplicate by message id.
The database transaction cannot atomically commit RabbitMQ. Holding the transaction open during network I/O also increases lock time and still cannot remove the failure window.
No. Explicitly adding an outbox row is valid and easy to understand. The interceptor is useful when many commands raise domain events and you want one central capture rule.
Not for the important behavior. Use the actual relational provider because transactions, locks, SQL syntax, and retry behavior are provider-specific.