The Write and the Event Must Agree
Write the row and the event in one database transaction, then let a relay publish them, so the two can never disagree.
BEGIN; INSERT INTO orders ...; INSERT INTO outbox ...; COMMIT; -- the event is as durable as the dataTwo writes, two systems, no shared transaction: the row in your database and the event in your broker. Commit the row and die before publishing, and downstream never hears about the order. Publish first and then roll back, and downstream hears about an order that does not exist. Both happen in production, and the second is worse, because nothing in your database records the lie.
One transaction, two rows
The transactional outbox makes the event a row in the same database. The request handler inserts the order and the outbox record in one transaction, so either both exist or neither does. A relay process then reads unpublished rows, publishes them to the broker, and marks them published. The broker is no longer part of your transaction; the only thing in it is a table you already trust.
The relay publishes at least once
If the relay crashes after publishing and before marking the row, it publishes again on restart. That is not a flaw to engineer away — it is why consumers deduplicate on an id carried in the event, normally the outbox row's id. Several relay instances can run safely if they claim rows with SELECT ... FOR UPDATE SKIP LOCKED, available in PostgreSQL 9.5 and later.
Or let the log be the relay
Change data capture reads the write-ahead log instead of polling. Debezium's io.debezium.transforms.outbox.EventRouter transform turns each outbox row into a message on a topic derived from its aggregate type, which removes the polling loop at the cost of running a connector.
Example
import { pool } from "./db.js";
// One transaction, two tables. If the commit fails, neither the order nor the
// event exists, so there is nothing published that has to be taken back.
export async function placeOrder(input) {
const client = await pool.connect();
try {
await client.query("BEGIN");
const { rows: [order] } = await client.query(
`INSERT INTO orders (user_id, total_cents, status)
VALUES ($1, $2, 'confirmed')
RETURNING id, created_at`,
[input.userId, input.totalCents]
);
// The event, as data. Its id becomes the consumer's deduplication key, so the
// relay publishing twice is harmless.
await client.query(
`INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
VALUES ('order', $1, 'order.placed', $2)`,
[
order.id,
JSON.stringify({
orderId: order.id,
userId: input.userId,
totalCents: input.totalCents,
occurredAt: order.created_at,
}),
]
);
await client.query("COMMIT");
return order;
} catch (err) {
await client.query("ROLLBACK");
throw err;
} finally {
client.release();
}
}
// Note what is absent: no broker client in this function, no try/catch around a
// publish, no "what do I do if SQS is down" branch in the request path.When to use it
- Stop orders existing in the database with no event published downstream.
- Remove a broker call from the request path so a broker outage does not fail writes.
- Give a partner integration a replayable record of exactly which events you emitted.
- Run three relay instances without any of them publishing the same row.
More examples
The relay
The whole mechanism in thirty lines, including the two things that are usually missing. The lock is what lets you run more than one relay, and the age alarm is what tells you the relay has stopped — a failure mode with no error anywhere, because the application is writing happily and only the consumers notice, hours later. Ordering within an aggregate is preserved only if you publish a batch in id order and do not parallelise inside it.
// Run several of these. SKIP LOCKED (PostgreSQL 9.5 and later) means each instance
// claims a different batch instead of blocking on the same rows.
async function relayBatch() {
const client = await pool.connect();
try {
await client.query("BEGIN");
const { rows } = await client.query(
`SELECT id, aggregate_id, event_type, payload
FROM outbox
WHERE published_at IS NULL
ORDER BY id
LIMIT 100
FOR UPDATE SKIP LOCKED`
);
for (const row of rows) {
await sqs.send(new SendMessageCommand({
QueueUrl: EVENTS_QUEUE_URL,
// { ...row.payload, eventId } is shorthand for
// Object.assign({}, row.payload, { eventId: row.id, type: row.event_type })
MessageBody: JSON.stringify({ ...row.payload, eventId: row.id, type: row.event_type }),
}));
// A crash here republishes this row on the next pass. That is the design:
// at-least-once publishing, deduplicated by eventId at the consumer.
}
if (rows.length > 0) {
await client.query(
"UPDATE outbox SET published_at = now() WHERE id = ANY($1)",
[rows.map((r) => r.id)]
);
}
await client.query("COMMIT");
return rows.length;
} catch (err) {
await client.query("ROLLBACK");
throw err;
} finally {
client.release();
}
}
// Poll on a short interval, and alarm on the age of the oldest unpublished row.
// A stopped relay is invisible otherwise: writes keep succeeding and nothing
// downstream happens.The schema
The partial index is the difference between a relay that stays fast at ten rows and at ten million. Keep published rows for a few days rather than deleting on publish: they are the cheapest possible answer to "did we emit that event, and what exactly did we send?", which is the question every integration argument comes down to.
CREATE TABLE outbox (
id bigserial PRIMARY KEY,
aggregate_type text NOT NULL, -- 'order'
aggregate_id text NOT NULL, -- the order id; useful as a partition key
event_type text NOT NULL, -- 'order.placed'
payload jsonb NOT NULL,
created_at timestamptz NOT NULL DEFAULT now(),
published_at timestamptz -- NULL until the relay has published it
);
-- The relay asks one question: what is unpublished. A partial index stays small
-- however large the table grows, and keeps the claim query off a sequential scan.
CREATE INDEX outbox_unpublished_idx ON outbox (id) WHERE published_at IS NULL;
-- Published rows are only kept for debugging and replay. Delete or move them, or
-- the table and its backups grow without bound.
DELETE FROM outbox
WHERE published_at IS NOT NULL
AND published_at < now() - interval '7 days';The consumer's side of the bargain
The reason the outbox works without distributed transactions: the producer guarantees at-least-once, the consumer supplies the deduplication, and together they give effectively-once. The composite key on (event_id, consumer) matters when several consumers read the same event — each needs its own record of having handled it, and sharing one row means the second consumer silently skips every event.
// The relay may publish the same row twice, so the consumer deduplicates on the
// id that came from the outbox table. It is stable across republishes, which a
// broker-assigned message id is not.
export async function handleOrderPlaced(message) {
const body = JSON.parse(message.Body);
const client = await pool.connect();
try {
await client.query("BEGIN");
const claim = await client.query(
`INSERT INTO consumed_events (event_id, consumer)
VALUES ($1, 'fulfilment')
ON CONFLICT (event_id, consumer) DO NOTHING`,
[body.eventId]
);
if (claim.rowCount === 0) {
await client.query("ROLLBACK");
return; // already handled by this consumer
}
await createFulfilmentJob(client, body);
await client.query("COMMIT");
} catch (err) {
await client.query("ROLLBACK");
throw err;
} finally {
client.release();
}
}
Discussion