- Book: Event-Driven Architecture Pocket Guide
- Also by me: Thinking in Go (2-book series) — Complete Guide to Go Programming + Hexagonal Architecture in Go
- My project: Hermes IDE | GitHub — an IDE for developers who ship with Claude Code and other AI coding tools
- Me: xgabriel.com | GitHub
Say your receipt job renders a PDF and emails it. It usually takes 20 seconds, and the worker's lease on the job is 30. Then one morning the PDF service is slow, the job takes 45 seconds, and a customer gets two receipts. Nothing crashed and nothing was logged: two workers ran the same job at once, each doing exactly what its code said.
This is a classic way for a Postgres-backed job queue to misbehave. Postgres is doing what you asked: a lease expired while the work was still running. This tutorial builds the queue in plain SQL plus a small Python worker, triggers the double run on purpose, then makes it safe three ways: a heartbeat, a fencing token, and an idempotency key. Everything runs against a local Postgres in Docker.
The queue: one table and one claim query
Start Postgres 18 from the official Docker image (18 is the current major release) and create two tables. jobs is the queue. side_effects stands in for "the email went out": one row every time a job's effect happens.
docker run --name pgq -e POSTGRES_PASSWORD=postgres \
-p 5432:5432 -d postgres:18
-- schema.sql
CREATE TABLE jobs (
id bigserial PRIMARY KEY,
payload jsonb NOT NULL DEFAULT '{}',
status text NOT NULL DEFAULT 'queued',
attempt int NOT NULL DEFAULT 0,
locked_by text,
lease_until timestamptz
);
CREATE TABLE side_effects (
job_id bigint NOT NULL,
attempt int NOT NULL,
worker text NOT NULL,
at timestamptz NOT NULL DEFAULT now()
);
docker exec -i pgq psql -U postgres < schema.sql
The claim is one statement. It picks the oldest job that is either queued or whose lease has expired, bumps attempt, and stamps a new lease:
UPDATE jobs
SET status = 'running',
attempt = attempt + 1,
locked_by = %(worker)s,
lease_until = now() + %(lease)s::interval
WHERE id = (
SELECT id FROM jobs
WHERE status = 'queued'
OR (status = 'running' AND lease_until < now())
ORDER BY id
LIMIT 1
FOR UPDATE SKIP LOCKED
)
RETURNING id, attempt
SKIP LOCKED has been in Postgres since 9.5, released January 2016. The SELECT docs describe it plainly: rows that cannot be locked immediately are skipped, which is "not suitable for general purpose work, but can be used to avoid lock contention with multiple consumers accessing a queue-like table." Two workers polling at the same instant never claim the same row.
Notice what happens next, though. The claim commits right away, and the row lock goes with it. From here on, the only thing saying "this job is taken" is lease_until. Every lease timestamp comes from Postgres's now(), which the docs define as the start of the current transaction, so clock drift between worker machines doesn't enter into it.
Why not keep the transaction open and hold the row lock for the whole job? You can, but each running job then pins a connection and an open transaction, and Postgres can't clean up a row version while it is still potentially visible to other transactions. It also has the same failure you're about to see: if the connection drops mid-job, the lock is released and another worker takes the row while the first process keeps running.
Reproduce the double run
Install psycopg 3 with pip install "psycopg[binary]" and save this as worker.py. The lease is 3 seconds and the job takes 10, so the lease will expire mid-job on purpose.
# worker.py
import os
import sys
import time
import psycopg
DSN = "postgresql://postgres:postgres@localhost:5432/postgres"
WORKER = sys.argv[1]
LEASE = "3 seconds"
JOB_SECONDS = 10
CLAIM = """
UPDATE jobs
SET status = 'running',
attempt = attempt + 1,
locked_by = %(worker)s,
lease_until = now() + %(lease)s::interval
WHERE id = (
SELECT id FROM jobs
WHERE status = 'queued'
OR (status = 'running' AND lease_until < now())
ORDER BY id
LIMIT 1
FOR UPDATE SKIP LOCKED
)
RETURNING id, attempt
"""
EFFECT = """
INSERT INTO side_effects (job_id, attempt, worker)
VALUES (%(id)s, %(attempt)s, %(worker)s)
"""
COMPLETE = """
UPDATE jobs
SET status = 'done', locked_by = NULL, lease_until = NULL
WHERE id = %(id)s
"""
def run_job(job_id):
for _ in range(JOB_SECONDS):
time.sleep(1) # pretend to render a PDF
def finish(conn, job_id, attempt):
args = {"id": job_id, "attempt": attempt, "worker": WORKER}
conn.execute(EFFECT, args)
conn.execute(COMPLETE, args)
def main():
print(f"{WORKER}: pid {os.getpid()}")
conn = psycopg.connect(DSN, autocommit=True)
while True:
row = conn.execute(
CLAIM, {"worker": WORKER, "lease": LEASE}
).fetchone()
if row is None:
time.sleep(1)
continue
job_id, attempt = row
print(f"{WORKER}: claimed {job_id}, attempt {attempt}")
run_job(job_id)
finish(conn, job_id, attempt)
print(f"{WORKER}: finished {job_id}")
main()
Start python worker.py A and python worker.py B in two terminals, then enqueue one job:
docker exec pgq psql -U postgres -c \
"INSERT INTO jobs (payload) VALUES ('{\"order_id\": 42}')"
Walk through what the code does. One worker, say A, claims job 1 as attempt 1 with a lease ending at t=3. B polls every second, and on the first poll after t=3 the lease has expired, so B claims job 1 as attempt 2. At t=10, A writes its side effect and marks the job done, even though B owns it now. A few seconds later B writes its side effect and marks it done again. Check the damage:
SELECT job_id, attempt, worker FROM side_effects;
Two rows for one job. That's two receipts.
When the lease expires mid-job, the queue hands the same job to a second worker.
Fix 1: a heartbeat that extends the lease
A slow job is still a live job, so the worker should keep saying it's alive. A heartbeat thread renews the lease every second while the job runs. It uses its own connection, and it renews only if this worker still owns the attempt it claimed:
import threading
HEARTBEAT = """
UPDATE jobs
SET lease_until = now() + %(lease)s::interval
WHERE id = %(id)s AND attempt = %(attempt)s
AND status = 'running'
"""
class LeaseLost(Exception):
pass
def heartbeat(job_id, attempt, stop, lost):
args = {"id": job_id, "attempt": attempt, "lease": LEASE}
try:
with psycopg.connect(DSN, autocommit=True) as hb:
while not stop.wait(1.0):
if hb.execute(HEARTBEAT, args).rowcount == 0:
break
except psycopg.Error:
pass
if not stop.is_set():
lost.set()
def run_job(job_id, lost):
for _ in range(JOB_SECONDS):
time.sleep(1) # pretend to render a PDF
if lost.is_set():
raise LeaseLost(job_id)
If the renewal touches zero rows, or the database can't be reached, the heartbeat assumes the lease is gone and sets lost. The new run_job replaces the first one and checks the flag between steps. In main(), replace the three lines after the "claimed" print with:
stop, lost = threading.Event(), threading.Event()
hb = threading.Thread(
target=heartbeat,
args=(job_id, attempt, stop, lost),
daemon=True,
)
hb.start()
try:
run_job(job_id, lost)
finish(conn, job_id, attempt)
print(f"{WORKER}: finished {job_id}")
except LeaseLost:
print(f"{WORKER}: lost lease on {job_id}")
finally:
stop.set()
hb.join()
Run the experiment again (TRUNCATE side_effects first) and the slow job finishes once: the lease keeps moving ahead of the work, so B never sees it expire. This is the same shape as Amazon SQS, where the default visibility timeout is 30 seconds and AWS recommends a heartbeat that calls ChangeMessageVisibility to extend it during long processing.
A heartbeat only works while the process is running, though. Pause worker A in its last couple of seconds of work, using the pid it printed:
kill -STOP <pid>; sleep 5; kill -CONT <pid>
While A is frozen, no heartbeat goes out, the lease expires, and B claims attempt 2. When A wakes up, its main thread may pass its final lost check before the heartbeat thread notices, and finish() runs anyway. A garbage-collection pause, a VM migration or a network partition produces the same gap. Martin Kleppmann's analysis of distributed locking walks through exactly this: a client paused past its lease expiry goes ahead with an unsafe write without knowing the lease is gone.
Fix 2: a fencing token on completion
Kleppmann's answer is a fencing token, a number that increases every time the lock is acquired, checked by the system that receives the write. You already have one: attempt. Make completion conditional on it, and put the side effect in the same transaction:
COMPLETE = """
UPDATE jobs
SET status = 'done', locked_by = NULL, lease_until = NULL
WHERE id = %(id)s AND attempt = %(attempt)s
AND status = 'running'
"""
def finish(conn, job_id, attempt):
args = {"id": job_id, "attempt": attempt, "worker": WORKER}
with conn.transaction():
if conn.execute(COMPLETE, args).rowcount == 0:
raise LeaseLost(job_id)
conn.execute(EFFECT, args)
In autocommit mode, psycopg's conn.transaction() opens a transaction for the block, commits on a clean exit and rolls back if an exception escapes. So a stale worker's UPDATE matches zero rows, LeaseLost is raised, and its side-effect row never commits.
The race between the two workers is covered too. If B's claim has bumped attempt but not yet committed, A's UPDATE waits. Under Read Committed, Postgres then re-evaluates the WHERE clause against the updated row, sees attempt 2, and matches nothing. If A gets there first, B's claim skips the locked row, and after A commits the job is done and no longer claimable.
The check deliberately compares the token, not the clock. If A's lease expired but nobody reclaimed the job, A's result is still the only one, and it should be accepted.
The fencing token: the database accepts a completion only from the attempt that currently owns the job.
Fix 3: an idempotency key for effects outside the transaction
The fencing token works because the token check and the side effect share one transaction in one database. A real receipt goes through an email API. By the time a stale worker reaches finish(), the email is already sent, and the API has no idea what your attempt column says.
For those effects, pass a key the receiver can deduplicate on, derived from the job id, never from the attempt. Attempts change on every reclaim, so an attempt-based key dedupes nothing. Inside Postgres it is a unique constraint:
TRUNCATE side_effects;
ALTER TABLE side_effects
ADD CONSTRAINT one_effect_per_job UNIQUE (job_id);
EFFECT = """
INSERT INTO side_effects (job_id, attempt, worker)
VALUES (%(id)s, %(attempt)s, %(worker)s)
ON CONFLICT (job_id) DO NOTHING
"""
Over HTTP it is a header, where the API supports one. Stripe's idempotent requests take an Idempotency-Key header, so you would send something like Idempotency-Key: receipt-job-42, and a retry with the same key returns the saved first response. Read the fine print: Stripe can prune keys once they are at least 24 hours old, so a job retried days later is a new request.
What you have now: at-least-once, made safe
None of this gives you exactly-once execution. A worker can still send the email and die before it commits, and the next attempt will send again unless the email provider dedupes on your key. AWS says the same about its own queue: under SQS's at-least-once model, there is no guarantee a message won't be delivered more than once within the visibility timeout.
What you get is a clear map of which layer handles which failure:
- Slow job: the heartbeat keeps the lease ahead of the work.
- Paused or partitioned worker: the fencing token rejects its late completion and rolls back its database writes.
- Effect outside your database: an idempotency key derived from the job id lets the receiver drop the duplicate.
- Crash between an external call and the commit: nothing here prevents it. Plan for the retry.
Two additions before production. Cap retries with attempt < 5 in the claim query and a failed status, or a job that kills its worker gets reclaimed forever. And once the table grows, index only the rows the claim reads with CREATE INDEX ON jobs (id) WHERE status IN ('queued', 'running'), so the claim skips finished and failed rows.
If this was useful
Once a consumer can run twice, you need a plan for the duplicate, and that problem shows up in every event-driven system, well beyond a Postgres table. The Event-Driven Architecture Pocket Guide works through the patterns around it: outboxes, sagas and consumers that have to stay idempotent, plus the traps that show up when those pieces are missing.
Top comments (0)