Transactional outbox

Deep dive

How it works

Parts:

• outbox table — same database: id (monotonic), aggregate_type, aggregate_id, type, payload, optional sent_at.
• writer — business code; inserts the row in the state change's transaction.
• relay — polling publisher or log-tailing connector; id order, key = aggregate_id.
• broker — carries the row id as message id.
• consumer + inbox — processed ids, written in the same transaction as the effect.

1. Write: BEGIN; UPDATE state…; INSERT INTO outbox…; COMMIT. A rollback discards both.
2. Polling relay: SELECT … WHERE sent_at IS NULL ORDER BY id LIMIT n FOR UPDATE SKIP LOCKED; publish the batch; wait for the ack; UPDATE … SET sent_at (or DELETE); commit.
3. Log-tailing relay (change data capture): a connector reads the WAL (PostgreSQL logical decoding) or binlog (MySQL, binlog_format=ROW), sees the outbox INSERT, emits it to a topic named from aggregate_type, keyed by aggregate_id, id in a header. Debezium's reference setup deletes the row in the same transaction: the insert is already in the log.
4. Crash windows: after publish but before mark (or before the connector's offset is stored) the row goes out again: repeated, never lost.
5. Consume: in one transaction insert the event id into the inbox (unique) and apply the effect; a conflict means duplicate: skip and ack.
6. Clean up: drop sent rows or old partitions; keep inbox ids past any redelivery.

tx: state UPDATE + outbox INSERT --commit--> DB (WAL)
  relay (poll, or tail the WAL) --publish(key=aggregate_id, id)--> broker --> consumer: inbox(id) + effect, one tx
  relay then marks sent / deletes   (crash before this = duplicate)

Code

A mutex is the transaction, a shadow copy gives all-or-nothing commit. The relay can crash between publish and mark; the consumer keeps an inbox.

use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::Mutex;

#[derive(Clone, Debug)]
pub struct Row { pub id: u64, pub aggregate: u32, pub delta: i64, pub sent: bool }

#[derive(Default, Clone)]
struct Db { balances: BTreeMap<u32, i64>, outbox: Vec<Row>, next_id: u64, commits: u64 }

/// One lock = one local transaction; the shadow copy is published only on Ok.
#[derive(Default)]
pub struct Store(Mutex<Db>);

impl Store {
    fn transact<F: FnOnce(&mut Db) -> Result<(), String>>(&self, f: F) -> Result<(), String> {
        let mut g = self.0.lock().unwrap();
        let mut shadow = g.clone(); // real DBs use an undo log, not a copy
        f(&mut shadow)?;
        shadow.commits += 1; // one WAL flush per commit
        *g = shadow;
        Ok(())
    }
    /// Business operation: change the balance AND write the event row, one commit.
    pub fn deposit(&self, acct: u32, delta: i64) -> Result<(), String> {
        self.transact(|db| {
            let b = db.balances.entry(acct).or_insert(0);
            if *b + delta < 0 { return Err("insufficient funds".into()); }
            *b += delta;
            db.next_id += 1; // ids follow commit order here; see Pitfalls
            let id = db.next_id;
            db.outbox.push(Row { id, aggregate: acct, delta, sent: false });
            Ok(())
        })
    }
}

pub struct Crash; // injected failure of the relay process

/// Polling relay: read a batch ordered by id, publish, then mark sent.
/// `crash_after = Some(n)` dies after n publishes, before marking.
pub fn relay_once(db: &Store, broker: &Mutex<Vec<Row>>, batch: usize, crash_after: Option<usize>) -> Result<usize, Crash> {
    let rows: Vec<Row> = {
        let g = db.0.lock().unwrap(); // SELECT .. WHERE NOT sent ORDER BY id LIMIT n FOR UPDATE SKIP LOCKED
        g.outbox.iter().filter(|r| !r.sent).take(batch).cloned().collect()
    };
    if rows.is_empty() { return Ok(0); }
    for (i, r) in rows.iter().enumerate() {
        if crash_after == Some(i) { return Err(Crash); }
        broker.lock().unwrap().push(r.clone());
    }
    if crash_after == Some(rows.len()) { return Err(Crash); }
    let ids: HashSet<u64> = rows.iter().map(|r| r.id).collect();
    db.transact(|d| { // one UPDATE .. WHERE id = ANY(..) for the whole batch
        d.outbox.iter_mut().filter(|r| ids.contains(&r.id)).for_each(|r| r.sent = true);
        Ok(())
    }).unwrap();
    Ok(rows.len())
}

/// Inbox: in a real system the dedupe row and the effect share a transaction.
#[derive(Default)]
pub struct Consumer { inbox: HashSet<u64>, pub mirror: BTreeMap<u32, i64>, pub applied: HashMap<u32, Vec<u64>> }

impl Consumer {
    pub fn handle(&mut self, m: &Row) {
        if !self.inbox.insert(m.id) { return; } // duplicate: ack and drop
        *self.mirror.entry(m.aggregate).or_insert(0) += m.delta;
        self.applied.entry(m.aggregate).or_default().push(m.id);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn crashes_lose_nothing_and_apply_once() {
        let (db, broker, mut c) = (Store::default(), Mutex::new(Vec::new()), Consumer::default());
        for i in 0..60u32 {
            db.deposit(i % 3, 10).unwrap();
            assert!(db.deposit(i % 3, -1_000_000).is_err()); // rolled back: no row either
        }
        let crashes = [0, 3, 8, 1, 8, 5]; // before, mid-batch, after last publish
        let mut cycle = 0;
        while db.0.lock().unwrap().outbox.iter().any(|r| !r.sent) {
            let _ = relay_once(&db, &broker, 8, crashes.get(cycle).copied());
            cycle += 1;
        }
        let msgs = broker.lock().unwrap().clone();
        msgs.iter().for_each(|m| c.handle(m));
        let ids: HashSet<u64> = msgs.iter().map(|m| m.id).collect();
        assert_eq!(ids.len(), 60);                           // nothing lost
        assert!(msgs.len() > 60);                            // duplicates were delivered
        assert_eq!(c.applied.values().map(Vec::len).sum::<usize>(), 60); // applied once
        assert_eq!(c.mirror, db.0.lock().unwrap().balances); // read model == state
        assert!(c.applied.values().all(|v| v.windows(2).all(|w| w[0] < w[1]))); // order kept
    }
}

Test result: 60 events, 85 messages published (25 duplicates), 0 lost, each applied once, per-aggregate order kept.

Complexity and constants

• Writer: +1 row insert inside the existing commit: row writes roughly double, commits (WAL flushes) do not. Extra WAL ≈ payload + index entry.
• Polling relay: one SELECT per interval even when idle, one UPDATE per batch; added latency ≈ interval/2 + batch time. A partial index on unsent rows keeps the query O(batch).
• Log tailing: no query; a replication slot retains WAL while the connector is down.
• Consumer: one unique-key insert per message.

Efficiency

Cost: one operation = one commit holding the state rows and one outbox row; relay = one publish per event plus one mark-commit per batch. In my in-memory sim (counted commits, not time), relay commits per event: 1.000, 0.100, 0.010, 0.001 at batch 1, 10, 100, 1000. Bound: the database's commit flush plus broker round trips.

Levers:

• Batch publish, one mark per batch — relay commits per event 1.0 to 0.01 at batch 100 (my sim) — whenever a backlog exists; MassTransit exposes QueryMessageLimit.
• Log tailing instead of polling — no poll query, no interval latency (Debezium docs) — high rates or tight latency; adds a connector and slot to run.
• Parallel relays with SKIP LOCKED — no blocking on each other's rows (PostgreSQL docs) — more throughput; shard by aggregate id.
• Drop partitions, not mark or DELETE — avoids the VACUUM overhead of bulk deletes (PostgreSQL partitioning docs) — high volume; mark-only tables need a purge (Spring Modulith docs).

Naive → standard → state of the art:

• Dual write (save, then publish) · crash window loses or invents events
• Outbox + batched polling relay, SKIP LOCKED · +1 row per op, latency ≈ half the interval
• Outbox via log tailing, keyed topics, inbox dedupe · no polling; slot and connector to run

Implementation notes

• Insert the row on the same connection and transaction, behind one API so nobody forgets (Richardson names forgetting as the main drawback).
• Mark only after the broker acks the whole batch.
• Several relays reorder one aggregate: hash aggregate_id to a relay, or run one.
• Inbox retention must exceed the longest redelivery (MassTransit: DuplicateDetectionWindow).
• PostgreSQL persists a logical slot's position only at checkpoint, so after a crash recent changes can be sent again (docs).

Pitfalls

• Sequence ids commit out of order: A takes id 5, B takes 6 and commits first. A cursor id > last passes 5 and loses A's event; a sent_at IS NULL scan loses nothing but publishes 6 before 5. Use a flag scan or CDC (commit order).
• Outbox row on another connection or in autocommit: a dual write again.
• A permanently failing row blocks its aggregate: count attempts, then dead-letter it.
• No cleanup: the table grows forever.
• A stalled connector holds a replication slot and WAL fills the disk (Debezium PostgreSQL docs: heartbeats, monitor lag).
• Inbox insert outside the effect's transaction: a crash between them loses the effect or applies it twice.
• "Exactly-once": the broker still sees duplicates; only the effect is once.

Variants and alternatives

In production

• Debezium outbox event router (Kafka Connect); its documented example: an order service feeding shipment and customer services (e-commerce).
• Cala, Galoy's Rust double-entry ledger library, keeps a persistent outbox of ledger events for consumers (banking).
• MassTransit, NServiceBus (.NET) and Spring Modulith (Java): outbox/inbox in messaging frameworks used across domains.

Implementations

Sources

• Chris Richardson, Pattern: Transactional outbox, microservices.io. page · polling publisher · transaction log tailing
• Chris Richardson, Microservices Patterns, Manning 2018.
• Gunnar Morling, Reliable Microservices Data Exchange With the Outbox Pattern, Debezium blog 2019. post
• Debezium docs: Outbox Event Router, PostgreSQL connector, MySQL connector · code
• PostgreSQL docs: SELECT, locking clause (SKIP LOCKED), logical decoding concepts, table partitioning
• Library docs: MassTransit, NServiceBus, Spring Modulith