Master the Outbox Pattern: Ensuring Data Consistency in Distributed .NET Systems

Master the Outbox Pattern

When you build Event Driven systems, one of the most challenging problems you'll face is ensuring that database changes and message publishing happen reliably together. It's like trying to make sure a letter is both written and mailed at the exact same time—if one fails, the whole operation should fail.

As we all know, 𝗠𝘂𝗿𝗽𝗵𝘆’𝘀 𝗟𝗮𝘄 states that things can go wrong quickly. Losing a message between two services can be a critical issue.

However, these benefits rely on certain assumptions, such as:

𝗥𝗲𝗹𝗶𝗮𝗯𝗹𝗲 𝗻𝗲𝘁𝘄𝗼𝗿𝗸
𝗭𝗲𝗿𝗼 𝗹𝗮𝘁𝗲𝗻𝗰𝘆
𝗜𝗻𝗳𝗶𝗻𝗶𝘁𝗲 𝗯𝗮𝗻𝗱𝘄𝗶𝗱𝘁𝗵
𝗛𝗼𝗺𝗼𝗴𝗲𝗻𝗲𝗼𝘂𝘀 𝗻𝗲𝘁𝘄𝗼𝗿𝗸

In my journey of refactoring I introduce Event Driven rewriting my legacy application and isn't somtehing I want that occur. That's why I use the Outbox Pattern—a battle-tested solution that has transformed how we handle distributed transactions.

The Problem: Dual-Write Challenge

In the previous article on event-driven architecture, we built an in-memory event system that worked perfectly within a single process. But what happens when we move to distributed systems?

Consider this scenario from our refactored OrderService:

// ❌ PROBLEM: What if the service bus call fails after the database commit?
public async Task CreateOrderAsync(Order order, CancellationToken cancellationToken = default)
{
    // Step 1: Save to database
    await _orderRepository.SaveAsync(order, cancellationToken);
    
    // Step 2: Publish to Service Bus
    // What if this fails? The order is already in the database!
    await _eventPublisher.PublishAsync(new OrderCreatedEvent
    {
        OrderId = order.Id,
        CustomerId = order.CustomerId,
        TotalAmount = order.Total
    }, cancellationToken);
}

Problems that can occur:

Database commits, Service Bus fails: Order is created but no one knows about it
Service Bus succeeds, database rollback: Handlers process an order that doesn't exist
Network timeout: We don't know if the message was published or not
Partial failure: Some downstream handlers process the event, others don't

This is the dual-write problem: coordinating writes to two different systems (database + message broker) without distributed transactions.

Understanding the Outbox Pattern

The Outbox Pattern solves this by treating message publishing as part of the database transaction. Instead of publishing messages directly, we:

Store events in an "outbox" table within the same database transaction
Commit everything together (data + events) - atomic operation
A separate process reads from the outbox and publishes messages reliably
Mark messages as published after successful delivery

This transforms the dual-write problem into a single-write problem with guaranteed eventual consistency.

Architecture Overview

Outbox Pattern

Step-by-Step Implementation in .NET

Let's build a production-ready Outbox Pattern implementation step by step, extending our OrderService from the previous article.

Step 1: Define the Outbox Table and Entity

First, we need a table to store our unpublished events:

// Domain/Outbox/OutboxMessage.cs
public class OutboxMessage
{
    public Guid Id { get; set; }
    public string EventType { get; set; } = string.Empty;
    public string EventData { get; set; } = string.Empty;
    public DateTime CreatedAt { get; set; }
    public DateTime? ProcessedAt { get; set; }
    public string? Error { get; set; }
    public int RetryCount { get; set; }
}

Key fields explained:

Id: Unique identifier for the outbox message
EventType: Fully qualified type name (e.g., "MyApp.Events.OrderCreatedEvent")
EventData: JSON serialized event data
CreatedAt: When the event was created
ProcessedAt: When successfully published (null = not yet processed)
Error: Last error message if publishing failed
RetryCount: Number of retry attempts

Step 2: Configure Entity Framework Core

Add the outbox table to your DbContext:

// Infrastructure/Persistence/ApplicationDbContext.cs
public class ApplicationDbContext : DbContext
{
    public ApplicationDbContext(DbContextOptions<ApplicationDbContext> options)
        : base(options)
    {
    }

    public DbSet<Order> Orders => Set<Order>();
    public DbSet<OutboxMessage> OutboxMessages => Set<OutboxMessage>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        base.OnModelCreating(modelBuilder);

        // Configure Outbox table
        modelBuilder.Entity<OutboxMessage>(entity =>
        {
            entity.ToTable("OutboxMessages");
            
            entity.HasKey(e => e.Id);
            
            entity.Property(e => e.EventType)
                .IsRequired()
                .HasMaxLength(500);
            
            entity.Property(e => e.EventData)
                .IsRequired();
            
            entity.Property(e => e.CreatedAt)
                .IsRequired();
            
            entity.HasIndex(e => e.ProcessedAt)
                .HasFilter("[ProcessedAt] IS NULL")
                .HasDatabaseName("IX_OutboxMessages_ProcessedAt_Pending");
            
            entity.HasIndex(e => e.CreatedAt)
                .HasDatabaseName("IX_OutboxMessages_CreatedAt");
        });

        // Configure Order entity
        modelBuilder.Entity<Order>(entity =>
        {
            entity.ToTable("Orders");
            entity.HasKey(e => e.Id);
            // ... other configurations
        });
    }
}

Important index notes:

Filtered index on ProcessedAt IS NULL for efficient querying of pending messages
Index on CreatedAt for processing messages in order

Step 3: Create the Outbox Event Publisher

Now let's create an implementation of IEventPublisher that writes to the outbox instead of publishing directly:

// Infrastructure/Events/OutboxEventPublisher.cs
public class OutboxEventPublisher : IEventPublisher
{
    private readonly ApplicationDbContext _dbContext;
    private readonly ILogger<OutboxEventPublisher> _logger;

    public OutboxEventPublisher(
        ApplicationDbContext dbContext,
        ILogger<OutboxEventPublisher> logger)
    {
        _dbContext = dbContext;
        _logger = logger;
    }

    public async Task PublishAsync<TEvent>(
        TEvent domainEvent, 
        CancellationToken cancellationToken = default) 
        where TEvent : IDomainEvent
    {
        // Serialize the event to JSON
        var eventData = JsonSerializer.Serialize(domainEvent, new JsonSerializerOptions
        {
            PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
            WriteIndented = false
        });

        // Create outbox message
        var outboxMessage = new OutboxMessage
        {
            Id = Guid.NewGuid(),
            EventType = typeof(TEvent).AssemblyQualifiedName ?? typeof(TEvent).FullName!,
            EventData = eventData,
            CreatedAt = DateTime.UtcNow,
            ProcessedAt = null,
            Error = null,
            RetryCount = 0
        };

        // Add to DbContext - will be saved when the transaction commits
        await _dbContext.OutboxMessages.AddAsync(outboxMessage, cancellationToken);

        _logger.LogInformation(
            "Event {EventType} with ID {EventId} added to outbox",
            typeof(TEvent).Name,
            domainEvent.EventId);
    }
}

Key points:

Events are serialized to JSON for storage
Store the fully qualified type name for deserialization later
No actual publishing happens here—just database insert
The insert participates in the current transaction

Step 4: Update OrderService to Use Outbox

Now our OrderService remains almost identical, but uses the outbox publisher:

// Application/Services/OrderService.cs
public class OrderService
{
    private readonly ApplicationDbContext _dbContext;
    private readonly IEventPublisher _eventPublisher; // This will be OutboxEventPublisher
    private readonly ILogger<OrderService> _logger;

    public OrderService(
        ApplicationDbContext dbContext,
        IEventPublisher eventPublisher,
        ILogger<OrderService> logger)
    {
        _dbContext = dbContext;
        _eventPublisher = eventPublisher;
        _logger = logger;
    }

    public async Task CreateOrderAsync(Order order, CancellationToken cancellationToken = default)
    {
        // Start a transaction
        await using var transaction = await _dbContext.Database.BeginTransactionAsync(cancellationToken);
        
        try
        {
            // 1. Save the order
            await _dbContext.Orders.AddAsync(order, cancellationToken);
            
            // 2. Publish event to outbox (adds OutboxMessage to context)
            await _eventPublisher.PublishAsync(new OrderCreatedEvent
            {
                OrderId = order.Id,
                CustomerId = order.CustomerId,
                TotalAmount = order.Total,
                Items = order.Items.Select(i => new OrderItem
                {
                    ProductId = i.ProductId,
                    Quantity = i.Quantity,
                    Price = i.Price
                }).ToList()
            }, cancellationToken);
            
            // 3. Save all changes in a single transaction
            await _dbContext.SaveChangesAsync(cancellationToken);
            
            // 4. Commit the transaction
            await transaction.CommitAsync(cancellationToken);
            
            _logger.LogInformation(
                "Order {OrderId} created and event added to outbox", 
                order.Id);
        }
        catch (Exception ex)
        {
            // Rollback on any error - nothing is saved
            await transaction.RollbackAsync(cancellationToken);
            _logger.LogError(ex, "Failed to create order {OrderId}", order.Id);
            throw;
        }
    }
}

What changed?

Now we explicitly use a transaction
Both the Order and OutboxMessage are saved together
Either both succeed or both fail (atomicity guaranteed)
No dual-write problem!

Step 5: Create the Outbox Processor

Now we need a background service that reads from the outbox and publishes events. This is where the actual message publishing happens:

// Infrastructure/BackgroundServices/OutboxProcessor.cs
public class OutboxProcessor : BackgroundService
{
    private readonly IServiceProvider _serviceProvider;
    private readonly ILogger<OutboxProcessor> _logger;
    private readonly TimeSpan _interval = TimeSpan.FromSeconds(5);

    public OutboxProcessor(
        IServiceProvider serviceProvider,
        ILogger<OutboxProcessor> logger)
    {
        _serviceProvider = serviceProvider;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("Outbox Processor started");

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await ProcessPendingMessagesAsync(stoppingToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error processing outbox messages");
            }

            await Task.Delay(_interval, stoppingToken);
        }

        _logger.LogInformation("Outbox Processor stopped");
    }

    private async Task ProcessPendingMessagesAsync(CancellationToken cancellationToken)
    {
        // Create a new scope for each batch
        await using var scope = _serviceProvider.CreateAsyncScope();
        var dbContext = scope.ServiceProvider.GetRequiredService<ApplicationDbContext>();
        var serviceBusPublisher = scope.ServiceProvider.GetRequiredService<IServiceBusPublisher>();

        // Fetch pending messages (not yet processed)
        var pendingMessages = await dbContext.OutboxMessages
            .Where(m => m.ProcessedAt == null)
            .OrderBy(m => m.CreatedAt)
            .Take(100) // Process in batches
            .ToListAsync(cancellationToken);

        if (!pendingMessages.Any())
        {
            return;
        }

        _logger.LogInformation(
            "Processing {Count} pending outbox messages", 
            pendingMessages.Count);

        foreach (var message in pendingMessages)
        {
            await ProcessMessageAsync(message, dbContext, serviceBusPublisher, cancellationToken);
        }
    }

    private async Task ProcessMessageAsync(
        OutboxMessage message,
        ApplicationDbContext dbContext,
        IServiceBusPublisher publisher,
        CancellationToken cancellationToken)
    {
        try
        {
            // Deserialize the event
            var eventType = Type.GetType(message.EventType);
            if (eventType == null)
            {
                throw new InvalidOperationException(
                    $"Could not resolve event type: {message.EventType}");
            }

            var domainEvent = JsonSerializer.Deserialize(
                message.EventData, 
                eventType,
                new JsonSerializerOptions
                {
                    PropertyNamingPolicy = JsonNamingPolicy.CamelCase
                });

            if (domainEvent == null)
            {
                throw new InvalidOperationException(
                    $"Could not deserialize event data for message {message.Id}");
            }

            // Publish to Service Bus
            await publisher.PublishAsync(domainEvent, eventType, cancellationToken);

            // Mark as processed
            message.ProcessedAt = DateTime.UtcNow;
            message.Error = null;

            _logger.LogInformation(
                "Successfully published outbox message {MessageId} of type {EventType}",
                message.Id,
                eventType.Name);
        }
        catch (Exception ex)
        {
            message.RetryCount++;
            message.Error = ex.Message;

            _logger.LogError(ex,
                "Failed to process outbox message {MessageId} (Retry {RetryCount})",
                message.Id,
                message.RetryCount);

            // Optional: implement max retry logic
            if (message.RetryCount >= 5)
            {
                _logger.LogWarning(
                    "Message {MessageId} has exceeded max retries and will be skipped",
                    message.Id);
                message.ProcessedAt = DateTime.UtcNow; // Mark as processed to stop retrying
            }
        }
        finally
        {
            // Save changes (processed or error state)
            await dbContext.SaveChangesAsync(cancellationToken);
        }
    }
}

Key features:

Runs continuously in the background
Processes messages in batches (100 at a time)
Handles errors gracefully with retry logic
Marks messages as processed after successful publishing
Uses a new scope for each batch to avoid stale DbContext

Note: I keep some hardcode parameters in this example, but we can easily leverage the option pattern to make them set them up in the appsetting.json.

Step 6: Create Service Bus Publisher Interface

Now that our IEventPublisher publish in the outbox table, we need a separate interface for publishing to Service Bus:

// Infrastructure/Messaging/IServiceBusPublisher.cs
public interface IServiceBusPublisher
{
    Task PublishAsync(object domainEvent, Type eventType, CancellationToken cancellationToken = default);
}

// Infrastructure/Messaging/AzureServiceBusPublisher.cs
public class AzureServiceBusPublisher : IServiceBusPublisher
{
    private readonly ServiceBusClient _serviceBusClient;
    private readonly ILogger<AzureServiceBusPublisher> _logger;

    public AzureServiceBusPublisher(
        ServiceBusClient serviceBusClient,
        ILogger<AzureServiceBusPublisher> logger)
    {
        _serviceBusClient = serviceBusClient;
        _logger = logger;
    }

    public async Task PublishAsync(
        object domainEvent, 
        Type eventType, 
        CancellationToken cancellationToken = default)
    {
        // Derive topic name from event type
        var topicName = eventType.Name.ToLowerInvariant().Replace("event", "");
        
        var sender = _serviceBusClient.CreateSender(topicName);

        try
        {
            var eventData = JsonSerializer.Serialize(domainEvent, new JsonSerializerOptions
            {
                PropertyNamingPolicy = JsonNamingPolicy.CamelCase
            });

            var message = new ServiceBusMessage(eventData)
            {
                MessageId = Guid.NewGuid().ToString(),
                ContentType = "application/json",
                Subject = eventType.Name
            };

            // Add custom properties
            message.ApplicationProperties.Add("EventType", eventType.AssemblyQualifiedName);
            message.ApplicationProperties.Add("PublishedAt", DateTime.UtcNow);

            await sender.SendMessageAsync(message, cancellationToken);

            _logger.LogInformation(
                "Published event {EventType} to topic {TopicName}",
                eventType.Name,
                topicName);
        }
        finally
        {
            await sender.DisposeAsync();
        }
    }
}

Step 7: Register Services in Program.cs

Finally, wire everything together in your dependency injection configuration:

// Program.cs
var builder = WebApplication.CreateBuilder(args);

// Database
builder.Services.AddDbContext<ApplicationDbContext>(options =>
    options.UseSqlServer(
        builder.Configuration.GetConnectionString("DefaultConnection"),
        sqlOptions => sqlOptions.EnableRetryOnFailure()));

// Azure Service Bus
builder.Services.AddSingleton(_ => 
    new ServiceBusClient(builder.Configuration["Azure:ServiceBus:ConnectionString"]));

// Event Publishing
builder.Services.AddScoped<IEventPublisher, OutboxEventPublisher>();
builder.Services.AddScoped<IServiceBusPublisher, AzureServiceBusPublisher>();

// Background Services
builder.Services.AddHostedService<OutboxProcessor>();

// Application Services
builder.Services.AddScoped<OrderService>();

var app = builder.Build();

// Apply migrations
using (var scope = app.Services.CreateScope())
{
    var db = scope.ServiceProvider.GetRequiredService<ApplicationDbContext>();
    await db.Database.MigrateAsync();
}

app.Run();

Now your application is set up to use the Outbox Pattern for reliable event publishing! The last step is to create the outbox table you might create it manually or create migration for it.

Step 8 (optionnal): Create EF Core Migration

Create the database migration for the outbox table:

dotnet ef migrations add AddOutboxPattern
dotnet ef database update

This will create a migration that looks like:

public partial class AddOutboxPattern : Migration
{
    protected override void Up(MigrationBuilder migrationBuilder)
    {
        migrationBuilder.CreateTable(
            name: "OutboxMessages",
            columns: table => new
            {
                Id = table.Column<Guid>(nullable: false),
                EventType = table.Column<string>(maxLength: 500, nullable: false),
                EventData = table.Column<string>(nullable: false),
                CreatedAt = table.Column<DateTime>(nullable: false),
                ProcessedAt = table.Column<DateTime>(nullable: true),
                Error = table.Column<string>(nullable: true),
                RetryCount = table.Column<int>(nullable: false)
            },
            constraints: table =>
            {
                table.PrimaryKey("PK_OutboxMessages", x => x.Id);
            });

        migrationBuilder.CreateIndex(
            name: "IX_OutboxMessages_ProcessedAt_Pending",
            table: "OutboxMessages",
            column: "ProcessedAt",
            filter: "[ProcessedAt] IS NULL");

        migrationBuilder.CreateIndex(
            name: "IX_OutboxMessages_CreatedAt",
            table: "OutboxMessages",
            column: "CreatedAt");
    }
}

Final Thoughts

The outbox pattern might seem like overkill when you're building a simple monolith, but it's essential for distributed systems. The beauty of the approach we've shown is that you can start simple with in-memory events and gradually evolve to the outbox pattern as your system grows.

Remember: premature optimization is the root of all evil, but planned evolution is the root of sustainable architecture.

Start with what you need today, but design for what you'll need tomorrow.

See you for more on www.devskillsunlock.com