Leasing Review Items to Concurrent Workers with FOR UPDATE SKIP LOCKED

When the Manual Review Queue scales past a single reviewer process, the dequeue step becomes a distributed-locking problem: several workers poll the same table at once, and without a locking discipline two of them can pull the same exception into ASSIGNED state at the same moment, producing a duplicate sign-off. Postgres solves this natively. SELECT ... FOR UPDATE SKIP LOCKED lets each worker’s transaction lock a batch of candidate rows and silently skip any row another transaction already holds, rather than blocking on it. This page implements that lease pattern end to end — schema, atomic lease query, worker loop, abandoned-lease reclaim, and idempotent completion — as a direct dequeue mechanism underneath the queue described in the parent page. It pairs with the asyncio-based Async Approval Queue Design, specifically its worked example Building an Asyncio Approval Queue with Backpressure, which covers the in-process producer/consumer side; this page covers the durable-store side that backs it.

Prerequisites

Step 1 — Table and status index

The queue table tracks lease ownership directly on the row: leased_by, leased_at, and lease_expires_at. A partial index on the pending set keeps the dequeue scan cheap regardless of how large the resolved history grows, since resolved rows never need to be scanned again.

sql
CREATE TABLE review_queue_item (
    id                 UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    idempotency_key    TEXT NOT NULL UNIQUE,
    ledger_entity      TEXT NOT NULL,
    variance_amount    NUMERIC(18, 2) NOT NULL,
    match_decision     TEXT NOT NULL,          -- MATCHED | PARTIAL | UNMATCHED
    state              TEXT NOT NULL DEFAULT 'PENDING',
    leased_by          TEXT,
    leased_at          TIMESTAMPTZ,
    lease_expires_at   TIMESTAMPTZ,
    trace_id           UUID NOT NULL DEFAULT gen_random_uuid(),
    source_hash        TEXT NOT NULL,
    created_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT valid_state CHECK (
        state IN ('PENDING', 'ASSIGNED', 'RESOLVED')
    )
);

-- Only PENDING and ASSIGNED rows are ever scanned by a worker; RESOLVED
-- rows fall out of this index automatically, keeping the dequeue O(1)-ish
-- against queue depth rather than total row count.
CREATE INDEX idx_review_queue_dequeue
    ON review_queue_item (created_at)
    WHERE state IN ('PENDING', 'ASSIGNED');

Step 2 — The atomic lease query

The lease is a single statement: a SELECT ... FOR UPDATE SKIP LOCKED scoped to a common table expression, joined into an UPDATE that stamps ownership and returns the leased rows. Because it is one statement inside one transaction, there is no window between “find a row” and “claim a row” for a second worker to race into — SKIP LOCKED guarantees a row another transaction is currently locking is excluded from this worker’s result set entirely, rather than causing it to wait and then fail.

sql
WITH candidate AS (
    SELECT id
    FROM review_queue_item
    WHERE state = 'PENDING'
    ORDER BY created_at
    LIMIT %(batch_size)s
    FOR UPDATE SKIP LOCKED
)
UPDATE review_queue_item AS rq
SET state             = 'ASSIGNED',
    leased_by         = %(worker_id)s,
    leased_at          = now(),
    lease_expires_at  = now() + make_interval(secs => %(lock_timeout_ms)s / 1000.0)
FROM candidate
WHERE rq.id = candidate.id
RETURNING rq.id, rq.idempotency_key, rq.ledger_entity, rq.variance_amount,
          rq.match_decision, rq.trace_id, rq.source_hash;

ORDER BY created_at inside the CTE keeps the lease roughly FIFO within a batch, while SKIP LOCKED breaks strict ordering across concurrent batches — an acceptable trade for a review queue, where fairness matters more than perfect arrival order. Setting lock_timeout_ms short (a few hundred milliseconds) at the session level, separately from the lease duration, prevents a worker from blocking on an unrelated exclusive lock elsewhere in the same table.

Step 3 — Python worker loop

The worker leases a batch, processes each item off the connection (so the transaction that holds the lease commits immediately rather than staying open for the duration of review), and only opens a new transaction to record completion.

python
import logging
import time
from decimal import Decimal
from datetime import datetime, timezone
from uuid import UUID

import psycopg
from psycopg.rows import dict_row

logger = logging.getLogger("reconciliation.review_queue.lease")

LEASE_SQL = """
WITH candidate AS (
    SELECT id FROM review_queue_item
    WHERE state = 'PENDING'
    ORDER BY created_at
    LIMIT %(batch_size)s
    FOR UPDATE SKIP LOCKED
)
UPDATE review_queue_item AS rq
SET state = 'ASSIGNED', leased_by = %(worker_id)s, leased_at = now(),
    lease_expires_at = now() + make_interval(secs => %(lock_timeout_ms)s / 1000.0)
FROM candidate
WHERE rq.id = candidate.id
RETURNING rq.id, rq.idempotency_key, rq.ledger_entity, rq.variance_amount,
          rq.match_decision, rq.trace_id, rq.source_hash;
"""


def lease_batch(
    conn: psycopg.Connection,
    worker_id: str,
    batch_size: int = 20,
    lock_timeout_ms: int = 30_000,
) -> list[dict]:
    with conn.cursor(row_factory=dict_row) as cur:
        cur.execute(
            LEASE_SQL,
            {
                "worker_id": worker_id,
                "batch_size": batch_size,
                "lock_timeout_ms": lock_timeout_ms,
            },
        )
        rows = cur.fetchall()
    conn.commit()  # release row locks immediately; the lease itself is the guard now
    for row in rows:
        row["variance_amount"] = Decimal(row["variance_amount"])
    return rows


def worker_loop(
    dsn: str,
    worker_id: str,
    poll_interval_ms: int = 500,
    batch_size: int = 20,
) -> None:
    with psycopg.connect(dsn) as conn:
        while True:
            batch = lease_batch(conn, worker_id, batch_size=batch_size)
            if not batch:
                time.sleep(poll_interval_ms / 1000.0)
                continue
            for item in batch:
                logger.info(
                    "review_item_leased",
                    extra={
                        "trace_id": str(item["trace_id"]),
                        "source_hash": item["source_hash"],
                        "match_decision": item["match_decision"],
                        "worker_id": worker_id,
                        "item_id": str(item["id"]),
                    },
                )
                # ... reviewer UI or automated adjudication happens here ...

Step 4 — Reclaiming abandoned leases

A worker that crashes, is rescheduled, or is killed mid-review leaves its rows stuck in ASSIGNED forever unless something reclaims them. A separate periodic job — never the worker that owns the lease — sweeps rows whose lease_expires_at has passed and returns them to PENDING, using the same SKIP LOCKED discipline so the reclaim job cannot collide with a worker that is mid-lease-renewal on the same row.

python
RECLAIM_SQL = """
WITH expired AS (
    SELECT id FROM review_queue_item
    WHERE state = 'ASSIGNED' AND lease_expires_at < now()
    ORDER BY lease_expires_at
    LIMIT %(reclaim_batch)s
    FOR UPDATE SKIP LOCKED
)
UPDATE review_queue_item AS rq
SET state = 'PENDING', leased_by = NULL, leased_at = NULL, lease_expires_at = NULL
FROM expired
WHERE rq.id = expired.id
RETURNING rq.id, rq.trace_id, rq.source_hash, rq.match_decision;
"""


def reclaim_abandoned(conn: psycopg.Connection, reclaim_batch: int = 100) -> int:
    with conn.cursor(row_factory=dict_row) as cur:
        cur.execute(RECLAIM_SQL, {"reclaim_batch": reclaim_batch})
        reclaimed = cur.fetchall()
    conn.commit()
    for row in reclaimed:
        logger.warning(
            "review_item_lease_reclaimed",
            extra={
                "trace_id": str(row["trace_id"]),
                "source_hash": row["source_hash"],
                "match_decision": row["match_decision"],
                "item_id": str(row["id"]),
            },
        )
    return len(reclaimed)

Run reclaim_abandoned on its own scheduled cadence (reclaim_after_s below governs how stale a lease must be before it is eligible), independent of the worker pool’s own polling loop, so a total worker-fleet outage cannot leave the reclaimer starved along with everything else.

Step 5 — Idempotent completion with an audit trail

Completion must be safe to retry: a worker that dies after committing the ledger effect but before acknowledging the queue must not double-post on restart. The guard is a conditional UPDATE keyed on both id and the current state, so a retry against an already-RESOLVED row is a no-op that still returns success to the caller.

python
COMPLETE_SQL = """
UPDATE review_queue_item
SET state = 'RESOLVED'
WHERE id = %(item_id)s AND state = 'ASSIGNED' AND leased_by = %(worker_id)s
RETURNING id, idempotency_key, trace_id, source_hash, match_decision;
"""


def complete_item(
    conn: psycopg.Connection,
    item_id: UUID,
    worker_id: str,
) -> bool:
    with conn.cursor(row_factory=dict_row) as cur:
        cur.execute(COMPLETE_SQL, {"item_id": str(item_id), "worker_id": worker_id})
        row = cur.fetchone()
    conn.commit()
    if row is None:
        # Already resolved by this worker on a prior attempt, or leased
        # elsewhere after a reclaim — either way, not a fresh completion.
        logger.info("review_item_completion_noop", extra={"item_id": str(item_id)})
        return False
    logger.info(
        "review_item_resolved",
        extra={
            "trace_id": str(row["trace_id"]),
            "source_hash": row["source_hash"],
            "match_decision": row["match_decision"],
            "idempotency_key": row["idempotency_key"],
            "item_id": str(row["id"]),
            "worker_id": worker_id,
            "resolved_at": datetime.now(timezone.utc).isoformat(),
        },
    )
    return True

Binding leased_by = %(worker_id)s into the WHERE clause is what makes this safe under reclaim: if a lease expired and a different worker picked the row up, the original worker’s late completion attempt matches zero rows instead of silently resolving an item it no longer owns.

Configuration boundary table

Parameter Default Valid range Notes
batch_size 20 1–200 Rows leased per dequeue call; larger batches reduce round trips but increase blast radius of a worker crash.
lock_timeout_ms 30000 1000–120000 Lease duration before a row becomes eligible for reclaim.
reclaim_after_s 45 10–600 Grace period past lease_expires_at before the reclaim job sweeps a row; keep above expected clock skew.
max_inflight 1000 100–10000 Ceiling on rows in ASSIGNED across the fleet; a reclaim-job alert fires if exceeded.
poll_interval_ms 500 50–5000 Worker idle-poll cadence when a lease batch returns empty; add jitter to avoid synchronised polling.

Verification and testing

The property to prove is simple to state and easy to get wrong under load: two workers polling concurrently must never receive an overlapping set of leased IDs. A fixture that opens two real connections and leases against the same seeded rows catches regressions that a single-connection unit test cannot.

python
import psycopg
import pytest


@pytest.fixture
def seeded_queue(pg_dsn):
    with psycopg.connect(pg_dsn) as conn:
        with conn.cursor() as cur:
            for i in range(40):
                cur.execute(
                    """
                    INSERT INTO review_queue_item
                        (idempotency_key, ledger_entity, variance_amount,
                         match_decision, source_hash)
                    VALUES (%s, 'ACME-CORP', 12.50, 'UNMATCHED', %s)
                    """,
                    (f"idem-{i}", f"hash-{i}"),
                )
        conn.commit()
    return pg_dsn


def test_two_workers_never_double_lease(seeded_queue):
    with psycopg.connect(seeded_queue) as conn_a, psycopg.connect(seeded_queue) as conn_b:
        batch_a = lease_batch(conn_a, "worker-a", batch_size=20)
        batch_b = lease_batch(conn_b, "worker-b", batch_size=20)

    ids_a = {row["id"] for row in batch_a}
    ids_b = {row["id"] for row in batch_b}

    assert ids_a.isdisjoint(ids_b)          # DOUBLE_LEASE would show up here
    assert len(ids_a) + len(ids_b) == 40    # every row leased exactly once

Run this fixture under a load-test harness that spawns 10–50 concurrent worker connections against a few hundred seeded rows, and assert the disjoint-set property holds across every pair. Also assert that reclaim_abandoned run against a queue with artificially backdated lease_expires_at values returns every stale row and that a subsequent lease_batch call can immediately re-lease them.

Troubleshooting

  • DOUBLE_LEASE — two workers report the same item_id as assigned. Root cause: the dequeue was split into a SELECT followed by a separate UPDATE in two statements, reopening the race window SKIP LOCKED is meant to close. Fix: dequeue and claim in one statement, as in Step 2 — never a bare SELECT ... FOR UPDATE SKIP LOCKED followed by an application-side UPDATE.
  • LEASE_LEAK — ASSIGNED row count grows without bound. Root cause: the reclaim job is not running, is crashing silently, or reclaim_after_s is set larger than the deploy cadence, so restarts outpace reclamation. Fix: run the reclaim job on its own supervised schedule independent of the worker pool, and alert on ASSIGNED count approaching max_inflight.
  • LOCK_CONTENTION — lease queries intermittently time out under lock_timeout_ms. Root cause: a long-running transaction elsewhere (a batch export, a schema migration) holds a conflicting lock on the table, and SKIP LOCKED only skips row-level locks held by other SELECT ... FOR UPDATE claimants, not table-level locks. Fix: set an explicit statement-level lock_timeout for the lease query, and audit for any ALTER TABLE or unindexed bulk UPDATE against review_queue_item during peak hours.
  • STARVATION — a subset of old rows never gets leased even though queue depth is low. Root cause: ORDER BY created_at combined with SKIP LOCKED means a row currently held by a slow worker is silently skipped every cycle, and if that worker never completes or times out, the row is passed over repeatedly. Fix: confirm lease_expires_at is set on every lease and that the reclaim job’s reclaim_after_s is short enough that a stuck row re-enters PENDING within one or two poll cycles.
  • PRECISION_FAULT on completion — variance totals drift by fractions of a cent. Root cause: variance_amount read back from the cursor as a Python float instead of Decimal. Fix: wrap every numeric column read in Decimal(str(value)) at the row-mapping boundary, as in lease_batch above.

Part of Manual Review Queue Design within Exception Routing & Human-in-the-Loop Workflows.