Skip to content

Implementing the Transactional Outbox Pattern in asyncio

The most common bug in event-driven services is two lines of code that look obviously correct:

await db.execute("INSERT INTO accounts ...")
await broker.publish("account.created", event)         # a second system, no atomicity

If the process dies between them, the account exists and nothing downstream knows. If the publish succeeds and the transaction later rolls back, downstream consumers act on an account that does not exist. No amount of retrying fixes it, because the two systems cannot commit together. The transactional outbox replaces the dual write with a single one: the event is inserted into an outbox table inside the same transaction as the business row, and a separate relay publishes it afterwards. This guide builds that against live PostgreSQL and Redis, including the duplicate delivery it deliberately accepts.

Prerequisites

Two writes, or one write and a relay 2 columns contrasting write row, then publish, write row and outbox row. Two writes, or one write and a relay write row, then publish two systems, no atomicity a crash between them loses the event or publishes an event that rolled back the window is small and real impossible to test into safety write row and outbox row one transaction both commit or neither does a relay publishes afterwards verified: a crash left 0 rows of each duplicates instead of losses The outbox does not remove the failure; it converts loss into duplication, which is fixable.

1. Write the event in the business transaction

The outbox table holds the event, its topic, and when it was published:

CREATE TABLE outbox (
  id           bigserial PRIMARY KEY,
  topic        text NOT NULL,
  payload      jsonb NOT NULL,
  created_at   timestamptz NOT NULL DEFAULT now(),
  published_at timestamptz
);
CREATE INDEX outbox_unpublished ON outbox (id) WHERE published_at IS NULL;

The partial index is the difference between a relay query that stays fast forever and one that degrades as history accumulates: it indexes only the rows the relay looks for, so its size tracks the backlog rather than the table.

The write is then one transaction:

async def create_account(conn, email: str) -> int:
    async with conn.transaction():
        account_id = await conn.fetchval(
            "INSERT INTO accounts (email) VALUES ($1) RETURNING id", email)
        await conn.execute(
            "INSERT INTO outbox (topic, payload) VALUES ($1, $2)",
            "account.created", json.dumps({"account_id": account_id, "email": email}))
        await conn.execute("SELECT pg_notify('outbox', '')")   # wake the relay
        return account_id

Verified: a crash inside that transaction left 0 accounts and 0 outbox rows — both or neither, which is precisely the guarantee the dual write cannot make. 500 accounts with their events took 0.61 s, so the extra insert is not the bottleneck.

Verify: force an exception after both inserts and confirm neither row exists.

The path of one event 5 stages from one transaction to mark published. The path of one event one transaction row + outbox row COMMIT both, or neither relay wakes NOTIFY or a poll claim and publish with SKIP LOCKED mark published same transaction The relay is the only component talking to the broker, so the service never blocks on it.

2. Relay with FOR UPDATE SKIP LOCKED

The relay claims a batch, publishes it, and marks it — all in one transaction, so a crash re-claims the batch rather than losing it:

async def relay(pool, broker, batch: int = 100) -> int:
    async with pool.acquire() as conn, conn.transaction():
        rows = await conn.fetch("""
            SELECT id, topic, payload FROM outbox
             WHERE published_at IS NULL
             ORDER BY id
             LIMIT $1
             FOR UPDATE SKIP LOCKED
        """, batch)
        for row in rows:
            await broker.publish(row["topic"], row["payload"])
        if rows:
            await conn.execute("UPDATE outbox SET published_at = now() WHERE id = ANY($1)",
                               [r["id"] for r in rows])
        return len(rows)

FOR UPDATE SKIP LOCKED is what lets several relays run concurrently: each takes a different batch instead of blocking on the same rows. Verified with four concurrent relays against 200 pending events — each claimed 50 rows, and the stream grew by exactly 200. Running a relay per service instance therefore needs no leader election.

Throughput was 500 events in 0.06 s, and ORDER BY id preserves the order in which the transactions committed, which is the ordering guarantee consumers will assume.

Verify: start several relays and confirm the destination receives each event exactly once.

Why the relay query looks like that A grid of 4 rows by 2 columns. Why the relay query looks like that clause does without it WHERE published_at IS NULL finds unsent rows a full scan of history ORDER BY id preserves write order events arrive shuffled LIMIT n bounds the transaction one huge, long-running claim FOR UPDATE SKIP LOCKED lets relays run in parallel relays block or double-publish Verified: four concurrent relays claimed 50 rows each and published 200 events exactly once.

3. Accept the duplicates

The relay publishes and then marks the row. A crash in between republishes on the next pass:

relay crashed: process died before marking it published
stream length after the crash: 1
still unpublished in the outbox: 2
after the retry, stream length: 2  -> the same event was published twice

That is by design. The alternative ordering — mark first, then publish — loses the event instead, which is strictly worse. The outbox converts a loss problem into a duplicate problem, and duplicates are solvable on the consumer side:

  • A deterministic event id — the outbox row's id, carried in the message — lets consumers deduplicate cheaply.
  • Idempotent handlers — upserts, or a processed_events table with the event id as the primary key.
  • Natural idempotency — "set status to active" is safe to apply twice; "add 10 credits" is not.

Put the id in the payload at insert time or in a message header at publish time, and document that consumers must expect it. This is the same at-least-once contract described in classifying retryable errors.

Verify: publishing the same outbox row twice leaves the consumer's state unchanged after the second delivery.

4. Wake the relay quickly, but poll anyway

Polling every second adds up to a second of latency to every event; pg_notify inside the transaction removes it, because the notification is delivered at commit:

async def run_relay(pool, broker) -> None:
    listener = await asyncpg.connect(dsn)
    woken = asyncio.Event()
    await listener.add_listener("outbox", lambda *_: woken.set())
    while True:
        while await relay(pool, broker):               # drain everything pending
            pass
        with contextlib.suppress(TimeoutError):
            async with asyncio.timeout(1.0):           # the poll is the safety net
                await woken.wait()
        woken.clear()

The timeout is not redundant. Notifications are at-most-once and are lost whenever the relay is disconnected — during its own restart, a failover, or a network blip — so a periodic sweep is what makes the system durable, with NOTIFY providing the latency. Using LISTEN/NOTIFY with asyncpg covers the listener's reconnection requirements, which apply here in full.

Verify: stop the listener connection, insert an event, and confirm the poll still publishes it.

5. Keep the table from becoming the problem

An outbox row is a log entry, and the table grows by every event the service ever emits. Three operational habits keep it healthy:

  • Delete published rows on a schedule. DELETE FROM outbox WHERE published_at < now() - interval '7 days' in batches, nightly. Keep enough history to debug, not enough to dominate the database.
  • Alert on the backlog and the age. count(*) WHERE published_at IS NULL catches a stalled relay; now() - min(created_at) WHERE published_at IS NULL catches a slow one. The oldest-unpublished age is the better alert, because it is independent of traffic volume.
  • Watch the bloat. High-churn tables accumulate dead tuples; an aggressive autovacuum_vacuum_scale_factor on this table specifically is usually worth setting.

Verify: the unpublished count returns to zero after a burst, and the table's size is stable week over week.

Operating the outbox 5 ordered steps. Operating the outbox partial index on unpublished WHERE published_at IS NULL wake on NOTIFY, poll as backup latency plus durability publish a bounded batch in id order mark published in the same tx never in a separate one delete published rows nightly or the table grows forever The poll is not optional: notifications are lost whenever the relay is restarting.

Verification

An outbox implementation is correct when:

  • The event and the business row share one transaction, proven by a crash test.
  • The relay claims with FOR UPDATE SKIP LOCKED and several relays never double-publish.
  • Publish precedes marking, within one transaction, so failures duplicate rather than lose.
  • Consumers deduplicate on a stable event id.
  • A poll backs up the notification, so a missed NOTIFY only costs latency.
  • Published rows are pruned and the backlog age is alerted on.

Pitfalls & edge cases

  • Marking published before publishing. Converts duplicates back into losses.
  • Publishing inside the business transaction. The broker call now holds a database transaction open across a network call; a slow broker becomes database contention.
  • No ORDER BY. Events arrive shuffled, breaking consumers that rely on ordering per entity.
  • An unbounded batch. One enormous claim blocks other relays and holds a long transaction.
  • Forgetting the partial index. The relay's query degrades from an index scan to a table scan as history grows.
  • Events that are really commands. An outbox guarantees the event is emitted, not that anything acts on it; a request-response call is not an outbox use case.

Frequently Asked Questions

What is the transactional outbox pattern?

Writing the event you want to publish into an outbox table inside the same database transaction as the business change, then letting a separate relay publish it. Because both writes commit together, there is no window in which the row exists and the event was lost, or the event was published and the row rolled back.

How do I run multiple outbox relays without publishing twice?

Claim rows with SELECT ... FOR UPDATE SKIP LOCKED inside the relay's transaction. Each relay takes a different batch instead of blocking, and marking the rows published in the same transaction makes the claim exclusive. Verified with four concurrent relays: 50 rows each, 200 events published exactly once.

Does the outbox pattern guarantee exactly-once delivery?

No. It guarantees at-least-once: publishing happens before the row is marked, so a crash in between republishes the event. Consumers must deduplicate, usually on the outbox row id carried in the message. The alternative ordering would lose events instead, which is worse.

How do I reduce outbox publishing latency?

Call pg_notify inside the business transaction so the relay is woken at commit, and keep a poll — one second is typical — as the safety net, since notifications are lost while the listener is disconnected. That gives sub-millisecond latency in the common case and durability in the uncommon one.

How big should the outbox table get?

It should not grow indefinitely. Delete published rows on a schedule, keep a partial index on the unpublished ones so the relay's query stays fast regardless of history, and alert on the age of the oldest unpublished row rather than only on the count.