Saved to Postgres, never queued: the transactional outbox for Node

Here's a bug that never shows up in tests. A request saves an order to Postgres, then calls queue.add('send-receipt', ...). The deploy restarts the pod between those two lines, or Redis blips for two seconds, and now you have a paid order that never gets a receipt, an invoice or a warehouse pick. Nothing errors. Nothing retries. The two writes were never one unit, and nobody notices until a customer emails.

I'm Krrish Kumar, a backend engineer. I wrote this with AI assistance for drafting and editing; the code and every number below come from a test script I ran against PostgreSQL 18.6 and Redis, with BullMQ 5.81.5 and 6.5.3. The fix is an old pattern called the transactional outbox, and this post builds a small version of it in Node, plus the three mistakes I made on the way.

There are two orders you can write in, and both are broken:

You can't wrap Redis and Postgres in one transaction. Even BullMQ 6, which can now keep its queues in PostgreSQL, doesn't help here: its backend interface "intentionally exposes no connection or transaction type", so queue.add() runs on its own connection, outside your transaction.

The outbox gets around this by not writing to the queue in the request at all. The request writes the order and a row describing the job in the same database transaction. A separate relay process copies outbox rows into the queue later. If the transaction commits, both rows exist. If it rolls back, neither does.

CREATE TABLE outbox (
  id           bigserial PRIMARY KEY,
  topic        text NOT NULL,
  payload      jsonb NOT NULL,
  created_at   timestamptz NOT NULL DEFAULT now(),
  published_at timestamptz
);

-- Only unpublished rows are indexed, so the relay's query stays cheap
-- even when the table holds millions of published rows.
CREATE INDEX outbox_unpublished ON outbox (id) WHERE published_at IS NULL;
js
export async function createOrder(pool, { customer, totalPaise }) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const { rows: [order] } = await client.query(
      'INSERT INTO orders (customer, total_paise) VALUES ($1, $2) RETURNING id',
      [customer, totalPaise],
    );
    await client.query(
      `INSERT INTO outbox (topic, payload) VALUES ('order.created', $1)`,
      [{ orderId: order.id, customer }],
    );
    await client.query('NOTIFY outbox'); // delivered only if the transaction commits
    await client.query('COMMIT');
    return order.id;
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

The NOTIFY is optional, and it's free: Postgres holds notifications sent inside a transaction until it commits, and drops them on rollback. The relay uses it to wake up immediately instead of waiting for its next poll.

export async function relayBatch(pool, queue, { limit = 100 } = {}) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const { rows } = await client.query(
      `SELECT id, topic, payload FROM outbox
        WHERE published_at IS NULL
        ORDER BY id
        LIMIT $1
        FOR UPDATE SKIP LOCKED`,
      [limit],
    );
    for (const row of rows) {
      await queue.add(row.topic, row.payload, { jobId: `outbox-${row.id}` });
    }
    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').catch(() => {});
    throw err;
  } finally {
    client.release();
  }
}

Three things carry the weight here:

FOR UPDATE SKIP LOCKED`` queue.add() calls between them, split 100/150/150/100.COMMIT, the transaction rolls back, the lock goes away, and the next run picks the same row again. That means the relay is jobId: outbox-<id>`` add() whose jobId already exists. I simulated the crash (throw after the adds, before the commit), ran the relay again, and the queue held one job, not two. One trade-off: the transaction stays open while you talk to Redis. With a batch of 100 that's a few milliseconds, but keep batches small so one slow Redis call doesn't hold locks for long.

The loop around it drains, then sleeps until a NOTIFY arrives or a poll timer fires, whichever comes first:

export async function runRelay(pool, queue, { pollMs = 5000, signal } = {}) {
  const listener = await pool.connect();
  let wake;
  const nudge = () => wake?.();
  listener.on('notification', nudge);
  await listener.query('LISTEN outbox');
  try {
    while (!signal?.aborted) {
      while ((await relayBatch(pool, queue)) > 0) {}
      await new Promise((resolve) => {
        wake = resolve;
        setTimeout(resolve, pollMs);
        signal?.addEventListener('abort', resolve, { once: true });
      });
    }
  } finally {
    listener.removeListener('notification', nudge);
    await listener.query('UNLISTEN outbox').catch(() => {});
    listener.release();
  }
}

With the poll timer set to 60 seconds, a new order was in the queue 9 ms after its commit, so NOTIFY did the work. Keep the poll anyway. A notification sent while the relay is reconnecting is simply lost, and the poll is what catches it.

My first relay remembered the highest id it had published and asked for WHERE id > $lastSeen. It looks fine, and it loses rows.

bigserial hands out ids when a row is inserted, not when its transaction commits. Transaction A inserts id 1 and is still working. Transaction B inserts id 2 and commits. The relay reads, sees only id 2, and sets lastSeen = 2. Then A commits. Id 1 is now in the table, and id > 2 will never return it. My test reproduced it: the first read saw [2], the second saw [], and row 1 was stuck forever.

Asking WHERE published_at IS NULL instead has no cursor to get ahead of. The same test with that query published both rows.

removeOnComplete: true The jobId dedupe only works while BullMQ still has the job. With removeOnComplete: true (or { count: 0 }), a completed job is deleted at once, and a later add() with the same jobId is treated as new. In my test the processor ran twice for outbox-1.

That matters for the outbox, because the duplicate add comes from a relay retry, and the first job may already be done by then. Keep completed jobs for longer than any retry could take:

const worker = new Worker('orders', processor, {
  connection,
  removeOnComplete: { age: 24 * 3600 }, // seconds
});

With that setting, a second add() of the same jobId inside the window was ignored, and the processor ran once. This narrows the window but doesn't close it. If your consumer sends money or email, have it check orderId before acting too.

I reached for a colon out of habit. BullMQ rejects it: Custom Id cannot contain :, because : is its Redis key separator. A bare number fails too, with Custom Id cannot be integers, so jobId: String(row.id) doesn't work either. A prefix with a dash (outbox-42) passes both checks, on BullMQ 5 and 6.

I hit mistakes 2 and 3 often enough that I wrote lint rules for them in eslint-plugin-bullmq, a small plugin I maintain. On a file that sets jobId: `outbox:${row.id}` together with removeOnComplete: true, it reports both:

5:52  error    BullMQ uses ':' as its Redis key separator, so a custom jobId containing ':' is rejected ...  bullmq/valid-job-id
5:72  warning  removeOnComplete: true deletes the job as soon as it finishes, so a later add() with the same jobId runs again ...  bullmq/no-jobid-dedupe-with-remove-on-complete

You don't need it for any of this. A code review checklist with the same two lines works too.

Published rows pile up. Delete them in batches from a daily job, so one big DELETE doesn't hold locks or bloat the table:

DELETE FROM outbox
 WHERE id IN (
   SELECT id FROM outbox
    WHERE published_at < now() - interval '7 days'
    LIMIT 5000
 );

Run it in a loop until it deletes 0 rows. Keep the retention longer than your BullMQ removeOnComplete age, so you can still trace a job back to its row.

If losing an occasional job is fine (analytics pings, cache warming), skip the outbox. If the work can live entirely in Postgres, a Postgres-backed job queue that shares your connection, such as graphile-worker or pg-boss, can enqueue inside your own transaction, and you get the same guarantee without a relay. The outbox is for when the job has to land somewhere else: Redis, Kafka, a webhook, another service.

WHERE published_at IS NULL ... FOR UPDATE SKIP LOCKED, not an id cursor.: and not a bare number.NOTIFY for speed, poll for safety. Disclosure: I'm Krrish Kumar, and I maintain eslint-plugin-bullmq. The test script behind the numbers here ran on PostgreSQL 18.6, BullMQ 5.81.5 and 6.5.3, ioredis and node-postgres.