Postgres as a Queue in 2026 — Boring, Reliable, Underrated

CREATE TABLE jobs (
    id          bigserial PRIMARY KEY,
    queue       text NOT NULL DEFAULT 'default',
    payload     jsonb NOT NULL,
    priority    int NOT NULL DEFAULT 0,
    run_at      timestamptz NOT NULL DEFAULT now(),
    attempts    int NOT NULL DEFAULT 0,
    locked_at   timestamptz,
    locked_by   text,
    created_at  timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX jobs_pickup ON jobs (queue, run_at, priority DESC)
    WHERE locked_at IS NULL;

async def enqueue(queue, payload, run_at=None, priority=0):
    await db.execute(
        "INSERT INTO jobs (queue, payload, run_at, priority) VALUES ($1, $2, $3, $4)",
        queue, payload, run_at or "now()", priority
    )

async with db.transaction():
    await db.execute("INSERT INTO orders ...")
    await enqueue("send_confirmation", {"order_id": order_id})

WITH next_job AS (
    SELECT id FROM jobs
    WHERE queue = $1 AND run_at <= now() AND locked_at IS NULL
    ORDER BY priority DESC, run_at
    LIMIT 1
    FOR UPDATE SKIP LOCKED
)
UPDATE jobs SET locked_at = now(), locked_by = $2, attempts = attempts + 1
WHERE id = (SELECT id FROM next_job)
RETURNING id, payload;

async def claim_job(queue, worker_id):
    return await db.fetchrow(SQL, queue, worker_id)

async def worker_loop(worker_id):
    while True:
        job = await claim_job("default", worker_id)
        if not job:
            await asyncio.sleep(1)
            continue
        try:
            await process(job["payload"])
            await db.execute("DELETE FROM jobs WHERE id = $1", job["id"])
        except Exception as e:
            await handle_failure(job, e)

async def handle_failure(job, error):
    if job["attempts"] < 5:
        delay = 2 ** job["attempts"] + random.uniform(0, 1)
        await db.execute(
            "UPDATE jobs SET locked_at = NULL, run_at = now() + interval '1 second' * $1 WHERE id = $2",
            delay, job["id"]
        )
    else:
        await db.execute(
            "INSERT INTO dead_jobs SELECT * FROM jobs WHERE id = $1", job["id"]
        )
        await db.execute("DELETE FROM jobs WHERE id = $1", job["id"])
        log.error(f"job {job['id']} dead-lettered: {error}")

-- Periodic cleanup
UPDATE jobs SET locked_at = NULL, locked_by = NULL
WHERE locked_at < now() - interval '5 minutes';

UPDATE jobs SET locked_at = now() WHERE id = $1 AND locked_by = $2;

-- On enqueue
NOTIFY jobs_default, '';

# Worker
async with db.acquire() as conn:
    await conn.execute("LISTEN jobs_default")
    while True:
        try:
            job = await claim_job(...)
            if job:
                await process(...)
            else:
                await conn.notifies.get(timeout=5)  # wakes on NOTIFY or after 5s
        except asyncio.TimeoutError:
            pass

async def claim_any(worker_id):
    return await db.fetchrow(SQL_WITH_ANY_QUEUE, worker_id)

await enqueue("cleanup", {}, run_at=tomorrow_3am)

async def send_email(payload):
    if await already_sent(payload["email_id"]):
        return
    await sendgrid.send(...)
    await mark_sent(payload["email_id"])