Implementing the Transactional Outbox Pattern for Webhooks

The bug is always the same shape. Your handler commits the order to PostgreSQL, then calls queue.publish(order_created_event) on the next line. Between those two lines the pod is evicted, the connection to the broker times out, or the process is OOM-killed. The order exists. The webhook does not, and never will — no retry can help, because the event was never recorded anywhere. This is the dual-write problem, and the transactional outbox is the standard fix. This guide, part of Delivery Guarantee Levels, implements it end to end in Python and PostgreSQL. If you have not yet decided what semantics you are promising consumers, settle that first in choosing a webhook delivery guarantee level — the outbox is how you actually deliver on at-least-once, not a substitute for choosing it.

The core idea is a single sentence: the event is written to your own database, in the same transaction as the state change it describes, and something else publishes it later. Two writes to two systems become one write to one system, and atomicity does the rest. What you buy is the elimination of lost events. What you pay is a background relay to operate and a table to prune.

Prerequisites

• PostgreSQL 14+ (for SKIP LOCKED and generated identity columns) with psycopg 3.1. The pattern works on MySQL 8 and SQL Server with minor syntax changes.
• Python 3.10+ with httpx 0.27 for outbound delivery.
• A stable aggregate identifier on every event — the account, order or tenant whose ordering must be preserved.
• Consumers that deduplicate on an event ID. The relay is at-least-once by construction; see how to design idempotent webhook consumers and, for the storage side, storing idempotency keys in Postgres.
• A dead-letter path for events the relay cannot deliver after its retry budget, as described in Dead-Letter Queue Architecture.

Step 1: Write the event to an outbox table in the same transaction

The outbox is an ordinary table in the same database and schema as your business data. That co-location is the entire mechanism: because it is the same database, the INSERT into outbox joins the same transaction as the INSERT into orders, and the two either both exist or neither does.

The schema carries the routing information the relay needs, plus the columns that make ordering and pruning possible later:

import psycopg

SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS outbox (
    id             BIGSERIAL PRIMARY KEY,
    event_id       UUID        NOT NULL UNIQUE,     -- consumer-facing idempotency key
    aggregate_type TEXT        NOT NULL,            -- 'order', 'account', ...
    aggregate_id   TEXT        NOT NULL,            -- ordering lane
    aggregate_seq  BIGINT      NOT NULL,            -- monotonic within the lane
    event_type     TEXT        NOT NULL,
    payload        JSONB       NOT NULL,
    created_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
    claimed_until  TIMESTAMPTZ,                     -- lease held by a relay worker
    attempts       INT         NOT NULL DEFAULT 0,
    sent_at        TIMESTAMPTZ,                     -- NULL means still pending
    UNIQUE (aggregate_type, aggregate_id, aggregate_seq)
);

-- Partial index: only unsent rows are ever scanned by the relay, so the index
-- stays tiny even when the table holds millions of already-sent rows.
CREATE INDEX IF NOT EXISTS outbox_pending_idx
    ON outbox (aggregate_id, aggregate_seq)
    WHERE sent_at IS NULL;
"""

def migrate(conn: psycopg.Connection) -> None:
    with conn.cursor() as cur:
        cur.execute(SCHEMA_SQL)
    conn.commit()

The write path is then a single transaction with two statements. The critical detail is that enqueue_event takes the cursor, not a connection or a session factory — passing a cursor makes it structurally impossible to accidentally write the event on a different connection, which is the most common way teams reintroduce the dual write they were trying to remove:

import json
import uuid

def enqueue_event(cur: psycopg.Cursor, *, aggregate_type: str, aggregate_id: str,
                  event_type: str, payload: dict) -> str:
    """Append an event to the outbox on the CALLER'S cursor, inside their transaction."""
    event_id = str(uuid.uuid4())
    cur.execute(
        """
        INSERT INTO outbox (event_id, aggregate_type, aggregate_id, aggregate_seq,
                            event_type, payload)
        VALUES (%s, %s, %s,
                (SELECT COALESCE(MAX(aggregate_seq), 0) + 1
                   FROM outbox
                  WHERE aggregate_type = %s AND aggregate_id = %s),
                %s, %s)
        """,
        (event_id, aggregate_type, aggregate_id,
         aggregate_type, aggregate_id, event_type, json.dumps(payload)),
    )
    return event_id

def create_order(conn: psycopg.Connection, customer_id: str, total_cents: int) -> str:
    with conn.transaction():                       # one transaction, two writes
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO orders (customer_id, total_cents) VALUES (%s, %s) "
                "RETURNING id",
                (customer_id, total_cents),
            )
            order_id = str(cur.fetchone()[0])
            enqueue_event(
                cur,
                aggregate_type="order",
                aggregate_id=order_id,
                event_type="order.created",
                payload={"order_id": order_id, "customer_id": customer_id,
                         "total_cents": total_cents},
            )
    return order_id

Note what is absent: no HTTP call, no broker publish, no network I/O of any kind inside the transaction. Holding a database transaction open across a network call is how outbox implementations turn into lock-contention incidents.

Step 2: Relay claimed rows and mark them sent

The relay is a separate process that repeatedly claims a batch of unsent rows, publishes them, and marks them sent. Correctness here rests on two things: multiple relay workers must never claim the same row, and the mark-sent update must happen in its own transaction after the publish succeeds.

SELECT ... FOR UPDATE SKIP LOCKED gives you the first property for free. Workers that collide simply skip the locked rows and take the next ones, so you can scale the relay horizontally with no coordination.

The relay loop, with a lease so a worker that dies does not strand its batch forever:

import datetime as dt
import httpx

CLAIM_SQL = """
UPDATE outbox
   SET claimed_until = now() + %s::interval,
       attempts = attempts + 1
 WHERE id IN (
       SELECT id FROM outbox
        WHERE sent_at IS NULL
          AND (claimed_until IS NULL OR claimed_until < now())
        ORDER BY aggregate_id, aggregate_seq
        LIMIT %s
          FOR UPDATE SKIP LOCKED
 )
RETURNING id, event_id, aggregate_id, aggregate_seq, event_type, payload, attempts;
"""

class Relay:
    def __init__(self, conn: psycopg.Connection, client: httpx.Client,
                 endpoint: str, lease: str = "60 seconds", batch: int = 200,
                 max_attempts: int = 8):
        self.conn = conn
        self.client = client
        self.endpoint = endpoint
        self.lease = lease
        self.batch = batch
        self.max_attempts = max_attempts

    def claim(self) -> list[tuple]:
        with self.conn.transaction(), self.conn.cursor() as cur:
            cur.execute(CLAIM_SQL, (self.lease, self.batch))
            return cur.fetchall()

    def mark_sent(self, row_id: int) -> None:
        with self.conn.transaction(), self.conn.cursor() as cur:
            cur.execute(
                "UPDATE outbox SET sent_at = now(), claimed_until = NULL "
                "WHERE id = %s AND sent_at IS NULL",
                (row_id,),
            )

    def release(self, row_id: int) -> None:
        """Give the row back immediately so the next poll retries it."""
        with self.conn.transaction(), self.conn.cursor() as cur:
            cur.execute("UPDATE outbox SET claimed_until = NULL WHERE id = %s",
                        (row_id,))

    def publish(self, row: tuple) -> None:
        row_id, event_id, agg_id, seq, event_type, payload, attempts = row
        resp = self.client.post(
            self.endpoint,
            json={"event_id": str(event_id), "type": event_type,
                  "aggregate_id": agg_id, "sequence": seq, "data": payload},
            headers={"X-Idempotency-Key": str(event_id),
                     "X-Event-Sequence": str(seq)},
            timeout=httpx.Timeout(connect=3.0, read=10.0, write=10.0, pool=3.0),
        )
        resp.raise_for_status()

    def run_once(self) -> dict:
        sent = failed = dead = 0
        for row in self.claim():
            row_id, attempts = row[0], row[6]
            try:
                self.publish(row)
                self.mark_sent(row_id)             # only after a 2xx
                sent += 1
            except Exception:
                if attempts >= self.max_attempts:
                    self.dead_letter(row)
                    self.mark_sent(row_id)         # terminal: stop re-claiming it
                    dead += 1
                else:
                    self.release(row_id)
                    failed += 1
        return {"sent": sent, "failed": failed, "dead": dead}

How you trigger run_once is a real trade-off, and it is where most of the latency budget goes:

Start with the 200 ms poll. It is one process, it has no extra moving parts, and the partial index makes the query cost effectively constant regardless of table size. Move to change data capture only when sub-100 ms delivery latency is a stated requirement, not because it sounds more elegant — a stalled replication slot can fill a disk and take the primary down with it.

Step 3: Preserve per-entity ordering with a monotonic sequence

ORDER BY id is not enough. BIGSERIAL values are handed out when the INSERT executes, but rows become visible when the transaction commits — so a long transaction that grabbed id 100 can commit after a short one that grabbed id 101, and a relay ordering by id will publish 101 first and may never see 100 again if it has already advanced past it.

The fix that matters in practice is to stop caring about global order and enforce order only where it is semantically required: within one aggregate. The aggregate_seq column, unique per (aggregate_type, aggregate_id), gives each entity its own strictly increasing counter, and the relay dispatches one lane per aggregate. Different aggregates run in parallel; one aggregate is always serial. This is the same discipline described in implementing strict ordering for financial webhooks, applied at the outbox rather than at the queue.

Dispatching in lanes is a grouping plus a thread pool that never puts two rows from one aggregate in flight at once:

from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor

def dispatch_in_lanes(relay: Relay, rows: list[tuple], max_lanes: int = 16) -> dict:
    lanes: dict[str, list[tuple]] = defaultdict(list)
    for row in rows:
        lanes[row[2]].append(row)                  # row[2] is aggregate_id
    for aggregate_rows in lanes.values():
        aggregate_rows.sort(key=lambda r: r[3])    # row[3] is aggregate_seq

    def run_lane(aggregate_rows: list[tuple]) -> int:
        delivered = 0
        for row in aggregate_rows:
            try:
                relay.publish(row)
                relay.mark_sent(row[0])
                delivered += 1
            except Exception:
                relay.release(row[0])
                break                              # stop the lane; later seqs must wait
        return delivered

    with ThreadPoolExecutor(max_workers=max_lanes) as pool:
        return {"delivered": sum(pool.map(run_lane, lanes.values()))}

The break is the whole point. When sequence 2 fails, sequences 3 and 4 for that aggregate stay unsent and unclaimed, and the next poll retries from 2. Without it, a transient failure on one event silently reorders the entity’s stream.

Step 4: Make consumers idempotent for at-least-once relay

Look again at the sequence diagram: the relay publishes, gets a 200, and then updates sent_at. If the process dies in between, the row is still unsent, the lease expires, another worker claims it, and the subscriber receives the same event twice. That is not a bug to be engineered away — it is the defining property of the pattern. A transactional outbox delivers at-least-once and nothing stronger. Any attempt to make it exactly-once by reordering the update before the publish converts duplicates into losses, which is strictly worse.

The mitigation lives on the consumer, and the relay’s job is to make it possible by sending a key that is stable across retries. event_id is generated once at insert time and never changes, which is exactly the property required:

def apply_event(cur: psycopg.Cursor, event: dict) -> str:
    """Consumer side: record the key first, and skip the work if it already exists."""
    cur.execute(
        "INSERT INTO processed_events (event_id, received_at) "
        "VALUES (%s, now()) ON CONFLICT (event_id) DO NOTHING",
        (event["event_id"],),
    )
    if cur.rowcount == 0:
        return "duplicate_ignored"                 # already applied, do nothing
    cur.execute(
        "UPDATE orders SET status = %s WHERE id = %s",
        (event["data"]["status"], event["data"]["order_id"]),
    )
    return "applied"

Both statements run in one consumer-side transaction, so the dedup key and the effect it guards commit together. Give processed_events a retention longer than your longest replay window, including anything sitting in a dead-letter queue, or a late replay will be treated as new.

Step 5: Prune the outbox table on a retention window

An unpruned outbox grows forever. The partial index keeps the relay fast, but the table itself consumes storage, slows backups, and eventually causes autovacuum to spend its budget on rows nobody will ever read again.

Delete in bounded batches. A single unbounded DELETE over millions of rows takes a long-lived lock and generates enough WAL to stall replication:

PRUNE_SQL = """
DELETE FROM outbox
 WHERE id IN (
       SELECT id FROM outbox
        WHERE sent_at IS NOT NULL
          AND sent_at < now() - %s::interval
        ORDER BY id
        LIMIT %s
 );
"""

def prune(conn: psycopg.Connection, retention: str = "30 days",
          batch: int = 5000, max_batches: int = 200) -> int:
    """Delete sent rows past the retention window, a bounded batch at a time."""
    removed = 0
    for _ in range(max_batches):
        with conn.transaction(), conn.cursor() as cur:
            cur.execute(PRUNE_SQL, (retention, batch))
            deleted = cur.rowcount
        removed += deleted
        if deleted < batch:
            break                                  # caught up
    return removed

Set retention to the longest period over which you might legitimately replay: usually the maximum of your dead-letter retention and any contractual “resend the last N days” commitment. Thirty days is a common, defensible default. Run prune on a schedule during off-peak hours, and alert if it ever hits max_batches, because that means deletion is losing the race against inserts.

Verification and testing

The property that justifies the whole pattern is atomicity, so test it by making the business write fail after the outbox insert and asserting that neither survives:

import pytest

def test_rollback_removes_both_the_order_and_the_event(conn):
    before = count(conn, "SELECT count(*) FROM outbox")
    with pytest.raises(RuntimeError):
        with conn.transaction():
            with conn.cursor() as cur:
                cur.execute("INSERT INTO orders (customer_id, total_cents) "
                            "VALUES ('c1', 100) RETURNING id")
                order_id = str(cur.fetchone()[0])
                enqueue_event(cur, aggregate_type="order", aggregate_id=order_id,
                              event_type="order.created", payload={})
                raise RuntimeError("payment authorization failed")
    assert count(conn, "SELECT count(*) FROM outbox") == before
    assert count(conn, "SELECT count(*) FROM orders WHERE customer_id = 'c1'") == 0

def test_crash_between_publish_and_mark_sent_causes_a_duplicate_not_a_loss(conn):
    relay = Relay(conn, client=RecordingClient(), endpoint="http://x/hooks")
    create_order(conn, "c2", 500)
    rows = relay.claim()
    relay.publish(rows[0])                          # delivered
    # process dies here: mark_sent never runs, lease expires
    conn.execute("UPDATE outbox SET claimed_until = now() - interval '1 minute'")
    again = relay.claim()
    assert len(again) == 1                          # re-claimed, will be re-sent
    assert again[0][1] == rows[0][1]                # same event_id: consumer dedupes

def test_lane_stops_at_the_first_failure(conn):
    rows = [(1, "e1", "acct-1", 1, "a", {}, 0),
            (2, "e2", "acct-1", 2, "b", {}, 0),
            (3, "e3", "acct-1", 3, "c", {}, 0)]
    relay = FailingAtSecond(conn)
    result = dispatch_in_lanes(relay, rows)
    assert result["delivered"] == 1                 # seq 3 must not overtake seq 2

Operationally, the two numbers to watch are outbox lag and unsent depth. Both come straight out of the table:

# Oldest unsent event: this is your true end-to-end delivery lag.
psql -c "SELECT coalesce(extract(epoch from now() - min(created_at)), 0) AS lag_seconds
           FROM outbox WHERE sent_at IS NULL;"

# Backlog and stuck-claim count. A rising claimed-but-unsent count means
# relay workers are dying mid-batch.
psql -c "SELECT count(*) FILTER (WHERE sent_at IS NULL) AS pending,
                count(*) FILTER (WHERE sent_at IS NULL
                                   AND claimed_until > now()) AS in_flight,
                count(*) FILTER (WHERE attempts > 5 AND sent_at IS NULL) AS struggling
           FROM outbox;"

A healthy relay keeps lag_seconds in the single digits, pending near zero between bursts, and struggling at zero.

Failure modes and gotchas

• The outbox insert on a different connection. Using a separate session, a separate pool checkout or an autocommit connection for the outbox write silently restores the dual write you set out to remove, and the failure is invisible until a crash at exactly the wrong moment. Pass the caller’s cursor (Step 1) and add a test that rolls back the business transaction and asserts the outbox row is gone.
• Marking sent before publishing. Reordering the update to run before the HTTP call looks like it removes duplicates. It converts every duplicate into a permanent loss, which is the exact failure the outbox exists to prevent. The update trails the publish, always.
• Ordering the relay by the surrogate key. ORDER BY id looks correct and breaks under concurrent transactions, because insert order is not commit order. Order by (aggregate_id, aggregate_seq) and dispatch in lanes as in Step 3, or accept that events can arrive out of order and say so in your contract.
• A poison event that blocks its lane forever. With the lane break, one permanently failing event stops every later event for that aggregate. That is correct for ordering and unacceptable indefinitely, so the max_attempts path in run_once must dead-letter the row and mark it sent — otherwise a single malformed payload freezes one customer’s entire event stream.
• No pruning, or unbounded pruning. Skipping retention grows the table until backups and autovacuum suffer; deleting in one statement takes a long lock and floods the WAL. Batch the deletes (Step 5), keep retention longer than your dead-letter retention, and alert when a prune run cannot catch up.

Frequently Asked Questions

Related

• Choosing a Webhook Delivery Guarantee Level
• How to Design Idempotent Webhook Consumers
• Implementing Strict Ordering for Financial Webhooks
• Delivery Guarantee Levels