Transactional Outbox with Wolverine

A transactional outbox exists so a committed fact and the message about that fact cannot drift apart.

This guide is that guarantee in Wolverine, with Marten on PostgreSQL. The message is written in the same transaction as the event or the document. A durability agent sends it after the commit. Consumers still see duplicates. The dead-letter table is something you operate.

How the handler decides to emit the message is in CQRS in .NET with Wolverine. What the event means inside the stream is in Marten Event Sourcing in Production. The same commit can hold a document instead of an event. That document is Marten as a Document Database. The design of that context is Marten, Polecat, and Wolverine architecture. The hands-on version, on your domain, is the Marten and Wolverine workshop.

The failure the pattern removes

Two writes want to succeed together: the database, and the broker. They will not. If you commit the order and then publish NotifyWarehouse, the process can die in the gap. The order is shipped and the warehouse is never told. If you publish first and then commit, the warehouse starts work and the commit rolls back. The warehouse is picking an order that does not exist.

Neither order is safe. The transactional outbox makes them one write. The outgoing message is a row in PostgreSQL, inserted with the business change. The broker is a later step, and it is allowed to be down.

One PostgreSQL transaction

Three things have to be on, or you still have two writes.

Marten has to be the message store. IntegrateWithWolverine, from the WolverineFx.Marten package, registers the inbox and outbox tables with Marten's schema management, starts the durability agent, and adds the transactional middleware. Marten and Wolverine without that package do not include the method. Pass a schema name when you want those tables under their own name.

The endpoint has to be durable. Middleware stages a message only when the destination is enrolled. A local queue enrolls with UseDurableLocalQueues or UseDurableInbox on that queue. A broker enrolls with UseDurableOutbox on the publishing endpoint. A RabbitMQ queue that is only buffered sends during the handler. The database can still roll back.

The handler has to use the session Wolverine owns. The middleware commits that session after the method returns. A session you open yourself, and save yourself, is a second transaction.

builder.Services.AddMarten(opts =>
{
    opts.Connection(connectionString);
})
.IntegrateWithWolverine();

builder.Host.UseWolverine(opts =>
{
    opts.Policies.UseDurableLocalQueues();

    opts.PublishMessage<NotifyWarehouse>()
        .ToRabbitQueue("warehouse")
        .UseDurableOutbox();

    opts.ListenToRabbitQueue("warehouse")
        .UseDurableInbox();
});

UseDurableLocalQueues is the in-process half. The CQRS guide is where that queue is tuned. UseDurableOutbox is the half that keeps NotifyWarehouse in PostgreSQL until RabbitMQ accepts it. I turn on the integration and the endpoint before I talk about retries.

The return is the outbox

The handler returns the event and the message. It does not publish on its own, and it does not call SaveChangesAsync. Wolverine appends the event, stages the message, and commits both.

On an aggregate handler the return value is an event unless you say otherwise. A tuple of (OrderShipped, NotifyWarehouse) appends both to the stream. The warehouse message never leaves. OutgoingMessages is the type that means "publish this, do not append it."

[AggregateHandler]
public static (Events, OutgoingMessages) Handle(ShipOrder command, Order order)
{
    if (order.Shipped)
        throw new InvalidOperationException("Already shipped.");

    var events = new Events();
    events += new OrderShipped(command.OrderId, command.ShippedAt);

    var messages = new OutgoingMessages();
    messages.Add(new NotifyWarehouse(command.OrderId));

    return (events, messages);
}

Throw before the return and nothing is staged. That is the refusal. Order here is the stream aggregate from the event sourcing guide. [AggregateHandler] loads it. This guide does not.

The same commit works for a document. There is no stream, so there is no Events collection. The returned message is an ordinary cascade, and the durable endpoint puts it in the outbox.

public static NotifyWarehouse Handle(ShipOrder command, IDocumentSession session)
{
    session.Store(new Shipment(command.OrderId, command.ShippedAt));
    return new NotifyWarehouse(command.OrderId);
}

Store versus Insert is the document guide. The outbox only cares that this session is the one the middleware commits. In a controller, where there is no middleware, the longhand is IMartenOutbox.PublishAsync and then your own SaveChangesAsync. Inside a handler, that second save commits early, and a message returned afterwards can miss it.

What is stored

Before the commit, the envelope is intent. After the commit, it is a row that says the message must leave. The agent marks it sent and deletes it, or it leaves it for another attempt. You do not delete that row from application code because the HTTP request finished. The request finishing is not the send.

Three tables, created with the Marten schema:

• wolverine_outgoing_envelopes. Waiting to be sent. A growing pile means the agent is down, the broker is refusing, or a row is stuck on a node that died. owner_id of 0 means any node may send it. OutboxStaleTime bumps old owned rows back to 0. It is opt-in, because it changes the schema. I set it in hours, from the longest send I am willing to wait, and I do not set it in seconds.
• wolverine_incoming_envelopes. The inbox. A handled row stays so a redelivery of the same envelope id is recognized, then the agent deletes it. The window is Durability.KeepAfterMessageHandling, and the default is five minutes. Shorter than the broker's redelivery and the handler runs twice. Longer and the table is a log. I leave five minutes until I have seen a duplicate arrive later than that.
• wolverine_dead_letters. Will not be sent again until a person acts.

I alert on the first and the third. A handled inbox row inside its five minutes is the design.

The consumer still sees duplicates

The outbox gives you at-least-once delivery. The agent can send, crash before it records the send, and send again. The broker can deliver twice. The inbox records the envelope id and drops a duplicate it has already executed. That covers the transport duplicate. It does not cover two different messages that mean the same business fact.

The handler has to tolerate both. A second NotifyWarehouse for an order that was already told is a no-op. Charging a customer twice is the error. The envelope id saves you from the retry. The order id saves you from a second command.

public static class NotifyWarehouseHandler
{
    public static void Configure(HandlerChain chain)
    {
        chain.OnException<DocumentAlreadyExistsException>()
            .Discard();
    }

    public static async Task Handle(NotifyWarehouse message, IDocumentSession session)
    {
        var notice = await session.LoadAsync<WarehouseNotice>(message.OrderId);
        if (notice is not null)
            return;

        session.Insert(new WarehouseNotice(message.OrderId));
    }
}

WarehouseNotice uses the order id as its document id. The load covers the ordinary retry. Two deliveries at the same moment can both see nothing. Insert only stages the write. The middleware commits after Handle returns, and the second commit fails with DocumentAlreadyExistsException. A catch around Insert never sees that exception.

The other delivery already stored the notice. I discard the exception on this handler chain, so the default policy does not move the message to wolverine_dead_letters. A process-wide discard would also swallow a first insert that failed because the document was already there for a real reason.

If this handler also sends mail, the send is a second system. Record the notice in this transaction, and let a later message do the mail. The mailer has to be idempotent on the order id as well. A send that happens before the commit, and a commit that then rolls back, is the original bug one level down.

Dead letters and cleanup

Retries are for a broker that blinked. A handler that rejects the same message every time is a poison message. It will fill the log and the CPU. After the retry policy is exhausted, Wolverine moves it to the error queue. With Marten as the message store, that queue is wolverine_dead_letters. Someone replays it after the bug is fixed, or discards it. Nobody should learn about it from a customer.

I keep the retry count small. A message that fails the same way three times will fail a fourth. Move it, page, and fix the handler.

Expiration of that table is off by default. When you turn on Durability.DeadLetterQueueExpirationEnabled, the retention is DeadLetterQueueExpiration, and the default of that value is ten days. I enable it once someone is actually reading the table. Until then, expiration hides messages nobody replayed. I still do not keep the table as the audit log. Audit belongs in the Marten stream, or in the document you already decided to keep. The outbox is a queue. A queue that never shrinks will slow the agent that is supposed to drain it.

Two checks I put on a dashboard, and only these: depth of wolverine_outgoing_envelopes, and depth of wolverine_dead_letters. Depth of the event stream is a different page.

The test is the rollback

The tracked session in the CQRS guide proves that a handler returned NotifyWarehouse. It does not prove the outbox. A host with no PostgreSQL has nothing to roll back.

The test I add here has two facts, and the host is one fixture with Marten, built once. An order that is already shipped: ShipOrder throws, and wolverine_outgoing_envelopes has no row for that attempt. An open order: the handler returns, and the same table has the envelope until the agent sends it. IncludeExternalTransports() waits on RabbitMQ. I turn that on only for the test that has a broker. The rollback fact is local to PostgreSQL.

Testcontainers belongs in this test. The cascade of return values does not need it, and that test stays in the other guide.