Review: Postgres Job Queue Worker Written by an AI Assistant

Published 10 Nov, 2026

Your team asked an AI coding assistant to replace a cron script with a background job queue stored in PostgreSQL. It opened this pull request. You are the reviewer, and nothing merges without your approval.

The assistant's description:

Adds a jobs table and a worker loop. Jobs are claimed oldest first, retried with exponential backoff up to 5 attempts, and enqueued inside the caller's transaction so a job only exists if the work that created it committed. Handlers live in HANDLERS, keyed by job kind.

Where it runs: 12 worker processes on 3 hosts, using psycopg 3 with its default type adapters, against PostgreSQL 15. The queue handles about 50,000 jobs a day (sending emails and charging subscriptions), and finished jobs are kept for 90 days. Workers are restarted on every deploy, several times a day. The web app's database sessions use the Europe/London time zone; the workers' sessions use UTC.

Your task

Flag every problem you would block the merge for. Click a line number to comment, and Shift-click to cover a range.
For each one, choose the issue type and severity, and explain what fails and how to fix it.
You are scored on the problems you catch, the right issue type and a clear explanation. Missing a critical problem fails the round.

Aim for 20 minutes, as in a real review round.

Two workers run claim_job at the same instant. Walk through both transactions.

What Python type does psycopg 3 return for a jsonb column?

MAX_ATTEMPTS is defined. Where is it used?

A deploy kills a worker halfway through a job. What is that row's status a day later?

Look at the column types and the index in the migration, with 90 days of finished jobs in the table.

A strong reviewer blocks this change on two grounds: as written no job ever succeeds, and once that is fixed, concurrent workers will run the same job twice, which means duplicate emails and duplicate subscription charges.

Two workers can claim the same job (lines 29–33 of worker.py, critical). The SELECT takes no lock, so two workers read the same oldest row, and both UPDATEs succeed because the update does not check the status. With 12 workers polling, duplicates are routine. Claim in one statement: UPDATE jobs SET status = 'running' ... WHERE id = (SELECT id ... FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING ....
Payload decoded twice (line 78, critical). psycopg 3 already turns jsonb into a dict, and json.loads(dict) raises TypeError. Every job fails, and is retried forever. Pass payload straight to the handler.
Retries never stop (lines 57–64, major). MAX_ATTEMPTS is never checked, so a poison job is retried for ever with growing delays and never reaches a dead state anyone can see. When attempts + 1 >= MAX_ATTEMPTS, set status = 'failed' and alert.
Jobs stuck in running (line 39, major). If a worker dies after claiming, which happens on every deploy, the row stays running for ever. Store a lease such as locked_until, reclaim expired leases, and handle SIGTERM so a worker finishes its current job before exiting.
Bookkeeping errors re-run work (lines 79–82, minor). finish is inside the try, so if the update fails after the handler succeeded, the job is requeued and runs again. If fail itself raises, the whole worker crashes. Separate handler errors from database errors and reconnect on connection loss.
Wrong index (line 14 of the migration, minor). The polling query filters by status and run_at, not kind, so 12 workers each scan millions of finished rows every second. Use a partial index: CREATE INDEX ON jobs (run_at) WHERE status = 'queued'.
Time zone mix-up (line 8 of the migration, minor). run_at is timestamp without time zone, so now() is stored as local time of the writing session. Jobs enqueued from the web app (London time) and compared by workers (UTC) run an hour late in summer. Use timestamptz.

import json
import logging
import time
import traceback

import psycopg

from .handlers import HANDLERS

log = logging.getLogger("worker")

POLL_INTERVAL = 1.0  # seconds to wait when the queue is empty
MAX_ATTEMPTS = 5

def enqueue(conn, kind, payload, delay=0):
    """Add a job. Runs in the caller's transaction, so it commits with their work."""
    with conn.cursor() as cur:
        cur.execute(
            "INSERT INTO jobs (kind, payload, run_at) "
            "VALUES (%s, %s, now() + %s * interval '1 second')",
            (kind, json.dumps(payload), delay),
        )

def claim_job(conn):
    """Take the oldest job that is due and mark it as running."""
    with conn.cursor() as cur:
        cur.execute(
            "SELECT id, kind, payload, attempts FROM jobs "
            "WHERE status = 'queued' AND run_at <= now() "
            "ORDER BY run_at LIMIT 1"
        )
        row = cur.fetchone()
        if row is None:
            conn.commit()
            return None
        cur.execute(
            "UPDATE jobs SET status = 'running', started_at = now() WHERE id = %s",
            (row[0],),
        )
    conn.commit()
    return row

def finish(conn, job_id):
    with conn.cursor() as cur:
        cur.execute(
            "UPDATE jobs SET status = 'done', finished_at = now() WHERE id = %s",
            (job_id,),
        )
    conn.commit()

def fail(conn, job_id, attempts, error):
    """Put the job back on the queue with exponential backoff."""
    delay = 2 ** attempts
    with conn.cursor() as cur:
        cur.execute(
            "UPDATE jobs SET status = 'queued', attempts = attempts + 1, "
            "last_error = %s, run_at = now() + %s * interval '1 second' "
            "WHERE id = %s",
            (error, delay, job_id),
        )
    conn.commit()

def run_forever(dsn):
    with psycopg.connect(dsn) as conn:
        log.info("Worker started")
        while True:
            job = claim_job(conn)
            if job is None:
                time.sleep(POLL_INTERVAL)
                continue
            job_id, kind, payload, attempts = job
            try:
                HANDLERS[kind](json.loads(payload))
                finish(conn, job_id)
            except Exception:
                log.exception("Job %s (%s) failed", job_id, kind)
                fail(conn, job_id, attempts, traceback.format_exc())

{"issues":[
{"id": "claim", "file": "worker.py", "lines": [29, 33], "category": "concurrency", "severity": "critical", "title": "Job claim is not atomic", "explanation": "The SELECT takes no lock and the UPDATE does not check the status, so two workers can claim and run the same job, sending emails or charging twice.", "keywords": ["skip locked", "for update", "race", "lock", "atomic", "twice", "duplicate", "same job"], "accept": ["bug"], "fix": "Claim in one statement: UPDATE ... WHERE id = (SELECT id ... FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING ..."},
{"id": "jsonb", "file": "worker.py", "lines": [78, 78], "category": "bug", "severity": "critical", "title": "jsonb payload decoded twice", "explanation": "psycopg 3 already returns jsonb as a dict, and json.loads(dict) raises TypeError, so every job fails and is retried.", "keywords": ["jsonb", "dict", "typeerror", "already", "json.loads", "decode", "adapter", "twice"], "accept": ["api"], "fix": "Pass payload to the handler as it is."},
{"id": "attempts", "file": "worker.py", "lines": [57, 64], "category": "bug", "severity": "major", "title": "MAX_ATTEMPTS is never enforced", "explanation": "Failed jobs are always put back as queued, so a poison job retries for ever and never reaches a visible dead state.", "keywords": ["max_attempts", "poison", "dead", "forever", "never", "limit", "give up", "failed"], "accept": ["design"], "fix": "When attempts + 1 >= MAX_ATTEMPTS set status to failed (dead letter) and alert."},
{"id": "stuck", "file": "worker.py", "lines": [39, 39], "category": "design", "severity": "major", "title": "Crashed workers leave jobs running for ever", "explanation": "A worker killed after claiming (e.g. on every deploy) leaves the row in running, and nothing ever retries it.", "keywords": ["lease", "heartbeat", "stuck", "crash", "timeout", "visibility", "sigterm", "reclaim", "restart"], "accept": ["concurrency", "bug"], "fix": "Add a lease column such as locked_until, reclaim expired leases and shut down gracefully on SIGTERM."},
{"id": "finish", "file": "worker.py", "lines": [79, 82], "category": "bug", "severity": "minor", "title": "Database errors in finish and fail are mishandled", "explanation": "If finish fails after the handler succeeded the job is requeued and runs again; if fail raises, the worker loop crashes.", "keywords": ["finish", "outside", "reconnect", "crash", "run again", "re-run", "separate", "connection"], "accept": ["design"], "fix": "Keep finish outside the handler try block, and catch database errors separately with reconnection."},
{"id": "index", "file": "migrations/0031_jobs.sql", "lines": [14, 14], "category": "perf", "severity": "minor", "title": "Index on the wrong column", "explanation": "The poll filters on status and run_at, so the index on kind is useless and every poll scans the finished jobs.", "keywords": ["index", "partial", "status", "run_at", "scan", "seq scan"], "fix": "CREATE INDEX ON jobs (run_at) WHERE status = 'queued'."},
{"id": "tz", "file": "migrations/0031_jobs.sql", "lines": [8, 8], "category": "edge", "severity": "minor", "title": "run_at uses timestamp without time zone", "explanation": "now() is stored in the writing session's local time, so jobs enqueued from London-time sessions are compared against UTC workers and run an hour late in summer.", "keywords": ["timestamptz", "time zone", "timezone", "utc", "london", "local", "dst"], "accept": ["bug"], "fix": "Make run_at timestamptz."}
]}

Topics #020 AI Code Review Concurrency Hard PostgreSQL Python