Building a job queue with SELECT FOR UPDATE SKIP LOCKED

Claim work with select(Job).where(...).with_for_update(skip_locked=True).limit(n), mark the rows running and commit before doing the work — SKIP LOCKED lets each worker step over rows others hold, so many workers share one table without duplicating or blocking. This guide belongs to transaction isolation and commit strategies.

Quick Answer

Two workers selecting the same "oldest pending job" either duplicate the work or block on each other. SKIP LOCKED makes them skip rows another transaction has locked.

Two workers, one job Left: both workers select the oldest pending job, both see the same row, and both process it — or one blocks on the other lock and waits. Right: FOR UPDATE SKIP LOCKED makes the second worker step over rows the first has locked, so each gets a different job and neither waits. SELECT ... WHERE status = pending both workers see job 1 both process it duplicate work or one blocks on the lock FOR UPDATE SKIP LOCKED worker A locks job 1 worker B skips it, takes job 2 no duplicates, no waiting scales with workers SKIP LOCKED is what turns a table into a queue: readers step over what others hold.

Before — two workers, one job:

from sqlalchemy import select

from shop.models import Job


async def claim_one(session) -> Job | None:
    job = await session.scalar(
        select(Job).where(Job.status == "pending").order_by(Job.run_after).limit(1)
    )
    if job is not None:
        job.status = "running"
        await session.commit()
    return job
# Both workers read job 1 before either commits, and both process it.

After — claim a batch, commit, then work:

import datetime as dt

from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import async_sessionmaker

from shop.models import Job


async def claim(Session: async_sessionmaker, worker: str, size: int = 5) -> list[int]:
    now = dt.datetime.now(dt.UTC)
    async with Session.begin() as session:          # transaction 1: short
        ids = list(await session.scalars(
            select(Job.id)
            .where(Job.status == "pending", Job.run_after <= now)
            .order_by(Job.run_after)
            .limit(size)
            .with_for_update(skip_locked=True)
        ))
        if not ids:
            return []
        await session.execute(
            update(Job)
            .where(Job.id.in_(ids))
            .values(status="running", locked_at=now, locked_by=worker)
        )
    return ids                                      # committed: the claim is durable

Committing the claim before doing the work is what makes the rest safe: the locks are released, the claim is visible to every other worker, and the work itself can take as long as it needs — including awaiting an external service — without holding a transaction open.

Execution Context & Async Workflow Integration

SELECT ... FOR UPDATE takes a row lock on every row it returns, so a second transaction selecting the same rows waits. SKIP LOCKED changes that: rows already locked are simply not returned. With an ORDER BY and a LIMIT, each worker gets the first rows nobody else holds, which is precisely the queue semantics a table otherwise lacks.

One cycle, two transactions Five steps. A short transaction selects a batch of pending jobs with FOR UPDATE SKIP LOCKED, marks them running, and commits — releasing the locks and making the claim visible to everyone. The work then happens outside any transaction, so nothing is held while an external call runs. A second short transaction records the outcome. A crash between them leaves a job marked running, which a reaper returns to pending after a timeout. BEGIN; SELECT ... FOR UPDATE SKIP LOCKED a batch of pending jobs locked UPDATE ... SET status = running; COMMIT the claim is durable locks released do the work outside any transaction external calls are safe here BEGIN; record the outcome; COMMIT done or failed short a reaper requeues stuck jobs running for too long Committing the claim is what makes the work safe to do outside a transaction.

The transaction structure matters as much as the locking. Three properties follow from claiming in one short transaction and working outside it.

No lock is held during the work. A worker that claims and then processes inside the same transaction is idle in transaction for the duration — holding locks, pinning a snapshot and occupying a connection while it awaits an HTTP call, which is the failure described in detecting idle-in-transaction sessions from async code.

The claim is durable. Because the status = running update is committed, a crash leaves visible evidence: a row marked running with a locked_at timestamp. That is what a reaper can act on. If the claim were only a lock, a crash would release it silently and the job would be picked up again immediately — which sounds convenient and makes "how many times has this been attempted" unanswerable.

The guarantee is at-least-once. A worker can crash after doing the work and before recording the outcome, so the job is requeued and the work repeats. That is inherent to any queue without distributed transactions, and it means handlers must be idempotent — an idempotency key checked before the effect, as in handling IntegrityError on concurrent inserts.

Under async, one detail is worth spelling out. Claiming n jobs and processing them concurrently means n concurrent units of work, each of which should open its own session — the rule from running concurrent queries with AsyncSession. A worker that claims five jobs and processes them in a TaskGroup therefore needs six connections at peak, not one, and the pool has to account for it.

Index design decides whether the queue stays fast as the table grows. The claim query filters on status and orders by run_after, and a partial index holding only pending rows stays small enough to remain cached whatever the size of the completed history:

from sqlalchemy import Index, text

Index(
    "ix_jobs_pending_claim",
    "run_after",
    postgresql_where=text("status = 'pending'"),
)

That is the partial-index technique from creating partial and expression indexes from SQLAlchemy models, and for a queue it is the difference between constant-time claims and a scan that grows with history.

Resolving Warnings, Errors & Common Mistakes

Exact error or symptomRoot CauseProduction Fix
Two workers process the same jobNo FOR UPDATE, or no SKIP LOCKED, or the claim was not committed.Claim with with_for_update(skip_locked=True) and commit it.
Workers block instead of skippingwith_for_update() without skip_locked=True.Add skip_locked=True.
ProgrammingError: FOR UPDATE is not allowed with aggregate functionsThe claim query has a GROUP BY or an aggregate.Select the rows or ids plainly; aggregate separately.
FOR UPDATE cannot be applied to the nullable side of an outer joinThe claim query outer-joins another table.Claim from the job table alone; join afterwards.
Jobs stuck in running foreverA worker crashed after claiming.A reaper that requeues jobs whose locked_at is older than a timeout.
A failing job retries foreverNo attempt counter or maximum.Increment attempts; move to failed at the limit.
Claims slow down as the table growsA full index on status, or none.A partial index on pending rows.
The pool is exhausted with a handful of workersEach worker processes several jobs concurrently, each needing a session.Size the pool for workers times concurrency.
Four columns that matter Four tiles. status, with an index, is what the claim query filters on. run_after lets a job be scheduled or retried with backoff. attempts bounds retries so a permanently failing job stops. And locked_at records when a claim was made, which is what lets a reaper find jobs abandoned by a crashed worker. status pending, running, done, failed indexed, partially run_after scheduling and backoff ordered by this attempts bounds retries and a max locked_at when the claim was made the reaper reads it A partial index on pending rows keeps the queue fast whatever the size of the history.

The reaper is the piece most often missing, and without it a crash removes a job from circulation permanently:

import datetime as dt

from sqlalchemy import update
from sqlalchemy.ext.asyncio import async_sessionmaker

from shop.models import Job

LOCK_TIMEOUT = dt.timedelta(minutes=10)
MAX_ATTEMPTS = 5


async def requeue_stuck(Session: async_sessionmaker) -> int:
    cutoff = dt.datetime.now(dt.UTC) - LOCK_TIMEOUT
    async with Session.begin() as session:
        result = await session.execute(
            update(Job)
            .where(
                Job.status == "running",
                Job.locked_at < cutoff,
                Job.attempts < MAX_ATTEMPTS,
            )
            .values(
                status="pending",
                attempts=Job.attempts + 1,
                locked_at=None,
                locked_by=None,
            )
        )
        # Jobs that have exhausted their attempts stop being retried.
        await session.execute(
            update(Job)
            .where(Job.status == "running", Job.locked_at < cutoff,
                   Job.attempts >= MAX_ATTEMPTS)
            .values(status="failed")
        )
    return result.rowcount

The timeout has to exceed the longest legitimate job, or the reaper will requeue work that is still running — producing exactly the duplicate processing SKIP LOCKED prevented. Where job durations vary widely, a heartbeat is the better mechanism: the worker updates locked_at periodically, and the reaper's timeout can then be short.

The outer-join restriction catches people enriching the claim query. FOR UPDATE cannot lock rows on the nullable side of an outer join, so the claim must select from the job table alone and any joined data fetched afterwards — which is also better for the index, since the claim stays a single-table index scan.

Advanced: Batching, Priorities and Backoff

Three refinements cover most real requirements.

Three failure modes Three modes. A worker crashes after claiming, leaving a job marked running forever — a reaper requeues jobs whose lock is older than a timeout. A job fails repeatedly — an attempt counter and a maximum move it to failed rather than looping. And the work is not idempotent, so a requeued job could double its effect — an idempotency key makes the second attempt safe. a worker crashes after claiming a reaper requeues jobs running longer than a timeout, incrementing attempts a job fails repeatedly attempts and a maximum, with exponential backoff in run_after the work is not idempotent an idempotency key the handler checks, because at-least-once is the guarantee

Batch size. Claiming several jobs at once amortises the claim transaction, and claiming too many leaves work parked in one worker while others idle. A batch of a few, processed concurrently with a bounded TaskGroup, is the usual balance:

import asyncio

from shop.db import Session
from shop.jobs import claim, handle, record_outcome


async def run_batch(worker: str, size: int = 5, concurrency: int = 5) -> int:
    ids = await claim(Session, worker, size=size)
    if not ids:
        return 0
    limit = asyncio.Semaphore(concurrency)

    async def process(job_id: int) -> None:
        async with limit:
            try:
                await handle(job_id)                     # its own session inside
            except Exception as exc:                     # noqa: BLE001 — recorded, not swallowed
                await record_outcome(job_id, "failed", str(exc))
            else:
                await record_outcome(job_id, "done", None)

    async with asyncio.TaskGroup() as tg:
        for job_id in ids:
            tg.create_task(process(job_id))
    return len(ids)

Each process call opens its own session, so the pool needs room for concurrency sessions plus the claim — the arithmetic from setting up asyncpg pool size for high concurrency.

Priorities are an extra ORDER BY column, and the partial index should lead with it so the claim stays an index scan:

from sqlalchemy import Index, text

Index(
    "ix_jobs_pending_priority",
    "priority", "run_after",
    postgresql_where=text("status = 'pending'"),
)

    .order_by(Job.priority.desc(), Job.run_after)

Beware starvation: a steady stream of high-priority work can mean low-priority jobs are never claimed. Either bound the priority range, or occasionally claim the oldest job regardless of priority.

Retry backoff belongs in run_after, which the claim query already filters on, so a failed job simply becomes invisible until its next attempt is due:

import datetime as dt

from sqlalchemy import update

from shop.models import Job


async def fail_with_backoff(session, job_id: int, attempts: int, error: str) -> None:
    delay = dt.timedelta(seconds=min(2 ** attempts, 3600))
    await session.execute(
        update(Job)
        .where(Job.id == job_id)
        .values(
            status="pending",
            attempts=attempts + 1,
            run_after=dt.datetime.now(dt.UTC) + delay,
            last_error=error[:1000],
            locked_at=None,
            locked_by=None,
        )
    )

Storing last_error is worth the column: a failed-jobs table with the reason in it is the difference between a queue you can operate and one you can only restart.

Notification removes polling latency. LISTEN/NOTIFY lets a worker wake immediately when a job is enqueued, but asyncpg's listener needs a dedicated connection outside SQLAlchemy's pool, so most implementations keep polling as the baseline and treat notification as an optimisation. A poll interval of a second with a partial index costs almost nothing.

Deciding Between a Table and a Broker

A database queue is a real option, not a compromise, and the decision turns on a small number of properties.

When a table is enough Left: a job table shares the application transaction, so a job can be enqueued atomically with the work that caused it, needs no extra infrastructure, and is queryable with SQL. Right: a broker handles very high throughput, fan-out and long retention better, at the cost of another system to run and no transactional link to the database. a job table enqueue in the same transaction no extra infrastructure inspectable with SQL good to thousands per second a dedicated broker higher throughput, fan-out mature retry and DLQ tooling another system to operate no transaction with your data The transactional enqueue is the strongest argument for a table, and it is often decisive.

The transactional enqueue is the strongest argument for a table. Inserting a job in the same transaction as the work that caused it means the job exists if and only if the work committed — no lost jobs after a rollback, no jobs for orders that were never created. With a broker, the enqueue is a separate system, which is why the outbox pattern exists: write the intent transactionally, deliver it afterwards. If you would need an outbox anyway, the outbox is a job table.

Operational simplicity is the second. No extra service to deploy, monitor, upgrade or secure; backups already include the queue; and the queue is inspectable with SQL — "how many jobs are pending, and how old is the oldest" is a query rather than a dashboard integration.

Throughput is where brokers win. A PostgreSQL queue comfortably handles hundreds to a few thousand jobs per second with the structure above, and the ceiling comes from write amplification: every claim and every outcome is a row update, which produces dead tuples that autovacuum must clean. Past that, a purpose-built broker is doing less work per message.

Fan-out and retention also favour brokers. Several independent consumers of the same stream, long retention, replay from a point in time — these are things a broker does natively and a table does awkwardly.

Two operational notes for the table version. Vacuum matters: a queue table is updated constantly, so it accumulates dead rows quickly, and an aggressive autovacuum setting for that table specifically is worth configuring. And completed jobs should not accumulate forever — archive or delete them on a schedule, in batches, using the approach in processing large tables in batches with partitions, so the table stays small even though the history is long.

If a library is preferable to hand-rolling, several build on exactly this pattern — a job table, SKIP LOCKED claims, retries and a reaper — and adopting one is a reasonable way to get the details right. The value in understanding the mechanism is that the failure modes are then legible: a stuck job is a row you can see, a duplicate is an at-least-once delivery you should have made idempotent, and a slow claim is an index you can check.

Frequently Asked Questions

What does SKIP LOCKED do?

It makes SELECT ... FOR UPDATE skip rows another transaction has locked instead of waiting for them. With ORDER BY and LIMIT, each worker claims the first rows nobody else holds, which is what makes a table behave as a queue.

Why commit the claim before doing the work?

So no lock or transaction is held while the work runs, and so a crash leaves visible evidence — a row marked running — that a reaper can requeue. Processing inside the claim transaction holds locks and a connection for the whole job.

Is a database queue exactly-once?

No. It is at-least-once: a worker can crash after doing the work and before recording it, so the job is requeued. Handlers must be idempotent, usually via an idempotency key.

When should I use a broker instead?

When throughput exceeds a few thousand jobs per second, when several independent consumers need the same stream, or when long retention and replay matter. Below that, the transactional enqueue usually outweighs the benefits.