10 - Outbox Pattern: Atomic Publishing of Events with CDC

Saving the order and announcing it: two systems, no transaction

A service saves an order in the database and must publish OrderPlaced on a broker. These are two writes on two systems, and no transaction covers them together. If you write the database first and then the broker, a failure in the middle leaves an order that nobody will ever know anything about.

If you write the broker first, a failure in the middle announces an order that does not exist. Retrying does not fix it: the problem, called dual write, is structural. The outbox pattern sidesteps it by writing the event in a table of the same database, in the same transaction as the order, and leaving to a separate process, the relay, the job of sending it to the broker.

Atomicity is provided by the database, which already has it; the price is that the relay can send an event twice. I run the mechanism on SQLite with a fake broker that you can bring down, first trying the two naive versions, then the outbox, then a relay that dies at the worst point, and I report from the Debezium documentation how it is done without polling. I need the concepts of the idempotent consumer, because repeated delivery is unavoidable, and of the fundamentals of events.

The three versions, run

The fake broker has a down switch and a log of what it actually received. The order has a CHECK (total >= 0) constraint, which lets you make the database write fail on demand with a negative total. Cases A and B are the two naive versions; C is the outbox with the write that fails; D to H are the relay.

import { DatabaseSync } from 'node:sqlite';

const db = new DatabaseSync(':memory:');
db.exec(`
  CREATE TABLE orders (id TEXT PRIMARY KEY, total INTEGER NOT NULL CHECK (total >= 0));
  CREATE TABLE outbox (id TEXT PRIMARY KEY, aggregate_id TEXT NOT NULL, event_type TEXT NOT NULL,
                       payload TEXT NOT NULL, published_at TEXT);
`);

// A fake broker: it can be "down" and it keeps a log of what it actually received.
const broker = { down: false, received: [] as string[], publish(m: string) { if (this.down) throw new Error('broker unreachable'); this.received.push(m); } };
const orders = () => (db.prepare('SELECT COUNT(*) AS n FROM orders').get() as { n: number }).n;

// --- Naive dual write: database and broker do not share a transaction.
function naiveOrderDbThenBroker(id: string, total: number): void {
  db.prepare('INSERT INTO orders VALUES (?, ?)').run(id, total);
  broker.publish(`OrderPlaced:${id}`);                  // if it fails, the order exists and nobody knows
}
function naiveOrderBrokerThenDb(id: string, total: number): void {
  broker.publish(`OrderPlaced:${id}`);
  db.prepare('INSERT INTO orders VALUES (?, ?)').run(id, total);   // if it fails, the event describes an order that does not exist
}
broker.down = true;
try { naiveOrderDbThenBroker('o-1', 100); } catch (e) { console.log('A) db then broker, broker down: orders in the db =', orders(), '| events =', broker.received.length); }
broker.down = false;
try { naiveOrderBrokerThenDb('o-2', -5); } catch (e) { console.log('B) broker then db, invalid total: orders in the db =', orders(), '| events =', broker.received.length, broker.received); }

// --- Outbox: order and event in the same transaction.
function orderWithOutbox(id: string, eventId: string, total = 100): void {
  db.exec('BEGIN');
  try {
    db.prepare('INSERT INTO orders VALUES (?, ?)').run(id, total);
    db.prepare('INSERT INTO outbox (id, aggregate_id, event_type, payload) VALUES (?, ?, ?, ?)').run(eventId, id, 'OrderPlaced', JSON.stringify({ orderId: id }));
    db.exec('COMMIT');
  } catch (e) { db.exec('ROLLBACK'); throw e; }
}
orderWithOutbox('o-3', 'evt-3');
try { orderWithOutbox('o-6', 'evt-6', -5); } catch { console.log('C) invalid total: rollback, events in outbox =', (db.prepare('SELECT COUNT(*) AS n FROM outbox').get() as { n: number }).n, '(only the one of o-3)'); }
orderWithOutbox('o-4', 'evt-4');

// --- Relay: reads the unpublished ones in order, publishes, then marks. It can die between the two actions.
function relay(dieAfterPublish = false): void {
  const rows = db.prepare('SELECT id, aggregate_id, event_type FROM outbox WHERE published_at IS NULL ORDER BY rowid LIMIT 100').all() as
    { id: string; aggregate_id: string; event_type: string }[];
  for (const r of rows) {
    broker.publish(`${r.event_type}:${r.aggregate_id}#${r.id}`);
    if (dieAfterPublish) throw new Error('relay died after publishing, before marking');
    db.prepare("UPDATE outbox SET published_at = 'now' WHERE id = ?").run(r.id);
  }
}
broker.received.length = 0;
broker.down = true;
try { relay(); } catch { console.log('D) broker down: events delivered =', broker.received.length, '| unpublished in outbox =', (db.prepare('SELECT COUNT(*) AS n FROM outbox WHERE published_at IS NULL').get() as { n: number }).n); }
broker.down = false;
try { relay(true); } catch { console.log('E) relay died after the first publish: delivered =', broker.received.length); }
relay();
console.log('F) after the relay restarts: delivered =', broker.received.length, broker.received);

// The consumer discards the copies using the event id.
const seen = new Set<string>();
const effective = broker.received.filter((m) => { const id = m.split('#')[1]; if (seen.has(id)) return false; seen.add(id); return true; });
console.log('G) effects after deduplication by event id =', effective.length);

// Cleanup: only the published rows are deleted.
const deleted = db.prepare("DELETE FROM outbox WHERE published_at IS NOT NULL").run().changes;
console.log('H) outbox rows cleaned =', deleted, '| left =', (db.prepare('SELECT COUNT(*) AS n FROM outbox').get() as { n: number }).n);

// Output:
// A) db then broker, broker down: orders in the db = 1 | events = 0
// B) broker then db, invalid total: orders in the db = 1 | events = 1 [ 'OrderPlaced:o-2' ]
// C) invalid total: rollback, events in outbox = 1 (only the one of o-3)
// D) broker down: events delivered = 0 | unpublished in outbox = 2
// E) relay died after the first publish: delivered = 1
// F) after the relay restarts: delivered = 3 [
//   'OrderPlaced:o-3#evt-3',
//   'OrderPlaced:o-3#evt-3',
//   'OrderPlaced:o-4#evt-4'
// ]
// G) effects after deduplication by event id = 2
// H) outbox rows cleaned = 2 | left = 0

Case A shows the lost event: order o-1 is in the database, and no event was published. Case B shows the phantom event: OrderPlaced:o-2 is in the broker, but the orders in the database are still one. In case C the order write fails, the transaction cancels the event too, and the outbox contains only the one for o-3: there is no event without its order.

In case D the broker is down, the relay delivers nothing and the two events stay in the outbox with published_at null, so nothing is lost. Case E is the worst moment: the relay publishes and dies before marking the row; on restart (F) it republishes it, and the broker has three messages, two of them identical. Case G shows the remedy, the consumer that discards the copies by event id and ends up with two effects. In case H only the rows already published are deleted.

The polling relay

The simplest relay is a loop: it reads the unpublished rows in insertion order, publishes them, marks them. It is easy to understand and to run without other components, but it has two costs: a continuous query on the database and a latency equal to the polling interval. When there is more than one relay, two instances must not take the same rows: on PostgreSQL you use SELECT ... FOR UPDATE SKIP LOCKED, which the guide points to for queue-like tables, and the read must live in the transaction that ends after the marking, otherwise the lock disappears immediately. The program has a single relay, so it does not have this problem; I could not try it with several connections. Case E remains: if the relay marks after publishing, it produces copies; if it marked first, it would lose events. The right choice is to produce copies and let the consumer absorb them.

Without polling: Change Data Capture with Debezium

Debezium avoids polling by reading the database's transaction log and publishing the changes to Kafka; its PostgreSQL guide requires wal_level=logical and a maximum number of replication slots. For the outbox it provides a transformation, the EventRouter, which reads the table and routes each row to a topic derived from a column. By default it expects a table with precisely named columns: id, aggregatetype, aggregateid, type, payload.

It is not the table of the program above (which uses aggregate_id and event_type): either you rename it, or you configure the names with the table.field.event.* options that the guide lists (id, key, payload, schema.version, timestamp). The connector below is the minimal schema to adapt, written from the documentation and not run: it needs a Kafka Connect instance and a PostgreSQL with replication permissions.

{
  "name": "outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "db.internal",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "<from the secrets manager>",
    "database.dbname": "orders_db",
    "topic.prefix": "orders",
    "table.include.list": "public.outbox",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter"
  }
}

topic.prefix is mandatory in recent versions and replaces database.server.name. The EventRouter is declared as a transformation with transforms.outbox.type; the topic name is by default derived from the value of aggregatetype, that is, an event of type Order goes to a topic that contains Order. Two warnings from the guide.

The first: Debezium, like Kafka, guarantees at-least-once delivery even under abnormal conditions, so the consumer's idempotency is not optional even with CDC; the guide itself suggests using the event id to eliminate duplicates. The second: replication slots retain the log until the connector consumes it, and the guide recommends monitoring them to avoid excessive disk consumption if a slot stays inactive for too long. The Kafka Connect side and the slot are covered in the article on Kafka Connect and Debezium, and the limit of the Kafka transaction guarantee, which does not cover the write to the database, in the one on exactly-once.

Cleanup and choice

The table grows: it must be cleaned, but only of the rows already published, as case H does, and in batches if it is large, so as not to hold long locks. To decide, start from polling: it has one component fewer and it is enough as long as the acceptable latency exceeds the polling interval and the load on the table is low. Move to CDC when polling becomes a cost (high volume, tight latency) and you are ready to manage Kafka Connect and the slots. I do not report latency figures: they depend on your load and must be measured.

What you can do now

You can explain why dual write loses or invents events, you can write an order and its event in the same transaction and you can design a relay that survives dying at the worst point, producing copies instead of losses. You know which column names change in the move to Debezium and why the consumer must stay idempotent. The next step is the scenario in which the event arrives and the read is still behind: eventual consistency and the user experience. For those coordinating several services with these events, there is the saga. If the write side is made of events, the event store is already a log to read from: see event sourcing and its combination with CQRS; if the read views are fed from the outbox, the model is that of CQRS.