Transactional Outbox
Publishing an event and committing the state change that caused it are two writes to two systems, and nothing makes them one transaction. A crash between them either publishes an event for a change that rolled back, or commits a change whose event never went out.
The outbox makes them one write. The handler writes the state change and a row in an outbox table inside the same database transaction, so they commit or roll back together. A publisher worker then reads the pending rows, hands each to the operator's publish function, and marks it done. A crash anywhere leaves the row pending and the worker picks it up on restart.
Delivery is therefore at least once, never exactly once: a crash between publishing and marking the row republishes it. Consumers have to tolerate a repeat, which is what the event id is for. A row that fails maxAttempts times stops being retried and is counted as dead rather than retried forever.
b.outbox.create(opts) #
{
externalDb: object, // the b.externalDb namespace; required
table: string, // outbox table name; required
publisher: function, // async (event) → void; required
pollIntervalMs: number, // how often the worker looks for work
batchSize: number, // rows claimed per pass
maxAttempts: number, // failures before a row is counted dead
claimReclaimMs: number, // how long a claim survives an unresponsive worker
retryBackoff: object, // { initialMs, maxMs, factor }
envelope: string, // event envelope format
connectorName: string, // name recorded on published events
connectorVersion: string, // version recorded on published events
dbName: string, // database name recorded on published events
name: string, // this outbox's name, for audit rows
audit: boolean, // default true
}Open an outbox against a table and answer the handle to work it. The handle carries declareSchema(xdb?), which creates the table; enqueue(event, txn), which writes a row inside the caller's transaction; start() and stop() for the publisher worker; and pendingCount() and deadCount().
enqueue takes the transaction client b.externalDb.transaction hands its callback, so the row lands in the same transaction as the state change. A call without one is refused rather than writing a row that commits on its own, which is the case the outbox exists to avoid.
The worker claims a batch, publishes each row through publisher, and marks it. A claim that is not marked within claimReclaimMs is taken back, so a worker that dies mid-batch does not strand its rows. A row that fails retries on the backoff and stops at maxAttempts, after which deadCount() counts it.
var b = require("@blamejs/core");
var outbox = b.outbox.create({
externalDb: b.externalDb,
table: "outbox_events",
publisher: async function (event) { await bus.publish(event); },
});
await outbox.declareSchema();
await b.externalDb.transaction(async function (xdb) {
await xdb.query("UPDATE accounts SET balance = balance - $1 WHERE id = $2", [10, "a1"]);
await outbox.enqueue({ topic: "account.debited", payload: { id: "a1" } }, xdb);
});
await outbox.start();