Reliable Background Jobs with Nothing but PostgreSQL

Most applications have work that should not happen inside an HTTP request: sending an email, generating a PDF, calling a slow third-party API. The reflex is to add a message broker. Before that, it is worth knowing how far a single table in the database you already run will take you, because the answer is usually "much further than you need", and it comes with one property brokers cannot give you: the job and the business data commit in the same transaction.

This post builds that queue step by step. All SQL is PostgreSQL; the worker is Python with psycopg 3, but every line maps directly to any other driver.

The table

CREATE TYPE job_status AS ENUM ('queued', 'running', 'done', 'dead');

CREATE TABLE jobs (
  id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  kind            text        NOT NULL,            -- which handler runs it
  payload         jsonb       NOT NULL,
  status          job_status  NOT NULL DEFAULT 'queued',
  run_at          timestamptz NOT NULL DEFAULT now(),   -- earliest start; moved forward on retry
  locked_until    timestamptz,                          -- visibility timeout while running
  attempts        int         NOT NULL DEFAULT 0,
  max_attempts    int         NOT NULL DEFAULT 8,
  idempotency_key text        UNIQUE,                   -- optional: dedupe enqueues
  worker          text,
  last_error      text,
  created_at      timestamptz NOT NULL DEFAULT now(),
  finished_at     timestamptz
);

-- The claim query only ever looks at live rows, so index only those.
CREATE INDEX jobs_claim_idx ON jobs (run_at)
  WHERE status IN ('queued', 'running');

The partial index is what keeps the claim query fast as finished jobs pile up. Finished jobs accumulate (or get archived by a nightly job), but the index the workers hit only contains the handful of rows that are queued or running.

Claiming a job without a race

The problem with a naive SELECT ... WHERE status = 'queued' LIMIT 1 followed by an UPDATE is that two workers select the same row. The classic fix, SELECT ... FOR UPDATE, makes the second worker wait for the first worker's transaction, which serialises your whole worker pool on one row.

FOR UPDATE SKIP LOCKED is the primitive built for queues: rows that are already locked by another transaction are skipped as if they did not exist, so each worker gets a different row with no waiting.

UPDATE jobs
   SET status       = 'running',
       locked_until = now() + make_interval(secs => 30),
       attempts     = attempts + 1,
       worker       = 'host-a:4121'
 WHERE id = (
       SELECT id
         FROM jobs
        WHERE (status = 'queued'  AND run_at <= now())
           OR (status = 'running' AND locked_until < now())   -- expired claim: owner probably died
        ORDER BY run_at
        FOR UPDATE SKIP LOCKED
        LIMIT 1)
RETURNING id, kind, payload, attempts, max_attempts;

One statement does the select, the lock, and the state change. The transaction around it is a few milliseconds long and is committed before the handler runs; the row is protected afterwards not by a lock but by locked_until.

Two workers run the same claim statement a few milliseconds apart and get different rows. Neither waits on the other.

Visibility timeouts

Why not keep the row locked for the duration of the handler? Because a lock dies with its connection. If the worker process is killed mid-handler, PostgreSQL releases the lock and the row is immediately claimable again, which is what you want. But if the handler legitimately takes four minutes, holding a transaction open that long blocks vacuum, pins a connection, and invites idle-in-transaction timeouts.

So the claim sets locked_until = now() + 30 s and commits. The row is invisible to the claim query until then. Three things can happen:

The handler finishes within 30 s and the worker writes status = 'done'.
The handler needs longer, and the worker extends locked_until every 10 s from a heartbeat thread. A worker that is alive keeps its claim; a worker that is dead stops heartbeating and the claim expires.
The worker crashes. Nothing is written, locked_until passes, and the OR (status = 'running' AND locked_until < now()) branch of the claim query hands the job to someone else.

WARNING

Never run the handler inside the claim transaction. It feels safer, but it turns every slow API call into a long-lived transaction holding a row lock and a connection, and a worker crash then rolls back the attempts increment, so a poison job retries forever. Claim, commit, work, then record the outcome in a second short transaction.

Retries with backoff and jitter

A failed job should not be retried immediately, and a hundred jobs that failed because a downstream API blipped should not all retry at the same instant. Exponential backoff solves the first problem; jitter solves the second. The delay before attempt $n + 1$ is

d_{n} = uniform (0, min (c, b \cdot 2^{n - 1}))

with base $b = 2$ s and cap $c = 300$ s. This is "full jitter": the deterministic exponential value becomes the ceiling and the actual delay is drawn uniformly below it. It spreads retries across the whole window instead of clustering them at the end.

Scatter plot of jittered retry delays per attempt number under a stepped exponential ceiling that saturates at 300 seconds — The dark line is the exponential ceiling; each dot is one sampled delay. From attempt 9 the ceiling hits the 300 s cap and retries are spread over the whole five-minute window.

With max_attempts = 8 there are seven delays between runs, with ceilings of 2, 4, 8, 16, 32, 64 and 128 s. Each delay is uniform below its ceiling, so the expected total wait before a job is declared dead is half the sum, about 127 s, and the worst case is 254 s. That is enough to ride out a short blip; to survive a longer incident, raise max_attempts (attempts 9 and 10 add ceilings of 256 s and the 300 s cap) rather than the base.

UPDATE jobs
   SET status       = CASE WHEN attempts >= max_attempts THEN 'dead' ELSE 'queued' END,
       run_at       = now() + make_interval(secs => $1),   -- the jittered delay
       locked_until = NULL,
       last_error   = $2
 WHERE id = $3;

Dead jobs are not deleted. They are the queue's error log, and a dead count greater than zero is the alert that matters most.

Idempotent handlers

Every guarantee above is at least once: a job whose worker died after the side effect but before status = 'done' will run again. Exactly-once delivery is not available; exactly-once effect is, and it is the handler's job.

Three patterns cover almost every case:

Check state before acting. Read welcome_sent_at and return early if it is set. The second run becomes a no-op.
Pass your own idempotency key downstream. Payment providers and email APIs accept one; derive it from the job id so a retry is recognised as the same request.
Make the write itself idempotent. INSERT ... ON CONFLICT DO NOTHING, UPDATE ... WHERE status = 'pending', or an upsert keyed on something the job controls.

The idempotency_key column on jobs handles the producer side: enqueueing "send invoice 42" twice inserts one row.

The worker

The complete worker is about 120 lines and leaves out the heartbeat thread. The claim and retry statements are the ones above; the loop around them is deliberately boring.

def run_one(conn) -> bool:
    with conn.transaction():
        job = conn.execute(CLAIM, {"lock": LOCK_SECONDS, "worker": WORKER}).fetchone()
    if job is None:
        return False
    try:
        HANDLERS[job["kind"]](conn, job["payload"])
        conn.execute(DONE, (job["id"],))
        conn.commit()
    except Exception as exc:  # any failure is a retry decision, not a crash
        conn.rollback()
        delay = backoff(job["attempts"])
        conn.execute(RETRY, {"delay": delay, "error": repr(exc)[:2000], "id": job["id"]})
        conn.commit()
        log.warning("job %s failed (attempt %s/%s), retry in %.1fs: %r",
                    job["id"], job["attempts"], job["max_attempts"], delay, exc)
    return True


def backoff(attempt: int) -> float:
    """Full jitter: uniform(0, min(cap, base * 2^(attempt-1)))."""
    return random.uniform(0, min(CAP, BASE * 2 ** (attempt - 1)))

Note the shape of the error path: the handler's partial writes are rolled back, then the retry bookkeeping is committed on its own. If the handler had already committed something through a different connection, that is exactly the at-least-once case the idempotency section is for.

The complete worker.py

"""Minimal PostgreSQL-backed job worker: claim with SKIP LOCKED, retry with full-jitter backoff."""
import logging
import os
import random
import signal
import socket
import time

import psycopg
from psycopg.rows import dict_row
from psycopg.types.json import Jsonb

log = logging.getLogger("worker")
DSN = os.environ["DATABASE_URL"]
WORKER = f"{socket.gethostname()}:{os.getpid()}"
LOCK_SECONDS = 30          # visibility timeout: how long a claim is honoured without a heartbeat
BASE, CAP = 2.0, 300.0     # backoff parameters (seconds)
IDLE_SLEEP = 1.0

CLAIM = """
UPDATE jobs
   SET status = 'running',
       locked_until = now() + make_interval(secs => %(lock)s),
       attempts = attempts + 1,
       worker = %(worker)s
 WHERE id = (
       SELECT id
         FROM jobs
        WHERE (status = 'queued'  AND run_at <= now())
           OR (status = 'running' AND locked_until < now())
        ORDER BY run_at
        FOR UPDATE SKIP LOCKED
        LIMIT 1)
RETURNING id, kind, payload, attempts, max_attempts;
"""

DONE = "UPDATE jobs SET status = 'done', finished_at = now(), locked_until = NULL WHERE id = %s"

RETRY = """
UPDATE jobs
   SET status       = CASE WHEN attempts >= max_attempts THEN 'dead' ELSE 'queued' END,
       run_at       = now() + make_interval(secs => %(delay)s),
       locked_until = NULL,
       last_error   = %(error)s
 WHERE id = %(id)s
"""

HANDLERS = {}


def handler(kind):
    def register(fn):
        HANDLERS[kind] = fn
        return fn
    return register


@handler("send_welcome_email")
def send_welcome_email(conn, payload):
    # Idempotent by construction: the "already sent" state is checked before
    # any side effect, and the provider call would carry its own key.
    user_id = payload["user_id"]
    row = conn.execute("SELECT welcome_sent_at FROM users WHERE id = %s", (user_id,)).fetchone()
    if row is None or row["welcome_sent_at"] is not None:
        return
    # email_provider.send(idempotency_key=f"welcome-{user_id}", ...)
    conn.execute("UPDATE users SET welcome_sent_at = now() WHERE id = %s", (user_id,))


def backoff(attempt: int) -> float:
    """Full jitter: uniform(0, min(cap, base * 2^(attempt-1)))."""
    return random.uniform(0, min(CAP, BASE * 2 ** (attempt - 1)))


def run_one(conn) -> bool:
    with conn.transaction():
        job = conn.execute(CLAIM, {"lock": LOCK_SECONDS, "worker": WORKER}).fetchone()
    if job is None:
        return False
    try:
        HANDLERS[job["kind"]](conn, job["payload"])
        conn.execute(DONE, (job["id"],))
        conn.commit()
    except Exception as exc:  # noqa: BLE001 - any failure is a retry decision, not a crash
        conn.rollback()
        delay = backoff(job["attempts"])
        conn.execute(RETRY, {"delay": delay, "error": repr(exc)[:2000], "id": job["id"]})
        conn.commit()
        log.warning("job %s failed (attempt %s/%s), retry in %.1fs: %r",
                    job["id"], job["attempts"], job["max_attempts"], delay, exc)
    return True


def main():
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    stop = False

    def on_signal(signum, _frame):
        nonlocal stop
        stop = True          # finish the current job, then exit: no half-done work is left locked
        log.info("received %s, draining", signal.Signals(signum).name)

    signal.signal(signal.SIGTERM, on_signal)
    signal.signal(signal.SIGINT, on_signal)

    with psycopg.connect(DSN, row_factory=dict_row, autocommit=False) as conn:
        while not stop:
            if not run_one(conn):
                time.sleep(IDLE_SLEEP)


def enqueue(conn, kind: str, payload: dict, *, key: str | None = None, delay_s: float = 0):
    """Call this inside the business transaction so the job and the data commit together."""
    conn.execute(
        """INSERT INTO jobs (kind, payload, run_at, idempotency_key)
           VALUES (%s, %s, now() + make_interval(secs => %s), %s)
           ON CONFLICT (idempotency_key) DO NOTHING""",
        (kind, Jsonb(payload), delay_s, key),
    )


if __name__ == "__main__":
    main()

Enqueue in the same transaction

This is the property that justifies the whole approach. When a request creates an order and needs an email sent, the INSERT INTO jobs runs in the same transaction as the INSERT INTO orders. Either both commit or neither does. There is no window where the order exists but the job was lost, or the job exists for an order that was rolled back. With an external broker you need the outbox pattern to get this; here the table is the outbox.

Lower latency with LISTEN/NOTIFY

Polling every second is fine for most workloads and costs one cheap index lookup per worker per second. When sub-second pickup matters, add a trigger that notifies on insert and have idle workers wait on the channel instead of sleeping:

CREATE FUNCTION jobs_notify() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
  PERFORM pg_notify('jobs', NEW.kind);
  RETURN NEW;
END $$;

CREATE TRIGGER jobs_notify AFTER INSERT ON jobs
  FOR EACH ROW EXECUTE FUNCTION jobs_notify();

The worker runs LISTEN jobs once and replaces time.sleep(IDLE_SLEEP) with a wait on conn.notifies(timeout=IDLE_SLEEP). Keep the poll as a fallback: notifications are not queued for disconnected listeners, so a worker that reconnects must still look at the table.

What to watch

Signal	Query	Means
queue depth	`SELECT count(*) FROM jobs WHERE status = 'queued' AND run_at <= now()`	work waiting right now
oldest ready job	`SELECT now() - min(run_at) FROM jobs WHERE status = 'queued'`	workers are behind, or stuck
expired claims	`... WHERE status = 'running' AND locked_until < now()`	workers are dying or not heartbeating
dead jobs	`SELECT count(*) FROM jobs WHERE status = 'dead'`	something needs a human
table bloat	`pg_stat_user_tables.n_dead_tup` for `jobs`	archive finished rows more often

Bloat is the easiest of these to miss. A queue table is update-heavy, which means dead tuples; make sure autovacuum is keeping up, and move done rows older than a week into an archive table so the live table stays small.