Transactional Outbox Pattern in Python: Never Lose an Event

A customer places order 101. The row commits. Then your process dies before it tells Kafka, and the confirmation email never goes out. The warehouse never hears about it either, and nothing in your logs says why. That’s the dual write problem, and the transactional outbox pattern is how you close it.

Updated 4 November 2026 · Tested with Python 3.12, PostgreSQL 18.6, psycopg 3.3.6, SQLAlchemy 2.0.30, Kafka 4.3.1

Part 3 of the Reliable Python services series · ← Rate limiting algorithms

The transactional outbox pattern in Python works like this. You write the event into an outbox table, in the same Postgres transaction as the business change. A separate relay reads that table and publishes to your broker. So no event is lost, and none is sent for a change that rolled back.

Simple, right? Most outbox tutorials think so. They stop at one polling loop and one worker, which is the half that works.

The trouble starts when you go further. Run a second copy for safety, and both publish the same rows. Add a reader that remembers id > last_seen, and it quietly skips rows. Keep the table small with DELETE, and dead rows pile up behind the relay’s query.

My position: the outbox table is the easy part. The relay decides whether your events arrive, so build it for several workers, at-least-once delivery and real cleanup from day one. By the end you’ll have:

• A schema and a SQLAlchemy write path that commit the event with the order.
• A psycopg 3 relay you can run on several workers without losing an event.
• Fixes for the two ordering traps, and for the cleanup that slows every claim.
• A clear answer on when LISTEN/NOTIFY is worth it, and when to hand the job to Debezium.
• Seven real dual write failures and fixes, from GitLab and Wix to Netflix, and a checklist for before you ship.

We’ll follow order 101 through every part of it. Along the way you’ll find out how many events a second relay “for safety” sends twice, whether eight workers beat four, and what a NOTIFY trigger really costs. Guess before each answer.

Jargon alert

• Relay: the process that reads the outbox table and publishes to the broker.
• At-least-once: every event arrives, and some arrive twice.
• SKIP LOCKED: a locking read that skips rows another transaction holds instead of waiting.
• xmin: the oldest transaction still running; everything older has finished.
• CDC: change data capture, reading changes from the write-ahead log (Debezium does this).

Why can’t you just publish after commit()?

Because Postgres and your broker don’t share a transaction. Commit first, and a crash before the publish loses the event. That’s order 101. Publish first, and a failed commit sends an event for an order that never existed. Whichever way round you do it, there’s a gap between the two calls, and one day a process dies inside it.

So retry the publish? That doesn’t help: the process that would retry is the one that crashed. Two-phase commit across Postgres and Kafka isn’t an option today either. Kafka’s KIP-939 adds it, but as of November 2026 it hasn’t shipped. The latest release is Kafka 4.4, and it’s aimed at 4.5.

There’s a second retry to plan for: the HTTP request. If a client times out and resends POST /orders, even a perfect outbox writes two orders and two events. Store idempotency keys for the API that triggers the write in the same transaction.

Who hit the dual write in production, and how did they fix it?

If this still sounds like a whiteboard problem, meet seven teams that wrote it up, some after it bit them and some before it could. Their fixes fall into three groups. Spot which one the outbox is in.

GitLab, Rails and the job that ran before its commit

If you’ve worked on a Rails app, you may know this one. Sidekiq’s FAQ has an entry for it: why you’d see a lot of Can't find ModelName with ID=12345 errors. The job ran before the transaction that created the model committed, and went looking for a row that didn’t exist yet.

GitLab went further. Its codebase raises an error in development and tests whenever a Sidekiq job is enqueued inside a transaction.

Then Rails addressed it in the framework. The pull request behind enqueue_after_transaction_commit, which shipped in Rails 7.2 in August 2024, calls enqueuing from a transaction “a fairly common mistake”. That setting waits for the commit, and drops the job if the transaction rolls back.

Look at where each fix puts the send: after COMMIT. Sidekiq’s transactional_push pushes a job to Redis only once the transaction commits. GitLab points developers to an after_commit hook, and Rails defers. That’s the trade Celery’s delay_on_commit() makes below: the timing is fixed, the gap isn’t.

But moving the send after COMMIT only swaps one failure for the other. As Brandur Leach described in 2017, a crash in between leaves the data saved and the work never done.

Wix: the payment that saved and the event that didn’t

Wix’s Natan Silnitsky put this first in a 2022 list of event-driven pitfalls. A service saves the payment status, then produces “payment completed” to Kafka. If the database or Kafka is down, “There could be situations where only one of the actions actually happens.” In his example, inventory drifts away from the real orders.

Wix uses two fixes. A resilient producer falls back to S3, and a separate service replays those messages into Kafka later, at the cost of out-of-order events. Debezium reads the MySQL binlog instead, and keeps order.

Five more teams, one line each

• DoorDash (2021): its search index drifted from the source of truth, so events now carry only an ID and the indexer re-reads the current state. Backfilling every store went from a week to 6.5 hours.
• Netflix (Delta, 2019): a second store, like a search index, drifts whenever the second write fails. Delta streams CDC events through Kafka instead.
• Airbnb (SpinalTap, 2018): dual writes can’t keep commit order or transaction boundaries, so SpinalTap streams the binlog into Kafka.
• Zalando (2022): an outbox on DynamoDB Streams, with a Python Lambda as the relay and a dead-letter queue for failures.
• SeatGeek (2025): no outbox table at all. Each event goes into the Postgres WAL inside the business transaction, and Debezium relays it. The client library came first in Python.

Keep DoorDash’s trick in mind: when events can arrive late or twice, send the ID and read the latest state. It comes back in the ordering section.

What every fix has in common

Now sort them. Sidekiq, GitLab, Rails and Celery move the send after the commit. Wix’s S3 fallback and DoorDash’s forced reindex make a lost event recoverable. Zalando, SeatGeek, Netflix and Airbnb put the event inside the commit: an outbox row, a WAL message or the database’s own log. Only that last group closes the gap.

That last group is where the outbox belongs. It removes the second call instead of moving it. The event row commits inside the order’s transaction, so a rollback leaves no row, and the relay only ever sees committed rows.

Brandur’s fix was the same idea: a staged_jobs table, drained by one enqueuer at a time. Here SKIP LOCKED replaces the single enqueuer, so several workers can drain the same table.

How does the transactional outbox pattern work?

Three parts, each with one job:

• The write puts the business row and the event row in one transaction.
• The relay reads unsent rows, publishes them, waits for the broker’s ack and marks them sent. Because it polls the table, it’s called a polling publisher; a relay that reads the write-ahead log instead is CDC, compared later.
• The consumer handles each event, and copes with seeing one twice. The outbox delivers at least once, not exactly once.

The name comes from the office out-tray. You finish the paperwork and drop the letter in the tray in one move, and the mail room collects it later. If the mail room is shut for an hour, the letter waits. It doesn’t vanish.

Here’s order 101 again, this time with an outbox. The order row and its event row commit together. The process dies a millisecond later? Fine. The row waits in the table until a relay picks it up.

Why only at least once? The relay publishes, then commits its “sent” mark. If it dies in between, the rows go out again, so order 101’s event can reach Kafka twice. The kill -9 test below shows how often. That’s why the row’s id travels with every message: it’s the consumer’s dedup key.

Outbox vs inbox: which side does each one fix?

The outbox fixes the producer: the service that changes data and must tell others. The inbox fixes the consumer. It records each processed message id in the same transaction as the side effect, and skips repeats. You usually want both, since the outbox is what hands the consumer those repeats.

The inbox, also called the idempotent consumer, is the next post in the series. Sagas, which coordinate a multi-step process across services, usually run on an outbox too.

Do you even need the transactional outbox pattern?

Not always. Before you build one, ask:

• Does anything outside this database need to know? If the only reader is the same Postgres, a query or a trigger is enough.
• What does a lost event cost? A cache invalidation that expires in a minute anyway can fire after commit and accept the rare miss. An order, a payment or an email can’t.
• Is the change in one database? The outbox needs the business row and the event row in one transaction. A write across two databases needs a saga, with an outbox in each service.
• Do you already stream the business tables with CDC? If consumers can live with your schema, you may not need a separate table.
• Can your consumers handle duplicates? If not, fix that first. The outbox will hand them some.

There are other ways out of the dual write, too. Listen-to-yourself publishes first, then consumes its own event to update the database. Durable execution engines handle it for you.

Still here? Then let’s build it, starting with the table.

What goes in the outbox table (and how Python writes to it)

Use Debezium’s five columns: id, aggregatetype, aggregateid, type and payload. Then add the relay’s bookkeeping: created_at, published_at, attempts, next_attempt_at, last_error and dead_at. Keep Debezium’s names, and a later move to CDC is a connector setting instead of a migration. From schema.sql:

CREATE TABLE outbox (
    id              uuid        NOT NULL DEFAULT uuidv7(),  -- time-ordered, built into PG 18
    aggregatetype   text        NOT NULL,                   -- becomes the topic: outbox.event.<aggregatetype>
    aggregateid     text        NOT NULL,                   -- becomes the message key
    type            text        NOT NULL,
    payload         jsonb       NOT NULL,
    txid            xid8        NOT NULL DEFAULT pg_current_xact_id(),  -- see "commit-order gap"
    created_at      timestamptz NOT NULL DEFAULT now(),
    published_at    timestamptz,
    attempts        integer     NOT NULL DEFAULT 0,
    next_attempt_at timestamptz NOT NULL DEFAULT now(),
    last_error      text,
    dead_at         timestamptz,                            -- set after too many failed attempts
    PRIMARY KEY (id, created_at)                            -- must include the partition key
) PARTITION BY RANGE (created_at);

-- Only rows the relay still has to send live in this index, so it stays small
-- however many published rows the partitions keep for replay.
CREATE INDEX outbox_pending ON outbox (next_attempt_at)
    WHERE published_at IS NULL AND dead_at IS NULL;

-- Rows that failed too often, copied here before their partition is dropped.
CREATE TABLE outbox_dead (LIKE outbox);

• uuidv7() is built into PostgreSQL 18 and time-ordered, so new ids land at the end of the index like a bigserial would. A test checks uuid_extract_version(id) = 7.
• Daily partitions make cleanup a DROP. Postgres wants the partition key in every unique constraint, hence the (id, created_at) key.
• The partial index holds only rows still waiting, so it stays small however long you keep published ones.
• Create partitions days ahead. Without one, the insert fails with no partition of relation "outbox" found for row (SQLSTATE 23514). It takes the order down with it.

Ignore txid until the ordering section. Here’s the write side, with SQLAlchemy 2.0 (write_path.py without its __main__ block):

import uuid
from datetime import datetime

from sqlalchemy import create_engine, func, text
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column


class Base(DeclarativeBase):
    pass


class Order(Base):
    __tablename__ = "orders"
    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, server_default=text("uuidv7()"))
    customer_id: Mapped[str]
    total_cents: Mapped[int]


class OutboxEvent(Base):
    __tablename__ = "outbox"
    # Only the columns the app sets; the relay's bookkeeping columns have defaults.
    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, server_default=text("uuidv7()"))
    created_at: Mapped[datetime] = mapped_column(primary_key=True, server_default=func.now())
    aggregatetype: Mapped[str]
    aggregateid: Mapped[str]
    type: Mapped[str]
    payload: Mapped[dict] = mapped_column(JSONB)


def emit(session: Session, aggregatetype: str, aggregateid: str, event_type: str,
         payload: dict) -> None:
    """Queue an event in the caller's transaction. It commits or rolls back with it."""
    session.add(OutboxEvent(aggregatetype=aggregatetype, aggregateid=aggregateid,
                            type=event_type, payload=payload))


def place_order(engine, customer_id: str, total_cents: int) -> uuid.UUID:
    with Session(engine) as session, session.begin():  # one transaction, committed on exit
        order = Order(customer_id=customer_id, total_cents=total_cents)
        session.add(order)
        session.flush()  # sends the INSERT, so Postgres has assigned order.id
        order_id = order.id
        emit(session, "order", str(order_id), "OrderPlaced",
             {"order_id": str(order_id), "customer_id": customer_id, "total_cents": total_cents})
    return order_id

Notice what’s missing: the request path never talks to Kafka, so a slow broker can’t fail an order. And the tests show a commit leaves both rows, while an exception inside the block leaves neither, even after both INSERTs reached Postgres.

Try it: python -m pytest tests/test_write_path.py -q. Every “Try it” runs in the repo for this post, after the two setup commands in its README. Every number in this post comes from my own runs, so yours will differ, and these scripts show you by how much.

Django and Celery: why task.delay() inside a transaction bites

delay() sends the task the moment you call it. A worker can run it before your commit and not find the order, or run it after a rollback.

Celery 5.4’s delay_on_commit() waits for Django’s on_commit hook, which fixes the timing. But the send still happens in your process after the commit. If the process dies in that gap, the task is gone.

The outbox version writes a row inside atomic() and lets a relay call apply_async():

from django.db import transaction
from django.utils import timezone

from shop.models import Order, OutboxEvent
from shop.tasks import send_order_email


# Instead of this
def place_order_delay(customer_id: str, total_cents: int) -> Order:
    with transaction.atomic():
        order = Order.objects.create(customer_id=customer_id, total_cents=total_cents)
        send_order_email.delay(order.id)  # may run before COMMIT, or after a rollback
        return order


# Or this: runs after COMMIT, but is lost if the process dies right after it
def place_order_on_commit(customer_id: str, total_cents: int) -> Order:
    with transaction.atomic():
        order = Order.objects.create(customer_id=customer_id, total_cents=total_cents)
        send_order_email.delay_on_commit(order.id)
        return order


# Use this: the event row commits with the order
def place_order(customer_id: str, total_cents: int) -> Order:
    with transaction.atomic():
        order = Order.objects.create(customer_id=customer_id, total_cents=total_cents)
        OutboxEvent.objects.create(
            aggregatetype="order", aggregateid=str(order.id), type="OrderPlaced",
            payload={"order_id": str(order.id), "total_cents": total_cents},
        )
        return order


# ...and a relay (a management command in a loop) turns rows into Celery tasks
def relay_batch(batch_size: int = 100) -> int:
    with transaction.atomic():
        events = list(
            OutboxEvent.objects.select_for_update(skip_locked=True)
            .filter(published_at__isnull=True, dead_at__isnull=True,
                    next_attempt_at__lte=timezone.now())
            .order_by("next_attempt_at")[:batch_size]
        )
        for event in events:
            # task_id = the outbox id, so the worker can drop a redelivered task
            send_order_email.apply_async(args=[event.payload["order_id"]], task_id=str(event.id))
        OutboxEvent.objects.filter(id__in=[e.id for e in events]).update(published_at=timezone.now())
        return len(events)

The outbox id doubles as the Celery task_id, so workers can drop redeliveries. On RabbitMQ, turn on publisher confirms too. One caveat: this file is only syntax-checked, because Django and Celery aren’t installed in my test environment.

An outbox relay that’s safe on several workers: FOR UPDATE SKIP LOCKED

Now the part that decides whether your events arrive. Claim a batch with FOR UPDATE SKIP LOCKED, publish it, wait for the acks, then commit, all in one transaction.

Picture a board of tickets. Each worker grabs one nobody is holding, and a worker that drops dead lets go of its tickets. So you can run as many copies as you like, with no leader election and no distributed lock.

First, the loop most tutorials stop at (naive_relay.py):

from __future__ import annotations

import psycopg

from relay import Event, Publisher


def naive_relay_batch(conn: psycopg.Connection, publisher: Publisher, batch_size: int = 100) -> int:
    # Instead of this: a plain SELECT, so a second relay sees the same rows
    rows = conn.execute(
        "SELECT id, aggregatetype, aggregateid, type, payload FROM outbox "
        "WHERE published_at IS NULL ORDER BY created_at LIMIT %s", (batch_size,)).fetchall()
    events = [Event(*row) for row in rows]
    if events:
        publisher.publish(events)
        conn.execute("UPDATE outbox SET published_at = now() WHERE id = ANY(%s)",
                     ([e.id for e in events],))
    return len(events)

One copy is fine. Quick quiz before the numbers: you start a second copy “for safety”. Over 10,000 events, how many go out twice? A few? A hundred? Guess.

All of them: 10,000 in the median of 5 runs, and 9,800 in the luckiest. The two copies read the same pending rows and walk through the table in lockstep, like two waiters taking the same order. Two SKIP LOCKED workers sent 0 twice (try it: python measure_naive.py). Here’s the core of relay.py:

from __future__ import annotations

import json
import logging
import time
from dataclasses import dataclass
from typing import Callable, Protocol, TextIO
from uuid import UUID

import psycopg
from confluent_kafka import KafkaError, KafkaException, Producer

MAX_ATTEMPTS = 10

# Claim and mark in one statement. The mark stays invisible until COMMIT, which
# only happens after the broker acked; a crash or error before that undoes it.
CLAIM = """
    WITH batch AS (
        SELECT id, created_at FROM outbox
        WHERE published_at IS NULL AND dead_at IS NULL AND next_attempt_at <= now()
        ORDER BY next_attempt_at
        LIMIT %s
        FOR UPDATE SKIP LOCKED
    )
    UPDATE outbox o SET published_at = now()
    FROM batch
    WHERE o.id = batch.id AND o.created_at = batch.created_at
    RETURNING o.id, o.aggregatetype, o.aggregateid, o.type, o.payload
"""

MARK_FAILED = """
    UPDATE outbox
    SET published_at = NULL,
        attempts = attempts + 1,
        last_error = %s,
        next_attempt_at = now() + least(2 ^ attempts, 300) * interval '1 second',
        dead_at = CASE WHEN attempts + 1 >= %s THEN now() END
    WHERE id = %s
"""

# If a relay hangs mid-batch, Postgres ends its session and releases the locks.
SESSION_OPTIONS = "-c idle_in_transaction_session_timeout=30s -c statement_timeout=10s"

log = logging.getLogger("outbox.relay")


@dataclass(frozen=True)
class Event:
    id: UUID
    aggregatetype: str
    aggregateid: str
    type: str
    payload: dict


class Publisher(Protocol):
    def publish(self, events: list[Event]) -> dict[UUID, str]:
        """Send the batch and wait for the broker's acks.

        Return {event id: error} for events the broker refused. Raise if the
        broker can't be reached: the whole batch then rolls back untouched.
        """
        ...


def relay_batch(conn: psycopg.Connection, publisher: Publisher, batch_size: int = 100) -> int:
    """Claim, publish and commit one batch in one transaction. Returns rows claimed."""
    with conn.transaction():
        rows = conn.execute(CLAIM, (batch_size,)).fetchall()
        events = sorted((Event(*row) for row in rows), key=lambda e: e.id)  # uuidv7: insert order
        if events:
            errors = publisher.publish(events)  # blocks until acked; row locks held
            for event_id, error in errors.items():
                conn.execute(MARK_FAILED, (error[:1000], MAX_ATTEMPTS, event_id))
    return len(events)

• One statement claims and marks. Nobody sees the mark until COMMIT, which runs only after the broker acked, so a crash or error before that undoes it.
• A refused event backs off (1 s, 2 s, 4 s, up to 5 minutes). It goes dead after 10 tries, so a poison row can’t block every batch.
• An outage burns no attempts. If publish() raises, the batch rolls back untouched. Otherwise a 15-minute Kafka outage would kill every pending event.

Why claim and mark in one statement? Because of the strangest number in my test runs.

In my test setup, any message over about 8 KB that the client sent to the database took 50 ms longer: 0.35 ms at 8,000 bytes, 50.01 ms at 9,000. The same queries run next to the database took under half a millisecond. A relay that sends a list of ids back pays that on every big batch. This one never sends them.

The Kafka publisher (the same file has the StubPublisher the benchmarks use):

class PublishError(Exception):
    """Some events weren't acked in time. The caller's transaction rolls the batch back."""

class KafkaPublisher:
    """Topic outbox.event.<aggregatetype> and key aggregateid, like Debezium's router."""

    def __init__(self, bootstrap_servers: str, topic_prefix: str = "outbox.event.",
                 timeout: float = 10.0):
        self.producer = Producer({
            "bootstrap.servers": bootstrap_servers,
            "enable.idempotence": True,  # acks=all; producer retries can't duplicate or reorder
            "linger.ms": 5,
            "message.timeout.ms": int(timeout * 1000),  # give up when the relay does
        })
        self.topic_prefix, self.timeout = topic_prefix, timeout

    def publish(self, events: list[Event]) -> dict[UUID, str]:
        acked: set[UUID] = set()
        refused: dict[UUID, str] = {}

        def on_delivery(err: KafkaError | None, event_id: UUID) -> None:
            if err is None:
                acked.add(event_id)
            elif not err.retriable() and err.code() != KafkaError._MSG_TIMED_OUT:
                refused[event_id] = err.str()  # the broker rejected this message

        for e in events:
            try:
                self.producer.produce(
                    self.topic_prefix + e.aggregatetype,
                    key=e.aggregateid,
                    value=json.dumps(e.payload),
                    headers={"id": str(e.id), "type": e.type},
                    on_delivery=lambda err, msg, event_id=e.id: on_delivery(err, event_id),
                )
            except KafkaException as exc:  # e.g. MSG_SIZE_TOO_LARGE, before it's sent
                refused[e.id] = str(exc)
        self.producer.flush(self.timeout + 1)  # every ack arrives before the caller commits
        missing = [e.id for e in events if e.id not in acked and e.id not in refused]
        if missing:
            raise PublishError(f"{len(missing)} of {len(events)} events not acked")
        return refused

enable.idempotence stops the producer’s own retries from duplicating. But a restarted relay is a new producer, and it sends its batch again.

The tests run against Kafka 4.3. A 2 MB payload is refused with MSG_SIZE_TOO_LARGE while its batch-mates go out, and an unreachable broker leaves every row pending with attempts = 0.

And the loop that runs it:

class TokenBucket:
    """Caps the publish rate, so a backlog after an outage doesn't flood the broker."""

    def __init__(self, rate: float, burst: float):
        self.rate, self.burst = rate, burst
        self.tokens, self.last = burst, time.monotonic()

    def take(self, n: float) -> None:
        while True:
            now = time.monotonic()
            self.tokens = min(self.burst, self.tokens + (now - self.last) * self.rate)
            self.last = now
            if self.tokens >= n:
                self.tokens -= n
                return
            time.sleep((n - self.tokens) / self.rate)


def run(dsn: str, publisher: Publisher, *, batch_size: int = 100, poll_interval: float = 0.1,
        max_per_second: float | None = None, stop: Callable[[], bool] = lambda: False) -> None:
    """Polling publisher: drain the table, sleep poll_interval, repeat."""
    bucket = TokenBucket(max_per_second, burst=batch_size) if max_per_second else None
    while not stop():
        try:
            with psycopg.connect(dsn, autocommit=True, options=SESSION_OPTIONS) as conn:
                while not stop():
                    claimed = relay_batch(conn, publisher, batch_size)
                    if bucket and claimed:
                        bucket.take(claimed)
                    if claimed < batch_size:
                        time.sleep(poll_interval)
        except Exception:
            log.exception("batch failed; its rows stay pending")
            time.sleep(1.0)  # broker or database down: back off, then reconnect

The catch: the transaction stays open while you publish, like holding the elevator door while you finish a phone call. Keep batches small, and let Postgres enforce a deadline.

With idle_in_transaction_session_timeout=30s, a stuck relay gets cut off and its rows unlock. A test does it with 1 s, and the server logs FATAL: terminating connection due to idle-in-transaction timeout.

If your broker takes seconds per batch, use a locked_until lease instead. And after an outage, max_per_second paces the backlog with a token bucket. A test checks that 1,000 rows at 500 a second take at least 1.7 s.

How many relay workers should you run?

Fewer than you’d think once batches get big. Let’s test it: 100,000 pending rows, and a stub publisher that waits about 2 ms per batch, like a broker round trip. Then 1, 2, 4 or 8 worker processes, five runs each. Place your bet first: at batch 100, do eight workers beat four?

They don’t. At batch 100, four workers did 3.4 times one worker’s 15,390 a second, and eight were slower than four. At batch 10 the broker wait dominates, and eight workers did 7.2 times what one did. At batch 500, two peaked at 63,139 a second. Eight managed 38,003, less than one worker’s 44,759.

Timing each step shows why. Publishing (2.4 to 2.7 ms) and the commit stayed flat. The claim slowed down: at batch 500, it went from 5.3 ms with one worker to 81.8 ms with eight. More workers means more elbowing past rows the others hold locked.

So start with two to four workers and batches of about 100, then measure. Across all 60 runs (6,000,000 events) there were 0 missing and 0 duplicates.

Then the crash test: four workers, batch 100, 800,000 rows and a real Kafka topic. Each run had ten hard kills (like kill -9), each followed by a fresh worker.

Over 5 runs and 50 kills: 0 missing, 2,800 duplicates (56 per kill), and never more than 2 copies of one event. Every run’s total was a multiple of 100, which fits the mechanism. A kill after Kafka’s ack re-sends that whole batch (try it: python measure_crash.py).

Does the outbox pattern keep events in order?

A relay that never loses an event can still deliver them out of order. You get order per aggregate if you design for it, and global order never for free. There are two separate problems here: several workers each publish whenever their batch is done, and ids are assigned at INSERT while rows only become visible at COMMIT.

Ordering across several workers

The SKIP LOCKED relay doesn’t know two rows belong to one order. In a test, worker A claims OrderCreated and stalls. Worker B skips A’s locked row and publishes OrderPaid, and the broker sees paid before created. For order per aggregate, add three things the relay here doesn’t do:

• Hash aggregateid into, say, 16 shards. Each shard is owned by one worker at a time via pg_try_advisory_lock(shard), which a crashed owner releases.
• Key Kafka messages by aggregateid (the publisher above does).
• When an aggregate’s event fails, hold back its later events until that one goes through.

But first ask yourself: does your consumer need order, or just the latest state? If each event carries a version number, the consumer can ignore anything older than what it has. Then order stops mattering. DoorDash’s ID-only events get there another way: the indexer always re-reads the latest state.

The commit-order gap: why id > last_seen loses rows

Because a later id can commit first. And yes, it’s order 101 again.

T1 takes id 101, T2 takes 102, and T2 commits. A reader asks for id > 100, gets 102 and moves its watermark to 102. Then T1 commits, and 101 sits behind the watermark forever. Think of a deli counter that calls “the number after the last one served” while ticket 101 is still choosing cheese.

The SKIP LOCKED relay is safe here, since it looks for published_at IS NULL: 101 just goes out after 102.

Watermark readers aren’t safe: replay tools, a second consumer of the table, a hand-rolled change tailer. UUIDv7 and timestamps don’t help, because they’re assigned at insert too. This two-connection test uses real uuidv7() ids and loses the row every run (excerpt from tests/test_ordering.py):

import psycopg

from conftest import wait_until
from db import ADMIN_DSN
from relay import CLAIM, Event, StubPublisher, relay_batch
from watermark import NaiveWatermark, SnapshotWatermark

INSERT = ("INSERT INTO outbox (aggregatetype, aggregateid, type, payload) "
          "VALUES ('order', %s, %s, '{}') RETURNING id")


def out_of_order_commits(dsn):
    """Open T1, then T2; T2 commits first. Returns (t1, id1, id2) with T1 still open."""
    t1 = psycopg.connect(dsn)
    t2 = psycopg.connect(dsn)
    id1 = t1.execute(INSERT, ("order-1", "OrderPlaced")).fetchone()[0]
    id2 = t2.execute(INSERT, ("order-2", "OrderPlaced")).fetchone()[0]
    assert id1 < id2  # uuidv7: the insert order
    t2.commit()
    t2.close()
    return t1, id1, id2


def test_naive_watermark_loses_the_row_that_commits_late(dsn, conn):
    reader = NaiveWatermark()
    t1, id1, id2 = out_of_order_commits(dsn)

    assert reader.poll(conn) == [id2]  # the watermark jumps past id1
    t1.commit()
    t1.close()
    assert reader.poll(conn) == []  # id1 is committed now, and never read
    assert conn.execute("SELECT count(*) FROM outbox WHERE id = %s", (id1,)).fetchone()[0] == 1

Try it: python -m pytest tests/test_ordering.py -q. The naive-watermark test passes only because the late row really does go missing, every run.

The fix stores the inserting transaction’s id with every row (txid xid8 DEFAULT pg_current_xact_id()). It reads only rows older than the oldest transaction still running, and moves the watermark on (txid, id):

from __future__ import annotations

from uuid import UUID

import psycopg


class NaiveWatermark:
    """Instead of this: WHERE id > last_seen. Loses rows that commit out of order."""

    def __init__(self) -> None:
        self.last_seen = UUID(int=0)

    def poll(self, conn: psycopg.Connection, limit: int = 100) -> list[UUID]:
        ids = [r[0] for r in conn.execute(
            "SELECT id FROM outbox WHERE id > %s ORDER BY id LIMIT %s",
            (self.last_seen, limit))]
        if ids:
            self.last_seen = ids[-1]
        return ids


class SnapshotWatermark:
    """Use this: only read rows whose transaction ended before every open one began."""

    def __init__(self) -> None:
        self.last_txid, self.last_id = "0", UUID(int=0)  # txid is xid8; keep it as text

    def poll(self, conn: psycopg.Connection, limit: int = 100) -> list[UUID]:
        rows = conn.execute(
            """
            SELECT txid, id FROM outbox
            WHERE (txid, id) > (%s::xid8, %s)
              AND txid < pg_snapshot_xmin(pg_current_snapshot())
            ORDER BY txid, id
            LIMIT %s
            """, (self.last_txid, self.last_id, limit)).fetchall()
        if rows:
            self.last_txid, self.last_id = rows[-1]
        return [r[1] for r in rows]

With the same two transactions, the snapshot reader gets nothing while T1 is open, then 101 and 102. Order 101 arrives late, but it arrives. A long transaction now costs latency instead of rows.

Which long transaction, though? I tested the sneaky case: a write transaction left open in a different database, one the reader never queries. The reader stalled anyway. Transaction ids are shared by every database in the cluster, so a long transaction anywhere on the server holds it back.

Can Postgres LISTEN/NOTIFY replace polling?

Polling every 100 ms is simple, but every event can wait up to 100 ms. So can Postgres tell the relay instead? As a wake-up call, yes. As the channel, no.

NOTIFY is a doorbell, not a parcel. It tells the relay “something committed”, and the relay still goes to the door and reads the table. Miss the doorbell, and the parcel is still on the step.

Why not send the event itself? The NOTIFY docs list the reasons:

• It isn’t durable. A notification reaches whoever is listening at commit time; a relay that’s down misses it.
• Payloads are small: shorter than 8000 bytes by default.
• The queue can fill. It’s 8 GB, and when it’s full, transactions calling NOTIFY fail at commit. A listener stuck in a long transaction blocks its cleanup, so it can break your writes.
• PgBouncer’s transaction pooling never supports LISTEN (NOTIFY works), so the listener needs a direct connection.

So the trigger sends an empty payload (notify.sql):

CREATE OR REPLACE FUNCTION outbox_notify() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
    PERFORM pg_notify('outbox', '');  -- empty payload: it's a wake-up, not the event
    RETURN NULL;
END $$;

CREATE OR REPLACE TRIGGER outbox_notify
    AFTER INSERT ON outbox
    FOR EACH STATEMENT EXECUTE FUNCTION outbox_notify();

And the relay drains first, then sleeps until a NOTIFY arrives or the fallback timer fires (wakeup.py):

from __future__ import annotations

import time
from typing import Callable

import psycopg

from relay import SESSION_OPTIONS, Publisher, log, relay_batch


def run_with_notify(dsn: str, publisher: Publisher, *, batch_size: int = 100,
                    fallback_poll: float = 1.0, stop: Callable[[], bool] = lambda: False) -> None:
    while not stop():
        try:
            with psycopg.connect(dsn, autocommit=True) as listener, \
                 psycopg.connect(dsn, autocommit=True, options=SESSION_OPTIONS) as conn:
                listener.execute("LISTEN outbox")
                while not stop():
                    # Drain first: rows committed while nobody listened sent no NOTIFY.
                    while relay_batch(conn, publisher, batch_size) == batch_size:
                        pass
                    # Sleep until a commit notifies us, or fallback_poll seconds pass.
                    # A NOTIFY that arrived during the drain is returned at once.
                    for _ in listener.notifies(timeout=fallback_poll, stop_after=1):
                        pass
        except Exception:
            log.exception("relay failed; reconnecting")
            time.sleep(1.0)

This needs psycopg 3.2.4 or later. Before that, notifications that arrived while you were draining were lost.

The tests cover the paths that matter. A commit wakes the relay within a second despite a 30-second fallback, and a rollback sends nothing. Rows committed while nobody listened go out at startup, the fallback poll catches a lost NOTIFY, and the relay survives losing its LISTEN connection.

What does it buy order 101? Four writers commit 200 orders a second between them. Latency runs from a writer calling COMMIT to the relay (batch 100) picking the event up. Five 10-second runs per mode:

What does a NOTIFY trigger cost?

Guess first: with 32 writers committing as fast as they can, how much throughput does the doorbell cost? Five percent? Twenty? Here are writer processes committing for 10 seconds, with and without the trigger (median of five runs):

85%. With one writer, the trigger cost nothing you’d notice, and with 8, commits per second fell by 61%. The reason is a lock. Up to PostgreSQL 18, a transaction that sent NOTIFY takes a global lock while it commits. Recall.ai hit this in production, and DBOS argues it’s manageable.

Notifying commits queue behind one another. So the database managed 940 a second with 8 writers and 927 with 32: a ceiling, not a slowdown.

Will PostgreSQL 19 fix this? Its draft release notes say NOTIFY will only wake backends that listen on the channel. They don’t say the lock is gone, so rerun this before you count on it.

So is it worth it? Here NOTIFY bought 84 ms at p99 and put a ceiling on every write that emits an event. If about 100 ms is fine for your consumers, poll every 100 ms and skip NOTIFY. Reach for it when you need tens of milliseconds, and your event writes stay well below that ceiling.

How do you clean up the outbox table? Drop partitions, don’t DELETE

Every published row stays in the table until you remove it, order 101’s included, and how you remove it matters. Drop whole days instead of deleting rows, because a DELETE doesn’t erase rows. It crosses them out.

Every removed row stays behind as a dead tuple until vacuum gets to it. The table keeps its size on disk, and one long-running transaction can stop vacuum from cleaning any of it. Dropping a partition removes a day’s rows, their index entries and their disk space in one quick step.

Let’s test it. Yesterday’s partition holds 1,000,000 published events, and today’s has 1,000 pending. Autovacuum is off on the old partition, so each step shows its own effect. After each step, the relay’s real CLAIM (batch 100) runs 20 times.

Everything runs twice: once on a quiet database, and once with a REPEATABLE READ transaction left open, like a long report. That transaction started before the rows were published.

Before you look: after deleting a million rows, how much disk do you get back?

None. DELETE left all 392 MB, VACUUM got it to 100 MB (the indexes stay bloated), and dropping the partition left 0.8 MB.

On the quiet database, that disk is most of what cleanup buys. The first claim after the publish took 295 ms (median), while it stepped over a million dead index entries once. After that, claims ran at about 3.9 ms (try it: python measure_cleanup.py).

With the long transaction open, every claim took 68 ms instead of about 2. DELETE didn’t help, and VACUUM barely did (52 ms, still 392 MB), because it can’t remove row versions that transaction might still read. The drop doesn’t care: 2.2 ms and 0.8 MB.

msdousti saw the same effect at a larger scale, with one query ranging from 48 ms to 18.5 s depending on autovacuum.

So create partitions ahead of time (cron or pg_partman), and keep published rows as long as audit and replay need. Drop a day only once nothing in it is waiting. The cleanup function refuses otherwise, and copies dead rows to outbox_dead first (excerpt from cleanup.py):

from __future__ import annotations

from datetime import date, datetime, timedelta

import psycopg
from psycopg import sql


def partition_name(day: date) -> str:
    return f"outbox_{day:%Y%m%d}"


def ensure_partitions(conn: psycopg.Connection, start: date, days: int) -> None:
    """Create one partition per UTC day in [start, start + days). Safe to rerun."""
    for i in range(days):
        day = start + timedelta(days=i)
        conn.execute(sql.SQL(
            "CREATE TABLE IF NOT EXISTS {} PARTITION OF outbox FOR VALUES FROM ({}) TO ({})"
        ).format(
            sql.Identifier(partition_name(day)),
            sql.Literal(f"{day} 00:00+00"),
            sql.Literal(f"{day + timedelta(days=1)} 00:00+00"),
        ))


def drop_partition(conn: psycopg.Connection, day: date) -> int:
    """Drop one day's partition if every row in it was published.

    Dead rows are copied to outbox_dead first. Returns the number of rows still
    waiting to be published: 0 means the partition is gone, anything else means
    it was kept.
    """
    part = sql.Identifier(partition_name(day))
    with conn.transaction():
        conn.execute("SET LOCAL lock_timeout = '2s'")  # don't queue writers behind us
        conn.execute(sql.SQL("LOCK TABLE {} IN SHARE MODE").format(part))  # no writes while we check
        pending = conn.execute(sql.SQL(
            "SELECT count(*) FROM {} WHERE published_at IS NULL AND dead_at IS NULL"
        ).format(part)).fetchone()[0]
        if pending:
            return pending
        conn.execute(sql.SQL(
            "INSERT INTO outbox_dead SELECT * FROM {} WHERE dead_at IS NOT NULL"
        ).format(part))
        conn.execute(sql.SQL("DROP TABLE {}").format(part))
    return 0

The tests keep a partition while it reports 1 pending row, drop it once that row is published, and move a dead row out first.

The lock_timeout earns its place in another test. Dropping a partition also locks the parent table, so I left one transaction open after it had read the outbox, then ran the drop. It waited 2 seconds and gave up with 55P03 canceling statement due to lock timeout.

That’s the good outcome: without the timeout, every insert would have queued behind the waiting drop. On a busy table, DETACH PARTITION ... CONCURRENTLY first avoids that lock.

What should you alert on?

The age of the oldest pending row, not the count. A backlog that drains in seconds is fine; one row stuck for ten minutes isn’t (monitoring.sql):

-- Alert on the age of the oldest unsent event, not on the backlog count:
-- a big backlog that drains in seconds is fine, one row stuck for ten minutes isn't.
SELECT
    coalesce(extract(epoch FROM now() - min(created_at)), 0) AS oldest_pending_seconds,
    count(*)                                                 AS pending,
    count(*) FILTER (WHERE attempts > 0)                     AS retrying
FROM outbox
WHERE published_at IS NULL AND dead_at IS NULL;

-- Dead rows need a human (or a replay script), so page on any.
SELECT count(*) AS dead FROM outbox WHERE dead_at IS NOT NULL;

Polling vs LISTEN/NOTIFY vs Debezium CDC: which relay should you run?

You’ve now seen three ways for the relay to find new rows. Before you read the table, guess which one I’d run for order 101.

Polling plus SKIP LOCKED, and that’s my recommendation for you too. It’s one process you already know how to run, and polling every 100 ms gets order 101’s event out within 106 ms at p99.

Add the NOTIFY wake-up only if you need a p99 nearer 22 ms and your writes stay under its ceiling. Honestly, most teams won’t, though your hardware may say otherwise. Move to Debezium once Kafka Connect is already part of your stack.

The move to Debezium is cheap if you kept the column names. Debezium’s Outbox Event Router (current release 3.7) reads the same five columns, routes to outbox.event.<aggregatetype> and keys by aggregateid, exactly like the Python relay.

For a partitioned outbox, create the publication with publish_via_partition_root = true (default false). Then changes arrive as the parent table, not one table per day.

What CDC costs you is a replication slot to look after. Think of it as a bar tab that keeps running after you leave. A stopped connector loses nothing, but its slot keeps WAL on the primary until it catches up, or until the disk fills.

Slots can follow a failover since PostgreSQL 17, if you set that up. Once you’re on CDC, pg_logical_emit_message() can even skip the table and write the event straight into the WAL, which is what SeatGeek does (Gunnar Morling explains it).

What does CDC cost you in production?

Shopify’s 2021 write-up reports a p99 under 10 seconds from MySQL insert to Kafka. That’s not Postgres, but it’s a public number for the “Not measured” cell above.

Latency isn’t the only cost. Netflix’s DBLog paper (2020) explains why CDC also has to copy full state: transaction logs keep limited history. DBLog does it without locks, and Debezium’s incremental snapshots (1.6 and later) credit that paper.

One more guess before the table. A CDC connector stops reading, and three hours pass. How much disk does its replication slot hold on to?

On a 20 GiB RDS instance, Gunnar Morling measured more than 2 GB. Here’s that lesson and two of Shopify’s, set against this outbox:

So the recommendation stands, with one warning. A slot that falls further behind than max_slot_wal_keep_size allows may not be able to continue. You trade a full disk for a connector to set up again. PostgreSQL 18’s idle_replication_slot_timeout invalidates unused slots, but it’s off by default. The polling relay has no slot: stop it, and events wait in the table.

Shopify and Slack in production: what runs out first?

For scale: over the Black Friday Cyber Monday weekend in 2020, Shopify’s change data capture processed about 65,000 records a second on average, with spikes to 100,000. It ran about 150 Debezium connectors, one per database shard, on 12 Kubernetes pods. That’s about 430 records a second per connector, a small fraction of the 15,390 one worker managed here.

Neither number is a benchmark to beat: Shopify counts row changes across whole tables, and the 51,619 here is one Postgres and a stub publisher. My lesson: relay throughput is rarely what runs out first. What bites is the backlog after an outage.

Slack’s job queue shows how. On its busiest days it processed over 1.4 billion jobs, peaking at 33,000 a second.

Then database contention slowed the jobs, Redis hit its memory limit, and nothing new could be enqueued. Dequeuing needed a little free memory too, so the queue stayed stuck after the database recovered. It took extensive manual intervention to get it moving.

Slack’s fixes were a durable buffer (Kafka in front of Redis) and rate limits in its relay, JQRelay. This design has both. The outbox table is the buffer, on disk rather than in a memory-capped Redis, and max_per_second is the rate limit.

Disk runs out too, eventually: the cleanup above won’t drop a day with pending rows. That’s why the monitoring query alerts on the age of the oldest one.

Should you just use a Python library?

Build it yourself first. The core is a table, one claim query and about a hundred lines of loop. Once you’ve written it, you’ll know what to look for in someone else’s. The Python outbox libraries are young and moving fast:

Release dates are from PyPI (checked 4 November 2026) and say nothing about quality. Whatever you pick, read its claim query. If it isn’t FOR UPDATE SKIP LOCKED or a lease, don’t run two copies.

AI agents and LLM gateways write twice too

Order 101 isn’t special. Anything that changes your database and then tells another system has the same gap, and AI features are full of them:

• Gateway spend. In the LLM gateway post, the billing record lives in a durable ledger. Anything else that needs the spend (an invoice, a budget alert) is a second write. Put the ledger row and an outbox row in one transaction, and let the relay tell the rest.
• Agent tool calls. A book_meeting tool that saves the booking and emails the invite has place_order‘s gap: an invite for a booking that rolled back, or no invite at all. Save the booking with an outbox row, and send the email keyed by the outbox id, so a redelivery can’t send a second invite.
• Embeddings for RAG. A document in Postgres and its vectors in a vector database are two writes to two systems. Add an outbox row with the document, and let the relay embed it and upsert by document id. An upsert by id is idempotent, so at-least-once is fine. Deletes need an event too, or the retriever keeps quoting a document you removed.

Two catches. An embedding call takes far longer than a Kafka ack. So that relay wants the locked_until lease from the relay section, not a transaction held open while the model runs. And the agent retrying the tool itself is Part 1‘s problem: derive the idempotency key in the tool runtime.

Transactional outbox pattern mistakes to check before you ship

Each of these passes the happy-path test: one worker, one event, nothing crashes. They show up the day you add a second worker, a long report or a broker outage.

• Two relays on a plain SELECT. The naive pair sent 10,000 of 10,000 events twice. Claim with FOR UPDATE SKIP LOCKED.
• A watermark on id > last_seen. A later id can commit first, and the watermark skips the row. Use the snapshot reader.
• An open transaction with no deadline. Set idle_in_transaction_session_timeout, or use a lease for slow batches.
• Counting an outage as a failed attempt. Roll the batch back instead, and save dead_at for rows the broker refused.
• Cleaning up with DELETE. Drop daily partitions, and create them days ahead: a missing one fails the order with SQLSTATE 23514.
• Alerting on the backlog count. Alert on oldest_pending_seconds, and page on any dead row.
• Consumers that assume exactly once. 50 kills sent 2,800 duplicates. Every consumer has to drop an id it has already seen.

If you do one thing this week, run two copies of your relay against a test database and count the duplicates. If the number isn’t zero, the claim query is the bug.

And order 101? This time its email goes out. Maybe twice. The transactional outbox pattern trades “did we publish?” for “did we publish twice?” The second question has a clean answer: the idempotent consumer pattern, next in this series.

Over to you: which relay do you run, and what broke first? The ordering, the cleanup, or the day someone started a second copy? Tell me in the comments.

The schema, the relay, the 36 tests and the benchmark scripts are on GitHub. And if this helped, join the newsletter for the next one.