A Durable PostgreSQL Job Queue with SKIP LOCKED

You do not always need Kafka, RabbitMQ, or a managed queue. If an application already depends on PostgreSQL and the workload is modest, a jobs table can be the simplest reliable option. The useful version is more than a table plus a polling loop: it needs safe concurrent claiming, idempotent enqueueing, retries, leases, and a dead-letter state.

This note builds that version. The queue is deliberately small. It is suitable for background email, webhook delivery, document processing, and similar work where PostgreSQL is already part of the system.

Start with the invariants

The code is easier to reason about if the rules are explicit:

• A ready job is claimed by at most one worker at a time.
• The database transaction used to claim a job is short.
• A worker crash does not leave a job stuck forever.
• Retry timing is stored with the job, not in worker memory.
• Duplicate enqueue requests can collapse to one row.
• The handler may run more than once, so side effects must be idempotent.

The last point matters most. This design provides at-least-once execution. Exactly-once delivery is not something a queue table can promise across arbitrary external side effects.

Create the table

CREATE TABLE jobs (
    id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    queue           text NOT NULL DEFAULT 'default',
    payload         jsonb NOT NULL,
    status          text NOT NULL DEFAULT 'ready'
                    CHECK (status IN ('ready', 'running', 'done', 'dead')),
    attempts        integer NOT NULL DEFAULT 0,
    max_attempts    integer NOT NULL DEFAULT 5,
    run_at          timestamptz NOT NULL DEFAULT now(),
    locked_at       timestamptz,
    locked_by       text,
    idempotency_key text UNIQUE,
    last_error      text,
    created_at      timestamptz NOT NULL DEFAULT now(),
    updated_at      timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX jobs_ready_idx
    ON jobs (queue, run_at, id)
    WHERE status = 'ready';

The partial index only contains rows workers can claim. Ordering by run_at, id makes delayed retries work and gives old jobs a stable position. The index does not guarantee strict FIFO once several workers are involved, but it avoids a full-table scan on every poll.

Enqueue idempotently

An idempotency key should describe the side effect, not the HTTP request. For example, invoice:1842:send-receipt is better than a random UUID generated by the caller.

INSERT INTO jobs (queue, payload, idempotency_key)
VALUES ('mail', '{"invoice_id": 1842}', 'invoice:1842:send-receipt')
ON CONFLICT (idempotency_key) DO NOTHING
RETURNING id;

If callers need the existing identifier, use ON CONFLICT ... DO UPDATE with a harmless assignment and return the row. Avoid changing payloads silently for an existing key; that makes retries difficult to audit.

Claim one job without blocking other workers

FOR UPDATE SKIP LOCKED is the key operation. Each worker ignores rows already locked by another worker and claims a different candidate.

WITH candidate AS (
    SELECT id
    FROM jobs
    WHERE status = 'ready'
      AND queue = %(queue)s
      AND run_at <= now()
    ORDER BY run_at, id
    FOR UPDATE SKIP LOCKED
    LIMIT 1
)
UPDATE jobs AS j
SET status = 'running',
    locked_at = now(),
    locked_by = %(worker_id)s,
    attempts = attempts + 1,
    updated_at = now()
FROM candidate
WHERE j.id = candidate.id
RETURNING j.*;

Run this statement in a transaction and commit immediately. Do not hold the row lock while the job runs. The running state and lease fields are what protect the job after the claim transaction ends.

A small Python worker

Install Psycopg 3 with pip install "psycopg[binary]", then keep one connection per worker process.

import json
import os
import socket
import time

import psycopg
from psycopg.rows import dict_row

CLAIM_SQL = """\
WITH candidate AS (
    SELECT id FROM jobs
    WHERE status = 'ready' AND queue = %s AND run_at <= now()
    ORDER BY run_at, id
    FOR UPDATE SKIP LOCKED
    LIMIT 1
)
UPDATE jobs AS j
SET status = 'running', locked_at = now(), locked_by = %s,
    attempts = attempts + 1, updated_at = now()
FROM candidate
WHERE j.id = candidate.id
RETURNING j.*
"""

worker_id = f"{socket.gethostname()}:{os.getpid()}"
conn = psycopg.connect(os.environ["DATABASE_URL"], row_factory=dict_row)

def claim(queue):
    with conn.transaction():
        return conn.execute(CLAIM_SQL, (queue, worker_id)).fetchone()

def complete(job_id):
    with conn.transaction():
        conn.execute(
            """UPDATE jobs SET status = 'done', locked_at = NULL,
               locked_by = NULL, updated_at = now()
               WHERE id = %s AND locked_by = %s""",
            (job_id, worker_id),
        )

def fail(job, error):
    delay = min(300, 2 ** job["attempts"])
    with conn.transaction():
        conn.execute(
            """UPDATE jobs
               SET status = CASE WHEN attempts >= max_attempts
                                 THEN 'dead' ELSE 'ready' END,
                   run_at = now() + (%s * interval '1 second'),
                   locked_at = NULL, locked_by = NULL,
                   last_error = %s, updated_at = now()
               WHERE id = %s AND locked_by = %s""",
            (delay, str(error)[:2000], job["id"], worker_id),
        )

while True:
    job = claim("mail")
    if job is None:
        time.sleep(0.5)
        continue

    try:
        payload = job["payload"]
        print("send receipt", json.dumps(payload))
        complete(job["id"])
    except Exception as exc:
        fail(job, exc)

The ownership predicate in complete and fail prevents a stale worker from updating a job that has been reclaimed. In a real handler, use the idempotency key when calling the downstream system too.

Recover abandoned leases

A worker can die after the claim commits. A periodic sweeper should return expired jobs to the ready state:

UPDATE jobs
SET status = 'ready',
    run_at = now(),
    locked_at = NULL,
    locked_by = NULL,
    updated_at = now()
WHERE status = 'running'
  AND locked_at < now() - interval '5 minutes';

The timeout must exceed normal job duration. If runtime varies widely, add a lease heartbeat that updates locked_at, and reclaim only jobs whose heartbeat has expired. A database advisory lock can ensure only one sweeper runs at a time, though the update itself is safe to run concurrently.

Test the failure paths

A queue is not tested when one worker successfully processes one job. Run the cases that break the invariants:

1. Start ten workers and enqueue one job. Confirm only one claim succeeds.
2. Kill a worker after claim and before completion. Confirm the sweeper makes the job runnable again.
3. Enqueue the same idempotency key concurrently. Confirm one row exists.
4. Make the handler fail repeatedly. Confirm exponential delay and eventual dead state.
5. Run a slow job past the lease. Confirm heartbeat or timeout policy prevents unsafe reclamation.

Know when to move on

This pattern is easy to operate because it reuses PostgreSQL transactions, backups, and monitoring. Move to a dedicated broker when queue traffic competes with application queries, when you need very high fan-out or partitioned ordering, or when retention and replay become product requirements. Until then, a small queue you can inspect with SQL is often a good tool.