Implementing the Outbox Pattern for Reliable Event Delivery
Written on
In contemporary distributed systems, maintaining consistent event delivery among various services is essential. A common issue arises when there is a need to store data in a database while also publishing it to an external service, such as a message queue or event broker. Any disruption in this process can lead to data inconsistency.
This is where the Outbox Pattern becomes significant.
What is the Outbox Pattern?
The Outbox Pattern is a reliable messaging pattern that maintains consistency between your application's transactional database and external systems. It ensures that once a database transaction is successfully completed, the corresponding events or messages are reliably published, even if a failure occurs.
Instead of sending messages directly to an external service (like Kafka, RabbitMQ, or Azure Service Bus) during a transaction, this pattern introduces an outbox table (or its equivalent) to temporarily store events alongside the primary database changes. A distinct process, typically a background worker, retrieves these events from the outbox and sends them to the external system.
The Problem it Solves: Preventing Data Inconsistency
Without the Outbox Pattern, you may encounter scenarios such as:
- Successful database commit but failed message delivery: The database saves the changes, but the message (event) never reaches the external system, leaving it unaware of the update.
- Message sent but database commit fails: The event is dispatched to the external system, but the database transaction fails, resulting in a mismatch between the database state and the external system.
For instance, in an e-commerce platform, if a customer places an order but the event is not communicated to the inventory service, the inventory may not update accurately. Such inconsistencies can lead to serious problems.
By employing the Outbox Pattern, both the database write and the outbox message insertion occur within the same transaction. If one fails, both fail, ensuring atomicity and consistency.
Use Case: Managing Orders with the Outbox Pattern
Let's consider a practical scenario where the Outbox Pattern is applied: Order Management in an E-Commerce Application.
The objective is to make sure that after an order is placed, an event is reliably sent to a message broker (like RabbitMQ, Azure Service Bus, or Kafka) to inform other services, such as inventory management, customer notifications, or analytics.
Architecture Overview
- Order Service: Manages order creation and saves data to the database.
- Outbox Table: Holds events related to the orders that have been placed and are waiting to be sent to a message broker.
- Message Dispatcher: A background service that reads from the outbox table and sends messages to the external service (e.g., Azure Service Bus).
- Message Broker: Distributes events to other microservices.
The outbox guarantees that if the database commit is successful but the message broker is temporarily down, the event will still be stored and can be retried later.
Step-by-Step Implementation
Define an Outbox Table:
Assume we are using a NoSQL database like MongoDB. If you're using SQL databases, the logic is quite similar. You'll need a collection or table to store outgoing events.
{
"OrderId": "ae12c2b4-4d6f-4a97-8aaf-2e234fb91b73",
"EventType": "OrderPlaced",
"EventPayload": {
"OrderId": "ae12c2b4-4d6f-4a97-8aaf-2e234fb91b73",
"CustomerId": "c123",
"TotalAmount": 199.99
},
"CreatedAt": "2024-09-12T08:30:00Z",
"ProcessedAt": null
}
The OutboxMessage collection contains events that need to be sent out with the following fields:
- OrderId: Links to the related order.
- EventType: Type of event (e.g., OrderPlaced, OrderCancelled).
- EventPayload: The serialized event data.
- CreatedAt: The timestamp when the message was created.
- ProcessedAt: The timestamp when the message was dispatched (null if not yet processed).
Insert Event into the Outbox on Order Creation:
Here’s the logic for placing an order and saving the event to the outbox.
public class OrderService
{
private readonly IMongoCollection<Order> _orderCollection;
private readonly IMongoCollection<OutboxMessage> _outboxCollection;
public OrderService(IMongoCollection<Order> orderCollection, IMongoCollection<OutboxMessage> outboxCollection)
{
_orderCollection = orderCollection;
_outboxCollection = outboxCollection;
}
public async Task PlaceOrderAsync(Order order)
{
using var session = await _mongoClient.StartSessionAsync();
session.StartTransaction();
try
{
// Save order to the database
await _orderCollection.InsertOneAsync(session, order);
// Create outbox event
var outboxMessage = new OutboxMessage
{
Id = Guid.NewGuid(),
EventType = "OrderPlaced",
EventPayload = JsonConvert.SerializeObject(new OrderPlacedEvent
{
OrderId = order.Id,
CustomerId = order.CustomerId,
TotalAmount = order.TotalAmount
}),
CreatedAt = DateTime.UtcNow
};
// Save event to outbox
await _outboxCollection.InsertOneAsync(session, outboxMessage);
// Commit transaction
await session.CommitTransactionAsync();
}
catch
{
await session.AbortTransactionAsync();
throw;
}
}
}
Implement a Message Dispatcher:
Next, we need a service that periodically reads unprocessed messages from the outbox and sends them to a message broker.
Here’s how you could implement this in an ASP.NET Core background service:
public class OutboxMessageDispatcher : BackgroundService
{
private readonly IMongoCollection<OutboxMessage> _outboxCollection;
private readonly IMessageBrokerService _messageBroker;
public OutboxMessageDispatcher(IMongoCollection<OutboxMessage> outboxCollection, IMessageBrokerService messageBroker)
{
_outboxCollection = outboxCollection;
_messageBroker = messageBroker;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
var unprocessedMessages = await _outboxCollection
.Find(message => message.ProcessedAt == null)
.ToListAsync(stoppingToken);
foreach (var message in unprocessedMessages)
{
// Deserialize event and send to message broker
var eventType = Type.GetType(message.EventType);
var eventPayload = JsonConvert.DeserializeObject(message.EventPayload, eventType);
await _messageBroker.PublishAsync(eventPayload);
// Mark the outbox message as processed
var update = Builders<OutboxMessage>.Update.Set(m => m.ProcessedAt, DateTime.UtcNow);
await _outboxCollection.UpdateOneAsync(m => m.Id == message.Id, update);
}
await Task.Delay(5000, stoppingToken); // Check every 5 seconds
}
}
}
In this code:
- We query for unprocessed outbox messages.
- Deserialize the event from the message payload and send it to the message broker.
- After successfully dispatching, we mark the message as Processed by updating the ProcessedAt timestamp.
Register and Run the Dispatcher:
In your Startup.cs or Program.cs (depending on .NET version), you would register the background service and configure it to run as part of your application.
public class Startup
{
public void ConfigureServices(IServiceCollection services)
{
services.AddSingleton<IMongoCollection<OutboxMessage>>(sp =>
new MongoClient("mongodb://localhost:xxxxx").GetDatabase("OrdersDb").GetCollection<OutboxMessage>("OutboxMessages"));
services.AddSingleton<IMessageBrokerService, AzureServiceBusBrokerService>();
services.AddHostedService<OutboxMessageDispatcher>();
}
}
This registration ensures that the OutboxMessageDispatcher is consistently running in the background, checking for and dispatching unprocessed events.
Explore Further
- Retries and Error Handling: Implement retry mechanisms to manage temporary failures in dispatching messages to the message broker. Libraries like Polly can assist with this.
- High Throughput: In high-throughput environments, process unprocessed messages in batches instead of one at a time to enhance performance.
- Message Ordering: If the order of messages is critical, the outbox pattern must consider the sequence in which events are processed and dispatched, achieved by querying the outbox table chronologically.
- Idempotency: Ensure that the external systems consuming the messages are idempotent. Despite the outbox pattern ensuring consistency, some edge cases might result in a message being processed more than once. Idempotent consumers can handle such cases without negative consequences.