# Scaling the Outbox Pattern (2B+ messages per day)

> Learn how to supercharge your Outbox pattern implementation, scaling to 32,500 messages per second. Through strategic optimizations in database queries, message publishing, and parallel processing, I'll show you how to handle over 2.8 billion messages daily while maintaining system reliability.

Published: 2024-10-12. Author: Milan Jovanović.

Canonical: https://milanjovanovic.tech/blog/scaling-the-outbox-pattern

A covering index on the unprocessed-message query, batched publishes, one batched `UPDATE`, and five parallel workers took my Outbox processor from 1,350 to about 32,500 messages per second.
That is over 2.8 billion messages per day.
Ordering is no longer guaranteed, and consumers still have to be idempotent.

In last week's newsletter, I talked about [implementing the Outbox pattern](https://milanjovanovic.tech/blog/implementing-the-outbox-pattern).
It's a crucial tool for reliable distributed messaging.
But implementing it is just the first step.

The real challenge? Scaling it to handle massive message volumes.

Today, we're taking it up a notch.
We'll start with a basic Outbox processor and transform it into a high-performance engine capable of handling over 2 billion messages daily.

Let's dive in!

## Starting Point

This is our starting point.
We have an `OutboxProcessor` that polls for unprocessed messages and publishes them to a queue.
The first few things we can tweak are the **frequency** and **batch size**.

```csharp
internal sealed class OutboxProcessor(NpgsqlDataSource dataSource, IPublishEndpoint publishEndpoint)
{
    private const int BatchSize = 1000;

    public async Task<int> Execute(CancellationToken cancellationToken = default)
    {
        await using var connection = await dataSource.OpenConnectionAsync(cancellationToken);
        await using var transaction = await connection.BeginTransactionAsync(cancellationToken);

        var messages = await connection.QueryAsync<OutboxMessage>(
            @"""
            SELECT *
            FROM outbox_messages
            WHERE processed_on_utc IS NULL
            ORDER BY occurred_on_utc LIMIT @BatchSize
            """,
            new { BatchSize },
            transaction: transaction);

        foreach (var message in messages)
        {
            try
            {
                var messageType = Messaging.Contracts.AssemblyReference.Assembly.GetType(message.Type);
                var deserializedMessage = JsonSerializer.Deserialize(message.Content, messageType);

                await publishEndpoint.Publish(deserializedMessage, messageType, cancellationToken);

                await connection.ExecuteAsync(
                    @"""
                    UPDATE outbox_messages
                    SET processed_on_utc = @ProcessedOnUtc
                    WHERE id = @Id
                    """,
                    new { ProcessedOnUtc = DateTime.UtcNow, message.Id },
                    transaction: transaction);
            }
            catch (Exception ex)
            {
                await connection.ExecuteAsync(
                    @"""
                    UPDATE outbox_messages
                    SET processed_on_utc = @ProcessedOnUtc, error = @Error
                    WHERE id = @Id
                    """,
                    new { ProcessedOnUtc = DateTime.UtcNow, Error = ex.ToString(), message.Id },
                    transaction: transaction);
            }
        }

        await transaction.CommitAsync(cancellationToken);

        return messages.Count;
    }
}
```

Let's assume that we run the `OutboxProcessor` continuously.
I increased that batch size to `1000`.

How many messages are we able to process?

I'll run the Outbox processing for 1 minute and count how many messages were processed.

The baseline implementation processed **81,000** messages in one minute or **1,350 MPS** (messages per second).

Not bad, but let's see how much we can improve this.

## Measuring Each Step

You can't improve what you can't measure. Right?
So, I'll use a `Stopwatch` to measure the total execution time and the time each step takes.

Notice that I also split the publish and update steps.
It's so I can measure the time for publishing and updating separately.
This will be important later because I want to optimize each step separately.

With the baseline implementation, here are the execution times for each step:

- Query time: ~70ms
- Publish time: ~320ms
- Update time: ~300ms

```csharp {10,11,26,56,70,75}
internal sealed class OutboxProcessor(
    NpgsqlDataSource dataSource,
    IPublishEndpoint publishEndpoint,
    ILogger<OutboxProcessor> logger)
{
    private const int BatchSize = 1000;

    public async Task<int> Execute(CancellationToken cancellationToken = default)
    {
        var totalStopwatch = Stopwatch.StartNew();
        var stepStopwatch = new Stopwatch();

        await using var connection = await dataSource.OpenConnectionAsync(cancellationToken);
        await using var transaction = await connection.BeginTransactionAsync(cancellationToken);

        stepStopwatch.Restart();
        var messages = (await connection.QueryAsync<OutboxMessage>(
            @"""
            SELECT *
            FROM outbox_messages
            WHERE processed_on_utc IS NULL
            ORDER BY occurred_on_utc LIMIT @BatchSize
            """,
            new { BatchSize },
            transaction: transaction)).AsList();
        var queryTime = stepStopwatch.ElapsedMilliseconds;

        var updateQueue = new ConcurrentQueue<OutboxUpdate>();

        stepStopwatch.Restart();
        foreach (var message in messages)
        {
            try
            {
                var messageType = Messaging.Contracts.AssemblyReference.Assembly.GetType(message.Type);
                var deserializedMessage = JsonSerializer.Deserialize(message.Content, messageType);

                await publishEndpoint.Publish(deserializedMessage, messageType, cancellationToken);

                updateQueue.Enqueue(new OutboxUpdate
                {
                    Id = message.Id,
                    ProcessedOnUtc = DateTime.UtcNow
                });
            }
            catch (Exception ex)
            {
                updateQueue.Enqueue(new OutboxUpdate
                {
                    Id = message.Id,
                    ProcessedOnUtc = DateTime.UtcNow,
                    Error = ex.ToString()
                });
            }
        }
        var publishTime = stepStopwatch.ElapsedMilliseconds;

        stepStopwatch.Restart();
        foreach (var outboxUpdate in updateQueue)
        {
            await connection.ExecuteAsync(
                @"""
                UPDATE outbox_messages
                SET processed_on_utc = @ProcessedOnUtc, error = @Error
                WHERE id = @Id
                """,
                outboxUpdate,
                transaction: transaction);
        }
        var updateTime = stepStopwatch.ElapsedMilliseconds;

        await transaction.CommitAsync(cancellationToken);

        totalStopwatch.Stop();
        var totalTime = totalStopwatch.ElapsedMilliseconds;

        OutboxLoggers.Processing(logger, totalTime, queryTime, publishTime, updateTime, messages.Count);

        return messages.Count;
    }

    private struct OutboxUpdate
    {
        public Guid Id { get; init; }
        public DateTime ProcessedOnUtc { get; init; }
        public string? Error { get; init; }
    }
}
```

Now, onto the fun part!

## Optimizing Read Queries

The first thing I want to optimize is the query for fetching unprocessed messages.
Performing a `SELECT *` query will have an impact if we don't need all the columns (hint: we don't).

Here's the current SQL query:

```sql
SELECT *
FROM outbox_messages
WHERE processed_on_utc IS NULL
ORDER BY occurred_on_utc LIMIT @BatchSize
```

We can modify the query to return only the columns we need.
This will save us some bandwidth but will not significantly improve performance.

```sql
SELECT id AS Id, type AS Type, content as Content
FROM outbox_messages
WHERE processed_on_utc IS NULL
ORDER BY occurred_on_utc LIMIT @BatchSize
```

Let's examine the execution plan for this query.
You'll see it's performing a table scan.
I'm running this on PostgreSQL, and here's what I get from `EXPLAIN ANALYZE`:

```
Limit  (cost=86169.40..86286.08 rows=1000 width=129) (actual time=122.744..124.234 rows=1000 loops=1)
  ->  Gather Merge  (cost=86169.40..245080.50 rows=1362000 width=129) (actual time=122.743..124.198 rows=1000 loops=1)
        Workers Planned: 2
        Workers Launched: 2
        ->  Sort  (cost=85169.38..86871.88 rows=681000 width=129) (actual time=121.478..121.492 rows=607 loops=3)
              Sort Key: occurred_on_utc
              Sort Method: top-N heapsort  Memory: 306kB
              Worker 0:  Sort Method: top-N heapsort  Memory: 306kB
              Worker 1:  Sort Method: top-N heapsort  Memory: 306kB
              ->  Parallel Seq Scan on outbox_messages  (cost=0.00..47830.88 rows=681000 width=129) (actual time=0.016..67.481 rows=666667 loops=3)
                    Filter: (processed_on_utc IS NULL)
Planning Time: 0.051 ms
Execution Time: 124.298 ms
```

Now, I'll create an index that "covers" the query for fetching unprocessed messages.
A covered index contains all the columns needed to satisfy a query without accessing the table itself.

The index will be on the `occurred_on_utc` and `processed_on_utc` columns.
It will include the `id`, `type`, and `content` columns.
Lastly, we'll apply a filter to index unprocessed messages only.

```sql
CREATE INDEX IF NOT EXISTS idx_outbox_messages_unprocessed
ON public.outbox_messages (occurred_on_utc, processed_on_utc)
INCLUDE (id, type, content)
WHERE processed_on_utc IS NULL
```

Let me explain the reasoning behind each decision:

- Indexing the `occurred_on_utc` will store the entries in the index in ascending order.
  This matches the `ORDER BY occurred_on_utc` statement in the query.
  This means the query can scan the index without sorting the results.
  The results are already in the correct sort order.
- Including the columns we select in the index allows us to return them from the index entry.
  This avoids reading the values from the table rows.
- Filtering for unprocessed messages in the index satisfies the `WHERE processed_on_utc IS NULL` statement.

**Caveat**: PostgreSQL has a maximum index row size of **2712B** (don't ask how I know).
The columns in the `INCLUDE` list are also part of the index row (B-tree tuple).
The `content` column contains the serialized JSON message, so it's the most likely culprit to make us exceed this limit.
There's no way around it, so my advice is to keep your messages as small as possible.
You could exclude this column from the `INCLUDE` list for a minor performance hit.

Here's the updated execution plan after creating this index:

```
Limit  (cost=0.43..102.82 rows=1000 width=129) (actual time=0.016..0.160 rows=1000 loops=1)
  ->  Index Only Scan using idx_outbox_messages_unprocessed on outbox_messages  (cost=0.43..204777.36 rows=2000000 width=129) (actual time=0.015..0.125 rows=1000 loops=1)
        Heap Fetches: 0
Planning Time: 0.059 ms
Execution Time: 0.189 ms
```

Because we have a covered index, the execution plan only contains an `Index Only Scan` and `Limit` operation.
There's no filtering or sorting that needs to happen, which is why we see a massive performance improvement.

What's the performance impact on the query time?

- Query time: 70ms → 1ms **(-98.5%)**

## Optimizing Message Publishing

The next thing we can optimize is how we're publishing messages to the queue.
I'm using the `IPublishEndpoint` from [**MassTransit**](https://milanjovanovic.tech/blog/using-masstransit-with-rabbitmq-and-azure-service-bus) to publish to RabbitMQ.

To be more precise here, we're publishing to an exchange.
The exchange will then route the message to the appropriate queue.

But how can we optimize this?

A micro-optimization we can do is introduce a cache for the message types used in serialization.
Performing reflection constantly for every message type is expensive, so we'll do the reflection once, and store the result.

```csharp
var messageType = Messaging.Contracts.AssemblyReference.Assembly.GetType(message.Type);
```

The cache can be a `ConcurrentDictionary`, and we'll use `GetOrAdd` to retrieve the cached types.

I'll extract this piece of code to the `GetOrAddMessageType` helper method:

```csharp
private static readonly ConcurrentDictionary<string, Type> TypeCache = new();

private static Type GetOrAddMessageType(string typeName)
{
    return TypeCache.GetOrAdd(
        typeName,
        name => Messaging.Contracts.AssemblyReference.Assembly.GetType(name));
}
```

This is what our message publishing step looks like.
The biggest problem is we're waiting for the `Publish` to complete by awaiting it.
The `Publish` takes some time because it's waiting for confirmation from the message broker.
We're doing this in a loop, which makes it even less efficient.

```csharp
var updateQueue = new ConcurrentQueue<OutboxUpdate>();

foreach (var message in messages)
{
    try
    {
        var messageType = Messaging.Contracts.AssemblyReference.Assembly.GetType(message.Type);
        var deserializedMessage = JsonSerializer.Deserialize(message.Content, messageType);

        // We're waiting for the message broker confirmation here.
        await publishEndpoint.Publish(deserializedMessage, messageType, cancellationToken);

        updateQueue.Enqueue(new OutboxUpdate
        {
            Id = message.Id,
            ProcessedOnUtc = DateTime.UtcNow
        });
    }
    catch (Exception ex)
    {
        updateQueue.Enqueue(new OutboxUpdate
        {
            Id = message.Id,
            ProcessedOnUtc = DateTime.UtcNow,
            Error = ex.ToString()
        });
    }
}
```

We can improve this by publishing the messages in a batch.
In fact, the `IPublishEndpoint` has a `PublishBatch` extension method.
If we peek inside, here's what we'll find:

```csharp
// MassTransit implementation
public static Task PublishBatch(
    this IPublishEndpoint endpoint,
    IEnumerable<object> messages,
    CancellationToken cancellationToken = default)
{
    return Task.WhenAll(messages.Select(x => endpoint.Publish(x, cancellationToken)));
}
```

So we can transform the collection of messages into a list of publishing tasks that we can await using `Task.WhenAll`.

```csharp {7}
var updateQueue = new ConcurrentQueue<OutboxUpdate>();

var publishTasks = messages
    .Select(message => PublishMessage(message, updateQueue, publishEndpoint, cancellationToken))
    .ToList();

await Task.WhenAll(publishTasks);

// I extracted the message publishing into a separate method for readability.
private static async Task PublishMessage(
    OutboxMessage message,
    ConcurrentQueue<OutboxUpdate> updateQueue,
    IPublishEndpoint publishEndpoint,
    CancellationToken cancellationToken)
{
    try
    {
        var messageType = GetOrAddMessageType(message.Type);
        var deserializedMessage = JsonSerializer.Deserialize(message.Content, messageType);

        await publishEndpoint.Publish(deserializedMessage, messageType, cancellationToken);

        updateQueue.Enqueue(new OutboxUpdate
        {
            Id = message.Id,
            ProcessedOnUtc = DateTime.UtcNow
        });
    }
    catch (Exception ex)
    {
        updateQueue.Enqueue(new OutboxUpdate
        {
            Id = message.Id,
            ProcessedOnUtc = DateTime.UtcNow,
            Error = ex.ToString()
        });
    }
}
```

What's the improvement for the message publishing step?

- Publish time: 320ms → 289ms **(-9.8%)**

As you can see, it's not significantly faster.
But this is needed for us to benefit from other optimizations I have in store.

## Optimizing Update Queries

The next step in our optimization journey is addressing the query updating the processed Outbox messages.

The current implementation is inefficient because we send one query to the database for each Outbox message.

```csharp
foreach (var outboxUpdate in updateQueue)
{
    await connection.ExecuteAsync(
        @"""
        UPDATE outbox_messages
        SET processed_on_utc = @ProcessedOnUtc, error = @Error
        WHERE id = @Id
        """,
        outboxUpdate,
        transaction: transaction);
}
```

If you didn't get the memo by now, batching is the name of the game.
We want a way to send one large `UPDATE` query to the database.

We have to construct the SQL for this batch query manually.
We'll use the `DynamicParameters` type from [**Dapper**](https://milanjovanovic.tech/blog/dapper-dotnet-guide) to provide all the parameters.

```csharp
var updateSql =
    @"""
    UPDATE outbox_messages
    SET processed_on_utc = v.processed_on_utc,
        error = v.error
    FROM (VALUES
        {0}
    ) AS v(id, processed_on_utc, error)
    WHERE outbox_messages.id = v.id::uuid
    """;

var updates = updateQueue.ToList();
var paramNames = string.Join(",", updates.Select((_, i) => $"(@Id{i}, @ProcessedOn{i}, @Error{i})"));

var formattedSql = string.Format(updateSql, paramNames);

var parameters = new DynamicParameters();

for (int i = 0; i < updates.Count; i++)
{
    parameters.Add($"Id{i}", updates[i].Id.ToString());
    parameters.Add($"ProcessedOn{i}", updates[i].ProcessedOnUtc);
    parameters.Add($"Error{i}", updates[i].Error);
}

await connection.ExecuteAsync(formattedSql, parameters, transaction: transaction);
```

This will produce a SQL query that looks something like this:

```sql
UPDATE outbox_messages
SET processed_on_utc = v.processed_on_utc,
    error = v.error
FROM (VALUES
    (@Id0, @ProcessedOn0, @Error0),
    (@Id1, @ProcessedOn1, @Error1),
    (@Id2, @ProcessedOn2, @Error2),
    -- A few hundred rows in beteween
    (@Id999, @ProcessedOn999, @Error999)
) AS v(id, processed_on_utc, error)
WHERE outbox_messages.id = v.id::uuid
```

Instead of sending one update query per message, we can send one query to update all messages.

This will obviously give us a noticeable performance benefit:

- Update time: 300ms → 52ms **(-82.6%)**

## How Far Did We Get?

Let's test out the performance improvement with the current optimizations.
The changes we made so far focus on improving the speed of the `OutboxProcessor`.

Here are the rough numbers I'm seeing for the individual steps:

- Query time: ~**1ms**
- Publish time: ~**289ms**
- Update time: ~**52ms**

I'll run the Outbox processing for 1 minute and count the number of processed messages.

The optimized implementation processed **162,000** messages in one minute or **2,700 MPS**.

For reference, this allows us to process more than 230 million messages per day.

But we're just getting started.

## Parallel Outbox Processing

If we want to take this further, we have to scale out the `OutboxProcessor`.
The problem we could face here is processing the same message more than once.
So, we need to implement some form of locking on the current batch of messages.

PostgreSQL has a convenient `FOR UPDATE` statement that we can use here.
It will lock the selected rows for the duration of the current transaction.
However, we must add the `SKIP LOCKED` statement to allow other queries to skip the locked rows.
Otherwise, any other query will be blocked until the current transaction is completed.

Here's the updated query:

```sql {5}
SELECT id AS Id, type AS Type, content as Content
FROM outbox_messages
WHERE processed_on_utc IS NULL
ORDER BY occurred_on_utc LIMIT @BatchSize
FOR UPDATE SKIP LOCKED
```

To scale out the `OutboxProcessor`, we simply run multiple instances of the background job.

I'll simulate this using `Parallel.ForEachAsync`, where I can control the `MaxDegreeOfParallelism`.

```csharp
var parallelOptions = new ParallelOptions
{
    MaxDegreeOfParallelism = _maxParallelism,
    CancellationToken = cancellationToken
};

await Parallel.ForEachAsync(
    Enumerable.Range(0, _maxParallelism),
    parallelOptions,
    async (_, token) =>
    {
        await ProcessOutboxMessages(token);
    });
```

We can process **179,000** messages in one minute or **2,983 MPS** with five (5) workers.

I thought this was supposed to be _much_ faster. What gives?

Without parallel processing, we were able to get ~2,700 MPS.

A new **bottleneck** appears: publishing the messages in batches.

The publish time went from ~289ms to ~1,540ms.

Interestingly, if you multiply the base publish time (for one worker) by the number of workers, you roughly get to the new publish time.

We're wasting a lot of time waiting for the acknowledgment from the message broker.

How can we fix this?

## Batching Message Publishing

<div className="note-panel">
  **Note**: `ConfigureBatchPublish` was deprecated with `MassTransit.RabbitMQ`
  v8.3.2. This version uses the new `RabbitMQ.Client` v7, which was rewritten to
  use the TPL and async/await. You will see a similar performance improvement
  just from upgrading to this version.
</div>

RabbitMQ supports publishing messages in batches.
We can enable this feature when configuring MassTransit by calling the `ConfigureBatchPublish` method.
MassTransit will buffer messages before sending them to RabbitMQ, to increase throughput.

```csharp {7-10}
builder.Services.AddMassTransit(x =>
{
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(builder.Configuration.GetConnectionString("Queue"), hostCfg =>
        {
            hostCfg.ConfigureBatchPublish(batch =>
            {
                batch.Enabled = true;
            });
        });

        cfg.ConfigureEndpoints(context);
    });
});
```

With only this small change, let's rerun our test with five workers.

This time around, we're able to process **1,956,000** messages in one minute.

Which gives us a blazing ~**32,500 MPS**.

This is more than 2.8 billion processed messages per day.

I could call it a day here, but there's one more thing I want to show you.

## Turning Off Publisher Confirmation (Dangerous)

One more thing you _can_ do (**which I don't recommend**) is turn off publisher confirmation.
This means that calling `Publish` won't wait until the message is confirmed by the broker (ack'd).
It could lead to **reliability issues** and potentially **losing messages**.

That being said, I did manage to get ~37,000 MPS with publisher confirmation turned off.

```csharp
cfg.Host(builder.Configuration.GetConnectionString("Queue"), hostCfg =>
{
    hostCfg.PublisherConfirmation = false; // Dangerous. I don't recommend it.
    hostCfg.ConfigureBatchPublish(batch =>
    {
        batch.Enabled = true;
    });
});
```

## Key Considerations for Scaling

While we've achieved impressive throughput, consider these factors when implementing these techniques in a real-world system:

1. **Consumer Capacity**:
   Can your consumers keep up?
   Boosting producer throughput without matching consumer capacity can create backlogs.
   Consider the entire pipeline when scaling.

2. **Delivery Guarantees**:
   Our optimizations maintain at-least-once delivery.
   Design consumers to be [**idempotent**](https://milanjovanovic.tech/blog/idempotent-consumer-handling-duplicate-messages) to handle occasional duplicate messages.

3. **Message Ordering**:
   Parallel processing with `FOR UPDATE SKIP LOCKED` may cause out-of-order messages.
   For strict ordering, consider the **Inbox pattern** on the consumer side to buffer messages.
   An Inbox allows us to process messages in the correct order, even if they arrive out of sequence.

4. **Reliability vs. Performance Trade-offs**:
   Turning off publisher confirmation increases speed but risks message loss.
   Weigh performance against reliability based on your specific needs.

By addressing these factors, you'll create a high-performance Outbox processor that integrates smoothly with your system architecture.

## Summary

We've come a long way from our initial Outbox processor.
Here's what we accomplished:

1. Optimized database queries with smart indexing
2. Improved message publishing with batching
3. Streamlined database updates with batching
4. Scaled out Outbox processing with parallel workers
5. Leveraged RabbitMQ's batch publishing feature

The result?
We boosted processing from 1,350 messages per second to an impressive **32,500 MPS**.
That's over 2.8 billion messages per day!

Scaling isn't just about raw speed - it's about identifying and addressing bottlenecks at each step.
By measuring, optimizing, and rethinking our approach, we achieved massive performance gains.

That's all for today. Hope this was helpful.

**P.S.** You can find the [source code](https://github.com/m-jovanovic/outbox-scaling) here.

---

## Frequently asked questions

### How many messages per second can the outbox pattern handle?

In this benchmark, a baseline outbox processor handled 1,350 messages per second. After adding a covering index, batching publishes and updates, running five parallel workers, and enabling RabbitMQ batch publishing, it reached about 32,500 messages per second, which is over 2.8 billion messages per day.

### How do you optimize the outbox query for unprocessed messages?

Create a covering index on occurred_on_utc and processed_on_utc that includes the id, type, and content columns and filters on processed_on_utc IS NULL. That turns a parallel sequential scan plus sort into an index-only scan, and in this test it cut query time from about 70ms to 1ms.

### How do you run multiple outbox processors in parallel without duplicate processing?

Add FOR UPDATE SKIP LOCKED to the SELECT query. FOR UPDATE locks the selected rows for the duration of the transaction, and SKIP LOCKED lets other workers skip those rows instead of blocking, so each worker claims its own batch of messages.

### Why should outbox update queries be batched?

Sending one UPDATE per message costs a database round trip for every row. Building a single UPDATE with a VALUES list updates the whole batch in one query, which reduced the update step from about 300ms to 52ms in this test, an 82.6 percent improvement.

### How does RabbitMQ batch publishing improve outbox throughput?

MassTransit could buffer messages before sending them to RabbitMQ via ConfigureBatchPublish, which took this benchmark from about 2,983 to 32,500 messages per second with five workers. That option was deprecated in MassTransit.RabbitMQ 8.3.2, where the rewritten async RabbitMQ.Client v7 delivers a similar improvement by default.

### Does scaling the outbox pattern preserve message ordering?

Not strictly. Parallel processing with FOR UPDATE SKIP LOCKED can produce out-of-order messages, so for strict ordering consider the inbox pattern on the consumer side to buffer and reorder them. Delivery remains at-least-once, so consumers must stay idempotent. Disabling publisher confirmation adds speed but risks losing messages.