A job queue on Postgres with SKIP LOCKED

The problem: a queue that agrees with the database

Snip's link creation now queues four jobs: fetch the preview, render the QR code, call the owner's webhook, update the search index (07.01 Why move work off the request path). With a separate queue service, the code looks like this:

await db.query("INSERT INTO links ...");
await queue.send({ type: "preview", code });   // what if this fails?

If the send fails after the insert commits, the link exists with no preview job, ever. Send first, and if the insert then fails, a worker fetches a preview for a link that doesn't exist. The two systems can't commit together.

If the jobs live in Postgres, in the same transaction as the link, both happen or neither does. River, a Postgres job queue, puts it simply: a job inserted in a transaction becomes visible to workers only after the transaction commits. For Snip's few hundred jobs a second, Postgres is easily fast enough, and it's one less system to run.

The table

CREATE TABLE jobs (
  id           bigserial   PRIMARY KEY,
  queue        text        NOT NULL,
  payload      jsonb       NOT NULL,
  status       text        NOT NULL DEFAULT 'ready',  -- ready, running, dead
  run_at       timestamptz NOT NULL DEFAULT now(),    -- not before this time
  attempts     int         NOT NULL DEFAULT 0,
  max_attempts int         NOT NULL DEFAULT 5,
  locked_until timestamptz,                           -- a running job's lease
  last_error   text
);

-- Workers only ever look for jobs that could run, in run_at order.
CREATE INDEX jobs_claimable ON jobs (queue, run_at) WHERE status IN ('ready', 'running');

Creating a link and its jobs is now one transaction:

BEGIN;
INSERT INTO links (code, long_url, owner_id) VALUES ('x7Kp', 'https://shop.example/spring', 4821);
INSERT INTO jobs (queue, payload) VALUES
  ('previews', '{"code": "x7Kp"}'),
  ('webhooks', '{"code": "x7Kp", "event": "link.created"}');
COMMIT;

Taking a job without fighting: SKIP LOCKED

Twenty workers asking "what's the oldest ready job?" at the same moment all find the same row. With a plain FOR UPDATE, nineteen of them wait for the first to finish with it. With FOR UPDATE SKIP LOCKED, each skips rows another transaction has locked and takes the next one. The PostgreSQL documentation is frank about the trade: skipping locked rows "provides an inconsistent view of the data", so it isn't for general use, but it suits "multiple consumers accessing a queue-like table". That's exactly this.

type Job = { id: string; payload: any; attempts: number; max_attempts: number };

/** Take the oldest runnable job, skipping any another worker has locked. */
async function claim(db: pg.Pool, queue: string, leaseSeconds = 300): Promise<Job | null> {
  const { rows } = await db.query(
    `UPDATE jobs
        SET status = 'running', attempts = attempts + 1,
            locked_until = now() + make_interval(secs => $2)
      WHERE id = (
        SELECT id FROM jobs
         WHERE queue = $1 AND run_at <= now()
           AND (status = 'ready' OR (status = 'running' AND locked_until < now()))
         ORDER BY run_at
         FOR UPDATE SKIP LOCKED
         LIMIT 1)
      RETURNING id, payload, attempts, max_attempts`,
    [queue, leaseSeconds],
  );
  return rows[0] ?? null;
}

async function complete(db: pg.Pool, job: Job) {
  await db.query("DELETE FROM jobs WHERE id = $1 AND attempts = $2", [job.id, job.attempts]);
}

async function fail(db: pg.Pool, job: Job, err: Error) {
  // Retry later with exponential backoff (2 s, 4 s, 8 s …, capped at an hour); give up after max_attempts.
  await db.query(
    `UPDATE jobs
        SET status = CASE WHEN attempts >= max_attempts THEN 'dead' ELSE 'ready' END,
            run_at = now() + make_interval(secs => least(power(2, attempts), 3600)),
            locked_until = NULL, last_error = $3
      WHERE id = $1 AND attempts = $2`,
    [job.id, job.attempts, err.message],
  );
}

export async function work(db: pg.Pool, queue: string, handle: (payload: any) => Promise<void>) {
  for (;;) {
    const job = await claim(db, queue);
    if (!job) {
      await new Promise((r) => setTimeout(r, 1000)); // nothing to do: poll again in a second
      continue;
    }
    try {
      await handle(job.payload);
      await complete(db, job);
    } catch (err) {
      await fail(db, job, err as Error);
    }
  }
}

A few details carry most of the reliability:

• Claiming is its own short transaction. The job is marked running and the transaction commits at once; the work itself happens outside any transaction. Holding a transaction open during a five-second preview fetch would hold locks and a database connection for five seconds per job.
• The lease. locked_until is the queue's version of SQS's visibility timeout (07.02 Queues vs logs). If a worker dies mid-job, nobody completes or fails it, and once the lease passes the claim query picks it up again.
• The attempts guard. complete and fail only touch the row if attempts still matches. A worker that was presumed dead and comes back late can't delete a job that another worker has since claimed.
• Backoff and a dead state. Failures retry after 2, 4, 8 … seconds; after max_attempts the job is dead, kept with its last error for a person to inspect: the table's own dead-letter queue.
• At least once, still. A job whose lease expires while its worker is merely slow runs twice. Handlers must be safe to repeat (07.03 Delivery guarantees and idempotent consumers), and the lease should be comfortably longer than the slowest job.

Tested

Run against PostgreSQL 16 with 8 workers in parallel:

Polling, and waking up faster

Idle workers poll once a second, which is cheap with the index but adds up to a second of delay. For faster pickup, the enqueuing transaction can run NOTIFY jobs and idle workers LISTEN jobs, waking immediately; keep the poll as a fallback, since notifications aren't stored.

Where it stops

A Postgres queue is a good default up to thousands of jobs a second. Beyond that, watch for:

• Table churn. Every job is inserted, updated and deleted, leaving dead row versions for autovacuum to clean up (04.05 MVCC). A queue table that vacuum can't keep up with gets slower.
• Load on the primary. Every claim is a write on the database that also serves your product.
• Event streams. 100,000 click events a second, read by several systems, is a job for a log (07.02 Queues vs logs), not a jobs table.

Recap

Interview questions