The outbox pattern

Naive side-effect right after the INSERT. Three ways to lose.

def handle_deposit(conn, event):
    with conn.transaction():
        post_entry(conn, deposit_entry(event))

    # (1) Commit could fail; SMS already in flight.
    # (2) SMS could fail; commit already on disk.
    # (3) Process could crash here; nobody knows what state we're in.
    sms.send(event.user_phone, f"Deposit of {event.amount_formatted} received")

Same bug in Go. Side-effect lives outside the transaction.

func HandleDeposit(ctx context.Context, db *sql.DB, ev DepositEvent) error {
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    if err := postEntry(ctx, tx, depositEntry(ev)); err != nil {
        tx.Rollback()
        return err
    }
    if err := tx.Commit(); err != nil {
        return err  // SMS not sent yet, fine
    }

    // Anything from here to the SMS call is unprotected:
    // crash, SMS-API failure, or successful Commit + lost notification.
    return sms.Send(ctx, ev.UserPhone, fmt.Sprintf("Deposit of %s received", ev.AmountFormatted))
}

The outbox table. Same transaction as your ledger writes.

CREATE TABLE outbox (
  id              uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  topic           text NOT NULL,             -- 'sms.deposit', 'webhook.partner', etc.
  payload         jsonb NOT NULL,
  idempotency_key text NOT NULL,             -- so workers never double-deliver
  created_at      timestamptz NOT NULL DEFAULT now(),
  sent_at         timestamptz,
  attempt_count   int NOT NULL DEFAULT 0,
  next_attempt_at timestamptz NOT NULL DEFAULT now()
);

CREATE UNIQUE INDEX outbox_idem ON outbox (topic, idempotency_key);

-- The worker pulls pending rows. ORDER BY + LIMIT + FOR UPDATE SKIP LOCKED
-- lets many workers consume in parallel without stepping on each other.
CREATE INDEX outbox_pending
  ON outbox (next_attempt_at)
  WHERE sent_at IS NULL;

Commit the entry and the side-effect together. One transaction.

import json

def handle_deposit(conn, event):
    with conn.transaction():
        post_entry(conn, deposit_entry(event))

        conn.execute(
            """
            INSERT INTO outbox (topic, payload, idempotency_key)
            VALUES (%s, %s::jsonb, %s)
            """,
            (
                "sms.deposit",
                json.dumps({"user_phone": event.user_phone, "amount": event.amount_formatted}),
                event.id,  # same key as the ledger entry
            ),
        )
    # Worker takes it from here. Nothing else to do.

Worker drain: SKIP LOCKED lets many workers consume in parallel.

BEGIN;

SELECT id, topic, payload, attempt_count
  FROM outbox
 WHERE sent_at IS NULL
   AND next_attempt_at <= now()
 ORDER BY next_attempt_at
 LIMIT 50
 FOR UPDATE SKIP LOCKED;

-- For each row, dispatch the side-effect. On success:
UPDATE outbox SET sent_at = now() WHERE id = $1;
-- On failure, backoff:
UPDATE outbox
   SET attempt_count = attempt_count + 1,
       next_attempt_at = now() + (interval '1 second' * pow(2, attempt_count))
 WHERE id = $1;

COMMIT;

Worker loop in Go. The SKIP LOCKED query is the only interesting line.

func DrainOutbox(ctx context.Context, db *sql.DB, dispatch func(topic string, payload []byte) error) error {
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    rows, err := tx.QueryContext(ctx, `
        SELECT id, topic, payload, attempt_count
          FROM outbox
         WHERE sent_at IS NULL AND next_attempt_at <= now()
         ORDER BY next_attempt_at
         LIMIT 50
         FOR UPDATE SKIP LOCKED
    `)
    if err != nil {
        return err
    }
    defer rows.Close()

    for rows.Next() {
        var id, topic string
        var payload []byte
        var attempts int
        if err := rows.Scan(&id, &topic, &payload, &attempts); err != nil {
            return err
        }
        if err := dispatch(topic, payload); err != nil {
            _, _ = tx.ExecContext(ctx, `
                UPDATE outbox
                   SET attempt_count = attempt_count + 1,
                       next_attempt_at = now() + (interval '1 second' * pow(2, attempt_count))
                 WHERE id = $1
            `, id)
            continue
        }
        if _, err := tx.ExecContext(ctx, `UPDATE outbox SET sent_at = now() WHERE id = $1`, id); err != nil {
            return err
        }
    }
    return tx.Commit()
}