Almost every system we have built with a human in the loop needs the same component: a queue that hands items to reviewers. KYC and KYB verification, transaction moderation, issuance tiers on a lending marketplace, and lately the reject path of an LLM content pipeline all converge on it. The naive implementation is a table ordered by created_at with an assigned_to column, and it fails in the same three ways every time: two reviewers open the same item, an item vanishes from the queue when somebody closes a browser tab, and a decision gets applied twice to the subject it was about.
The constraint that makes this harder than a worker queue is that the consumer is a person. A Celery worker holds a job for milliseconds and you can wrap the whole thing in a transaction. A compliance officer holds an item for eight minutes, walks away for coffee, and submits at minute twenty from a tab that has been open since before the shift change. You cannot keep a database transaction open across that, so the exclusivity has to live in data rather than in a lock. And the queue is never the source of truth for the outcome: approving a case means something happens elsewhere — an account is unblocked, a payout is released, a document is re-issued — and that something can fail independently of the click that requested it.
Claiming: SKIP LOCKED for the instant, a lease for the minutes
We use two different mechanisms for two different time scales. A row lock resolves the race between two reviewers who press "next item" in the same millisecond. A lease resolves the much longer window during which the item is on somebody's screen.
UPDATE review_items
SET lease_owner = %(reviewer_id)s,
lease_token = gen_random_uuid(),
lease_expires = now() + %(lease_ttl)s::interval,
lease_count = lease_count + 1
WHERE id = (
SELECT id
FROM review_items
WHERE queue = %(queue)s
AND state = 'pending'
AND (lease_expires IS NULL OR lease_expires < now())
ORDER BY priority DESC, created_at
FOR UPDATE SKIP LOCKED
LIMIT 1
)
RETURNING id, lease_token, payload;
The inner select takes a real row lock for the microseconds the statement runs, and SKIP LOCKED makes concurrent claimers walk past a contended row instead of queueing behind it. The outer update writes a lease that outlives the transaction. Expiry is evaluated at claim time, which means there is no sweeper job to run and nothing to page on at 3am: an abandoned item becomes claimable because the next claim query stops excluding it.
The index that makes this cheap is partial, and it matters more than the query:
CREATE INDEX review_items_claimable
ON review_items (queue, priority DESC, created_at)
WHERE state = 'pending';
Decided items stay in the table for audit but leave the index, so the structure a claim walks stays roughly the size of the backlog rather than the size of the history.
The lease token is a fence, not a receipt
Handing the token to the client is what turns the lease into a correctness mechanism. Every decision submission carries it, and the write only succeeds if the token still matches. A tab that has been open past expiry — and whose item may already have been reviewed by somebody else — gets a clean rejection instead of silently overwriting a colleague's verdict.
class StaleLease(Exception):
pass
def submit_decision(conn, item_id, lease_token, reviewer_id, verdict, reason_code, evidence_hash):
with conn.transaction():
closed = conn.execute(
"""
UPDATE review_items
SET state = 'decided', decided_at = now(), lease_expires = NULL
WHERE id = %s AND lease_token = %s AND state = 'pending'
RETURNING id
""",
(item_id, lease_token),
).fetchone()
if closed is None:
raise StaleLease(item_id)
decision_id = conn.execute(
"""
INSERT INTO review_decisions
(item_id, reviewer_id, verdict, reason_code, evidence_hash)
VALUES (%s, %s, %s, %s, %s)
RETURNING id
""",
(item_id, reviewer_id, verdict, reason_code, evidence_hash),
).fetchone()[0]
return decision_id
The evidence_hash is the second thing worth insisting on. A reviewer decides on the basis of a specific set of documents, a specific provider response, a specific rendered payload. Those can change afterwards — a user re-uploads a passport photo, a provider re-scores a session. Hashing what was actually shown and storing it with the verdict is the difference between an audit trail that answers "why was this approved" and one that only answers "who clicked".
Decisions are append-only; queue state is derived
The items table is mutable working state. The decisions table is not written to twice for the same fact and is never updated in place.
CREATE TABLE review_decisions (
id bigserial PRIMARY KEY,
item_id bigint NOT NULL REFERENCES review_items(id),
reviewer_id bigint NOT NULL,
verdict text NOT NULL
CHECK (verdict IN ('approve', 'reject', 'escalate', 'overturn')),
reason_code text NOT NULL,
evidence_hash text NOT NULL,
supersedes bigint REFERENCES review_decisions(id),
created_at timestamptz NOT NULL DEFAULT now()
);
REVOKE UPDATE, DELETE ON review_decisions FROM app_role;
A correction is a new row with verdict = 'overturn' and supersedes pointing at the row it replaces. The privilege revocation is deliberate: it is easy for an ORM, a data-fix script, or a well-meaning admin action to update a decision row, and the point of this table is that nobody can. Compliance conversations are much shorter when the storage layer itself refuses to rewrite history.
The reason_code must come from a closed enum, never a free-text field. Free text is unqueryable, so the queue can never tell you which rejection reason dominates, and the reasons that dominate are exactly the ones worth automating out of the human queue entirely. A closed enum turns the review backlog into a dataset about your own policy.
Applying the outcome is a separate, idempotent step
The submission above does not unblock an account, release a payout, or call a provider. It records a decision and returns. Everything downstream is driven off the decision row, and the decision id is the idempotency key.
@app.task(queue="review_apply", autoretry_for=(TransientError,), retry_backoff=True)
def apply_decision(decision_id: int) -> None:
decision = load_decision(decision_id)
with db.transaction() as conn:
claimed = conn.execute(
"""
INSERT INTO applied_decisions (decision_id, applied_at)
VALUES (%s, now())
ON CONFLICT (decision_id) DO NOTHING
RETURNING decision_id
""",
(decision_id,),
).fetchone()
if claimed is None:
return # already applied; a retry, not a second outcome
effects_for(decision.verdict)(conn, decision)
This split exists because the two halves have different failure semantics. A reviewer's click must never be lost and must never be asked for twice; a downstream effect may fail, be retried, and be observed by an operator. Coupling them means a provider timeout either loses a verdict a human already made or leaves the reviewer staring at a spinner and clicking again. Keeping the effect on its own queue also keeps it away from the pools that carry latency-sensitive traffic, which is a lesson we have written about elsewhere and keep re-learning in new shapes.
Abandonment is a signal, not noise
lease_count is the cheapest diagnostic in the whole design. An item that has been leased four times and still has no decision is almost never a run of lazy reviewers — it is an item the policy does not cover, and every reviewer who opened it correctly concluded that deciding was riskier than closing the tab.
UPDATE review_items
SET queue = 'escalation', priority = priority + 10
WHERE state = 'pending'
AND lease_count >= 3
AND (lease_expires IS NULL OR lease_expires < now());
Without this, ambiguous items circulate forever and quietly consume the most expensive resource in the system. With it, the queue surfaces the gaps in the rules to whoever owns the rules.
What it costs to run
Two tables, one partial index, one small join table for applied decisions, and no scheduled sweeper. The tuning surface is a single number — lease TTL — and it has an honest tradeoff on both ends: too short and reviewers lose work mid-analysis, too long and a crashed session parks an item for the duration. We start from the observed time-to-decision distribution and set the TTL above its long tail rather than picking a round number.
Three things are worth measuring: the histogram of lease_count (rising means the policy is drifting away from reality), the age of the oldest pending item per queue (the only backlog metric operations actually feels), and the count of decisions with no corresponding row in applied_decisions older than a few minutes, which is the alert that means a human made a call and the system did not carry it out.
None of this is exotic. It is one SKIP LOCKED query, one fenced write, one insert-only table, and one worker. The reason we build it the same way every time is that the alternative fails in production in ways that are expensive and specific: a duplicate approval on a case that should have been rejected, or an audit request that the database cannot answer because the row was updated in place.
Originally published on shipmindlabs.com — where we write about payment systems, infrastructure and marketplace backends.
Top comments (0)