Skip to content

Building a Durable Job Queue on Postgres with asyncio

A job queue in the database you already run has one property no broker can offer: the job and the data it operates on commit together. No dual write, no orphan job for a rolled-back row, no lost job for a committed one. The usual objection is throughput, and it is worth putting a number on before accepting it — four asyncio workers against one local PostgreSQL drained 5,000 jobs in 2.50 s, about 2,001 per second, with 5,000 enqueues taking 0.05 s. That is more than most services will ever need, and it comes with a queue you can inspect, correct and report on using SQL.

Prerequisites

Throughput of a Postgres job queue 2 bars comparing enqueue, executemany with the others. Throughput of a Postgres job queue enqueue, executemany 5,000 rows in 0.05 s drain, 4 workers 2,001 jobs/s Trivial handlers, batches of 20 claimed per query, on one local PostgreSQL 18. Two thousand jobs a second covers most services; past that, a dedicated broker earns its keep.

1. Design the table around the claim query

CREATE TABLE jobs (
  id           bigserial PRIMARY KEY,
  kind         text NOT NULL,
  payload      jsonb NOT NULL,
  run_at       timestamptz NOT NULL DEFAULT now(),
  attempts     int NOT NULL DEFAULT 0,
  max_attempts int NOT NULL DEFAULT 3,
  locked_at    timestamptz,
  locked_by    text,
  status       text NOT NULL DEFAULT 'pending',
  last_error   text,
  created_at   timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX jobs_ready ON jobs (run_at, id) WHERE status = 'pending';

The partial index is the performance decision: it covers only rows a worker might claim, so its size tracks the backlog rather than the table's history. Without it, the claim query degrades as jobs accumulate — which is the single most common reason a database queue "stops scaling".

run_at gives you both delayed jobs and retry backoff with no extra machinery, attempts/max_attempts bound the retries, and locked_at/locked_by are what make a crashed worker recoverable.

Enqueueing is an insert, so it joins the transaction that produced the work:

async with conn.transaction():
    order_id = await conn.fetchval("INSERT INTO orders (...) VALUES (...) RETURNING id")
    await conn.execute("INSERT INTO jobs (kind, payload) VALUES ($1, $2)",
                       "send_receipt", json.dumps({"order_id": order_id}))

Verify: roll back that transaction and confirm neither the order nor the job exists.

Why the job and the data share a transaction 5 stages from business write to worker claims it. Why the job and the data share a transaction business write the row changes insert the job same transaction COMMIT both, or neither NOTIFY at commit a worker wakes worker claims it the data is there A queue in another system cannot make this promise; that is the whole argument for this design.

2. Claim a batch atomically

Claiming is one statement — the UPDATE selects, locks, marks and returns in a single round trip:

UPDATE jobs SET status='running', locked_at=now(), locked_by=$2, attempts=attempts+1
 WHERE id IN (
   SELECT id FROM jobs
    WHERE status='pending' AND run_at <= now()
    ORDER BY run_at, id
    LIMIT $1
    FOR UPDATE SKIP LOCKED)
 RETURNING id, kind, payload, attempts, max_attempts

FOR UPDATE SKIP LOCKED is what allows many workers with no coordination: each takes rows nobody else has locked rather than blocking on them. ORDER BY run_at, id keeps processing roughly in order, and the LIMIT is your batch size — 20 worked well in the measurements, amortising the round trip without holding many rows hostage if the worker dies.

Incrementing attempts at claim time rather than on failure is deliberate: a worker that crashes mid-job has still consumed an attempt, so a job that reliably kills its worker eventually reaches max_attempts instead of looping forever.

rows = await conn.fetch(CLAIM, batch_size, worker_name)

Verify: run several workers and confirm no job is processed twice.

One claim-and-run cycle 5 ordered steps. One claim-and-run cycle UPDATE ... RETURNING claim and mark in one statement FOR UPDATE SKIP LOCKED other workers take other rows run the handler with a timeout DELETE on success the table stays small defer with backoff on failure or mark it dead Claiming and marking in one statement is what makes a crash mid-claim harmless.

3. Complete, retry or bury

Success deletes the row. Keeping completed jobs is the second-most-common reason these tables get slow, and a separate job_history table (or nothing) is a better place for the audit trail.

async def complete(conn, job_id: int) -> None:
    await conn.execute("DELETE FROM jobs WHERE id = $1", job_id)


async def fail(conn, job_id, attempts, max_attempts, error: str) -> None:
    if attempts >= max_attempts:
        await conn.execute("UPDATE jobs SET status='dead', last_error=$2 WHERE id=$1",
                           job_id, error)
    else:
        backoff = datetime.timedelta(seconds=2 ** attempts)      # a timedelta, not a string
        await conn.execute(
            "UPDATE jobs SET status='pending', locked_by=NULL, locked_at=NULL, "
            "run_at = now() + $2, last_error=$3 WHERE id=$1",
            job_id, backoff, error)

Verified: a job with max_attempts=2 that raised came back as {'status': 'pending', 'attempts': 1, 'deferred': True, 'last_error': 'boom'} — deferred rather than dead, exactly as intended, and dead on the following failure.

Note the timedelta. Passing '2 seconds' as a string produces an asyncpg error that is not obviously about types:

asyncpg.exceptions.DataError: invalid input for query argument $2: '2 seconds'
('str' object has no attribute 'days')

Add jitter to the backoff when many jobs can fail together, for the reasons in retry and backoff strategies.

Verify: a failing job's run_at moves into the future and its attempts increments.

The state machine a job row moves through A grid of 5 rows by 2 columns. The state machine a job row moves through state entered when left when pending inserted, or a retry is scheduled a worker claims it running claimed with SKIP LOCKED it completes, fails or times out pending (deferred) it failed below max_attempts run_at passes dead attempts reached max_attempts a human looks at it deleted it succeeded never: the row is gone Keeping completed jobs as rows is the main reason these tables get slow; delete them.

4. Recover jobs from crashed workers

A worker that dies mid-job leaves its rows in running forever. A visibility timeout returns them:

UPDATE jobs SET status='pending', locked_by=NULL, locked_at=NULL
 WHERE status='running' AND locked_at < now() - $1::interval
 RETURNING id

Verified: a row locked ten minutes ago by dead-worker was reclaimed by a five-minute timeout. Run this periodically — every minute is typical — from any worker; it is idempotent and cheap.

The timeout must be comfortably longer than the slowest legitimate job, or you will reclaim work that is still running and process it twice. Where jobs vary enormously, store a per-job visibility_seconds and use that instead of one global value. Either way, handlers must be idempotent: this design is at-least-once, like every durable queue, and the reclaim is exactly the case that produces a duplicate.

Verify: kill a worker mid-batch and confirm its jobs are reprocessed after the timeout, not lost.

5. Wake workers with NOTIFY, poll as the safety net

Polling every second costs a query per worker per second and adds up to a second of latency. pg_notify inside the enqueue transaction removes the latency:

    await conn.execute("SELECT pg_notify('jobs', $1)", kind)     # delivered at commit

    while True:
        while await claim_and_run(pool):                          # drain everything ready
            pass
        with contextlib.suppress(TimeoutError):
            async with asyncio.timeout(1.0):                      # the poll is the backstop
                await woken.wait()
        woken.clear()

The poll is not redundant: notifications are lost while a listener is reconnecting, and delayed jobs (run_at in the future) are never announced by a notification at all — only a poll will pick them up when their time arrives.

Verify: with the listener disconnected, a newly enqueued job still runs within the poll interval.

Verification

A Postgres job queue is production-ready when:

  • Jobs are enqueued in the transaction that creates the work they describe.
  • Claims are atomic, one UPDATE ... RETURNING with SKIP LOCKED.
  • The partial index exists and the claim query uses it.
  • Failures back off and reach a dead state after max_attempts.
  • Stale running rows are reclaimed by a visibility timeout.
  • Completed jobs are deleted, and the table size is stable.
  • Handlers are idempotent, because reclaim and retry both duplicate.

Pitfalls & edge cases

  • No partial index. The claim query scans more of the table as history grows.
  • Keeping completed rows. The table and its indexes grow without bound; delete or archive.
  • A visibility timeout shorter than the slowest job. Running jobs are reclaimed and duplicated.
  • Passing intervals as strings. asyncpg wants a timedelta; the error message blames str.
  • SELECT then UPDATE in two statements. A crash between them leaves the job claimed by nobody.
  • One queue table for wildly different jobs. A flood of one kind starves the others; add a kind filter per worker pool, or separate tables.
  • Long-running jobs holding a transaction. Claim, commit, then work — never hold the claim transaction open across the handler.

Frequently Asked Questions

How do I build a job queue on PostgreSQL?

A jobs table with status, run_at and attempts columns, a partial index on the pending rows, and a claim query that updates rows selected with FOR UPDATE SKIP LOCKED and returns them. Workers claim a batch, run the handlers, delete on success and defer with backoff on failure.

How fast is a Postgres-backed job queue?

Fast enough for most services. Measured with four asyncio workers, batches of 20 and trivial handlers: 5,000 jobs drained in 2.50 s, about 2,001 per second, with 5,000 enqueues taking 0.05 s. Past a few thousand per second, a dedicated broker starts to earn its operational cost.

What does FOR UPDATE SKIP LOCKED do in a job queue?

It makes concurrent claiming lock-free: each worker's query skips rows another worker has already locked instead of waiting for them. Without it, workers serialise on the same rows or, worse, claim the same job twice.

How do I handle a worker that crashes mid-job?

Record locked_at when claiming and run a periodic query that resets rows whose lock is older than a visibility timeout. Verified with a job locked ten minutes earlier and a five-minute timeout, it was returned to pending. Set the timeout above your slowest job, and make handlers idempotent.

Should I keep completed jobs in the table?

No. Delete them on success and, if you need an audit trail, write it to a separate history table. Completed rows bloat the table and its indexes, which is the usual reason a database queue is said not to scale.