Postgres job queue processes the same job twice: fixing it with FOR UPDATE SKIP LOCKED

Support forwarded a handful of tickets on a Monday: customers of our B2B invoicing product had received the same invoice email twice, a few seconds apart. Over the weekend we had scaled the email worker deployment from 2 to 8 replicas to drain a backlog faster. The duplicates started right after that change, and the rate grew with the replica count. Our downstream table that records sent emails finally made it loud:

ERROR:  duplicate key value violates unique constraint "invoice_emails_job_id_key"
DETAIL:  Key (job_id)=(4815162) already exists.

That unique constraint only existed on the audit table, not on the email provider call itself, so by the time the insert failed the second email was already gone.

Reproducing it

The queue was a plain table, jobs, with a status column. Each worker ran this every second:

UPDATE jobs
SET status = 'running', started_at = now(), worker = $1
WHERE id IN (
    SELECT id FROM jobs
    WHERE status = 'pending'
    ORDER BY id
    LIMIT 10
)
RETURNING id, payload;

With two workers it almost never misbehaved. With eight workers polling at the same moment, a load test with 20,000 pending jobs produced around 40 double claims per run. We could reproduce it on a laptop by starting two psql sessions, opening a transaction in each, and running the statement in both before committing the first.

What we ruled out

• Retries in the email client. Provider logs showed two distinct API requests, each from a different pod name, not one request retried.
• Duplicate enqueue. The jobs table had exactly one row per invoice; the duplicates shared the same jobs.id.
• Replica lag or a stale read pool. All workers wrote to the primary; there was no read routing in this service.

Root cause

Under the default READ COMMITTED isolation level, the subquery picks candidate ids from a snapshot taken when the statement starts. Two workers can both see job 4815162 as pending. The first one locks and updates the row. The second blocks on the row lock, and when the first commits, Postgres re-checks the updated row against the outer WHERE clause only. That clause is id IN (...) with the already computed list, and the id still matches, so the second worker updates the row again and receives it in RETURNING. The status = 'pending' condition lived in the subquery, which is not re-evaluated. Both workers believed they owned the job.

What is happening under the hood

When an UPDATE in READ COMMITTED finds that its target row was changed by a concurrent transaction that has since committed, Postgres follows the update chain to the newest row version and re-evaluates the statement's own qualification against that version. The docs describe this in the "Read Committed Isolation Level" section of the concurrency chapter. The important detail is that a subquery or joined relation is not rescanned to see whether it would still pick the row. If the condition that makes a row claimable is not checked on the target row itself, a concurrent claim slips through.

A plain SELECT ... FOR UPDATE in the subquery closes the race, because the second worker waits for the lock and then re-checks status = 'pending' on the locked row. But it creates a different problem: every worker queues up behind the same first ten rows, so throughput collapses to roughly one worker at a time. SKIP LOCKED tells Postgres to skip rows another transaction has locked instead of waiting, so each worker grabs a disjoint batch.

Environment

Minimal reproduction

CREATE TABLE jobs (
    id         bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    status     text NOT NULL DEFAULT 'pending',
    payload    jsonb NOT NULL,
    worker     text,
    started_at timestamptz
);
INSERT INTO jobs (payload)
SELECT jsonb_build_object('invoice', g) FROM generate_series(1, 20) g;

Session A:

BEGIN;
UPDATE jobs SET status = 'running', worker = 'A'
WHERE id IN (SELECT id FROM jobs WHERE status = 'pending' ORDER BY id LIMIT 3)
RETURNING id;
-- returns 1, 2, 3; do not commit yet

Session B, while A is open:

BEGIN;
UPDATE jobs SET status = 'running', worker = 'B'
WHERE id IN (SELECT id FROM jobs WHERE status = 'pending' ORDER BY id LIMIT 3)
RETURNING id;
-- blocks

Commit A. Session B unblocks and returns 1, 2, 3 as well. Both sessions claimed the same three jobs.

Who is affected

Any hand-rolled queue on Postgres (or another MVCC database at a similar isolation level) where the "is this job still free" condition is evaluated in a subquery, a CTE without row locks, or a separate SELECT before an UPDATE. It also hits the two-step pattern "SELECT pending rows, then UPDATE by id" even in one transaction. Libraries that already use SKIP LOCKED internally are not affected.

The fix

We moved the claim into one statement that locks candidate rows, skips rows locked by others, and re-checks status on the target row. We also added a lease so a crashed worker does not strand jobs forever.

ALTER TABLE jobs ADD COLUMN locked_until timestamptz;
ALTER TABLE jobs ADD COLUMN attempts int NOT NULL DEFAULT 0;
CREATE INDEX jobs_claimable_idx ON jobs (id) WHERE status = 'pending';

- UPDATE jobs
- SET status = 'running', started_at = now(), worker = $1
- WHERE id IN (
-     SELECT id FROM jobs
-     WHERE status = 'pending'
-     ORDER BY id
-     LIMIT 10
- )
- RETURNING id, payload;
+ WITH picked AS (
+     SELECT id FROM jobs
+     WHERE status = 'pending'
+     ORDER BY id
+     LIMIT 10
+     FOR UPDATE SKIP LOCKED
+ )
+ UPDATE jobs j
+ SET status = 'running',
+     started_at = now(),
+     locked_until = now() + interval '5 minutes',
+     attempts = j.attempts + 1,
+     worker = $1
+ FROM picked
+ WHERE j.id = picked.id
+ RETURNING j.id, j.payload;

The worker commits the claim immediately, does the work outside any open transaction, and then marks the job done:

import psycopg

CLAIM_SQL = open("claim.sql").read()

def run_once(conn: psycopg.Connection, worker_id: str) -> int:
    with conn.transaction():
        rows = conn.execute(CLAIM_SQL, (worker_id,)).fetchall()
    for job_id, payload in rows:
        send_invoice_email(payload, idempotency_key=f"invoice-job-{job_id}")
        with conn.transaction():
            conn.execute(
                "UPDATE jobs SET status = 'done', locked_until = NULL "
                "WHERE id = %s AND worker = %s",
                (job_id, worker_id),
            )
    return len(rows)

A small sweeper returns expired leases to the pool:

UPDATE jobs
SET status = 'pending', worker = NULL, locked_until = NULL
WHERE status = 'running' AND locked_until < now() AND attempts < 5;

Finally, we passed an idempotency key to the email provider so that a lease expiry during a slow send cannot produce a duplicate either. A queue gives at-least-once delivery; the side effect has to tolerate a repeat.

Alternatives and trade-offs

• Keep the lock for the whole job. Claim with FOR UPDATE SKIP LOCKED, process inside the same transaction, mark done, commit. No lease needed because the lock dies with the session. The cost is a long open transaction per job, which holds back vacuum and pins a connection. Fine for jobs under a second, bad for email sends that sometimes take 30 seconds.
• SERIALIZABLE isolation. The original query would raise could not serialize access due to concurrent update instead of double claiming. Correct, but every collision becomes a retry, and with eight hot workers we saw retry rates above 30 percent.
• Advisory locks. pg_try_advisory_xact_lock(id) works, but it is easy to get wrong with LIMIT because the planner may evaluate the function on more rows than it returns.
• A ready-made library. If you would rather not own this code, mature queue libraries for most languages already implement the same pattern.

How to test the fix

We wrote a concurrency test that starts 16 threads against a seeded table and asserts every job id is claimed exactly once:

import threading, psycopg

def test_no_double_claims(dsn):
    claimed, lock = [], threading.Lock()

    def worker(n):
        with psycopg.connect(dsn) as conn:
            while True:
                with conn.transaction():
                    rows = conn.execute(CLAIM_SQL, (f"w{n}",)).fetchall()
                if not rows:
                    return
                with lock:
                    claimed.extend(r[0] for r in rows)

    threads = [threading.Thread(target=worker, args=(i,)) for i in range(16)]
    [t.start() for t in threads]
    [t.join() for t in threads]
    assert len(claimed) == len(set(claimed)) == 20000

docker run -d --name pgtest -e POSTGRES_PASSWORD=test -p 55432:5432 postgres:16.4
pytest tests/test_queue_claim.py -q

Against the old query the assertion failed on every run; against the new one it passed 200 runs in a row.

Confirming in production

After deploying, we checked that no job was processed twice by looking at the email audit table:

SELECT job_id, count(*)
FROM invoice_emails
WHERE created_at > now() - interval '7 days'
GROUP BY job_id
HAVING count(*) > 1;

Expected output is zero rows, and that is what we got for the following two weeks. We also watched pg_stat_activity while the workers ran at full speed:

SELECT wait_event_type, wait_event, count(*)
FROM pg_stat_activity
WHERE application_name = 'invoice-worker'
GROUP BY 1, 2;

With SKIP LOCKED there were no Lock / transactionid waits. During an earlier attempt with plain FOR UPDATE that same query showed seven of eight workers waiting on transactionid, which is the throughput collapse described above. Backlog drain time went from about 50 minutes to 14 minutes for 180,000 jobs.

Regression guard

The concurrency test runs in CI against a real Postgres service container, not a mock. We also added a unique index on invoice_emails(job_id) that was already there, plus an alert on the count of attempts >= 3 so a job that keeps expiring its lease gets a human look.

Security implications

The bug itself leaked nothing, but duplicate sends of financial documents confuse customers and are a plausible phishing cover ("you were charged twice"). The fix does not change the attack surface. Two notes: the worker role needs UPDATE on jobs only, not on the business tables, and the job payload should hold ids, not full invoice data, so a compromised worker host cannot read customer details from the queue.

Prevention checklist

• Put every "is this row still claimable" condition on the row being updated, never only in a subquery.
• Use FOR UPDATE SKIP LOCKED for multi-consumer claims; plain FOR UPDATE serialises workers.
• Add a lease column and a sweeper so crashed workers do not strand jobs.
• Make side effects idempotent; queues are at-least-once.
• Test claiming with real concurrent connections in CI.

References

• PostgreSQL docs, SELECT, The Locking Clause: https://www.postgresql.org/docs/16/sql-select.html
• PostgreSQL docs, Transaction Isolation: https://www.postgresql.org/docs/16/transaction-iso.html
• psycopg 3 documentation, Transactions management: https://www.psycopg.org/psycopg3/docs/basic/transactions.html