The Outbox Pattern in Rust for Reliable Event Publishing

Write the database change and the event it should publish in one transaction, then publish from an outbox table with a relay. Fixes the dual write problem in Rust with SQLx.

4 min read

Summarize in AI

Quick Summary

When a service updates its database and also publishes a message, the two can disagree. The outbox pattern writes the event to an outbox table in the same transaction as the data change, and a relay reads unsent rows with FOR UPDATE SKIP LOCKED, publishes them and marks them sent. Delivery is at least once.

Key takeaways

The dual write problem: committing to the database and publishing to a broker are two operations that cannot be atomic together.
The outbox row commits or rolls back with the business data, so an event exists if and only if the change does.
A relay polls (or follows the write-ahead log) and publishes; it can crash and resume without losing events.
SELECT ... FOR UPDATE SKIP LOCKED lets several relays share the table without double-sending.
At-least-once delivery means duplicates are possible; consumers dedupe on event id.

A typical service handles a request by changing its database and telling the world: "order placed". These are two systems, and no transaction spans both. Whichever order you pick, a crash in the gap leaves them inconsistent. The outbox pattern removes the gap. Everything compiles on Rust 1.90 with edition 2024, sqlx 0.8 and tokio 1 on PostgreSQL 17.

The problem

rust

// tempting, and wrong
async fn place_order(db: &sqlx::PgPool, broker: &Broker, order: NewOrder) -> Result<(), Error> {
    insert_order(db, &order).await?;
    broker.publish("order.placed", &order).await?; // crash before this line: event lost
    Ok(())
}

Reversing the two lines fails the other way: the event goes out and the insert fails. Retrying does not help without more state.

The idea

Save the event as a row, in the same transaction as the change. Publishing becomes a separate, retryable job that reads those rows.

migrations/0005_outbox.sqlSQL

CREATE TABLE outbox (
    id          BIGSERIAL PRIMARY KEY,
    topic       TEXT NOT NULL,
    payload     JSONB NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
    sent_at     TIMESTAMPTZ
);
CREATE INDEX outbox_unsent ON outbox (id) WHERE sent_at IS NULL;

The partial index keeps polling cheap as the table grows.

Writing atomically

rust

use sqlx::PgPool;

pub async fn place_order(pool: &PgPool, customer: i64, total_cents: i64) -> sqlx::Result<i64> {
    let mut tx = pool.begin().await?;

    let order_id: i64 = sqlx::query_scalar(
        "INSERT INTO orders (customer_id, total_cents) VALUES ($1, $2) RETURNING id",
    )
    .bind(customer)
    .bind(total_cents)
    .fetch_one(&mut *tx)
    .await?;

    sqlx::query("INSERT INTO outbox (topic, payload) VALUES ('order.placed', $1)")
        .bind(serde_json::json!({ "order_id": order_id, "total_cents": total_cents }))
        .execute(&mut *tx)
        .await?;

    tx.commit().await?;
    Ok(order_id)
}

If the commit fails, neither row exists. If it succeeds, both do. The request path never touches the broker, so a broker outage cannot fail the order (transactions).

The relay

rust

use sqlx::PgPool;
use std::time::Duration;

async fn publish(_topic: &str, _payload: &serde_json::Value) -> Result<(), String> {
    Ok(()) // send to Kafka, NATS, SQS, a webhook...
}

pub async fn relay_once(pool: &PgPool) -> sqlx::Result<usize> {
    let mut tx = pool.begin().await?;
    let rows: Vec<(i64, String, serde_json::Value)> = sqlx::query_as(
        "SELECT id, topic, payload FROM outbox
         WHERE sent_at IS NULL
         ORDER BY id
         LIMIT 100
         FOR UPDATE SKIP LOCKED",
    )
    .fetch_all(&mut *tx)
    .await?;

    let mut done = Vec::new();
    for (id, topic, payload) in &rows {
        if publish(topic, payload).await.is_ok() {
            done.push(*id);
        } else {
            break; // keep order: stop at the first failure
        }
    }

    sqlx::query("UPDATE outbox SET sent_at = now() WHERE id = ANY($1)")
        .bind(&done)
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;
    Ok(done.len())
}

pub async fn relay_loop(pool: PgPool) {
    loop {
        match relay_once(&pool).await {
            Ok(0) | Err(_) => tokio::time::sleep(Duration::from_millis(500)).await,
            Ok(_) => {}
        }
    }
}

FOR UPDATE SKIP LOCKED locks the chosen rows for this transaction and lets other relay instances skip them, so you can run several without sending twice in the normal case. If the process crashes after publishing and before the commit, the rows stay unsent and the next run publishes them again: a duplicate. That is the at-least-once guarantee.

Consumers must dedupe

Each event needs a stable id (the outbox id works) so consumers can ignore repeats:

SQL

CREATE TABLE processed_events (event_id BIGINT PRIMARY KEY);
-- consumer: INSERT INTO processed_events VALUES ($1) ON CONFLICT DO NOTHING; skip if 0 rows

Do the dedupe insert and the consumer's own change in one transaction, as with idempotency keys. Ordering is preserved per relay only if you stop at the first failure and use one relay (or partition by key).

Operating it

**Polling interval versus latency:** 500 ms is simple; LISTEN/NOTIFY can wake the relay immediately, with polling as a safety net (LISTEN/NOTIFY).
**Cleanup:** delete or archive rows older than a retention period, in batches.
**Monitoring:** alert on the age of the oldest unsent row and on table size.
**Poison events:** a row that always fails blocks later ones if you stop at failure; add an attempt counter and a dead-letter state.
**Alternative to polling:** change data capture reads the write-ahead log (for example with Debezium) and removes the relay code, at the cost of more infrastructure.

The outbox is the usual companion of event sourcing and CQRS when read models are updated asynchronously.

Also inDatabases

Taggedsqlx outbox