The Weekend That Changed How I Queue Jobs
Our e-commerce app had three async tasks piling up: order confirmation emails, product image resizing, and PDF invoice generation. None of them could block the web request. The obvious answer was Redis and Celery. But on a 2GB VPS already running PostgreSQL, adding Redis meant another stateful service — its own persistence config, memory limits, and something else to wake up for at 3am.
What if PostgreSQL — already running, already backed up — could handle the job queue too? Turns out it can. Two features buried in the docs make it work: SKIP LOCKED and Advisory Locks.
Why Background Task Queues Break — The Root Cause
The core problem with any job queue is concurrent workers competing for the same task. Without proper locking, two workers grab the same job, process it twice, and your customer gets duplicate emails — or your payment API gets double-charged.
Each popular tool attacks this differently:
- Redis + BullMQ: Uses atomic Redis operations to dequeue safely. Works well, but you’re now maintaining Redis alongside your database — separate backup, separate monitoring, separate eviction policy tuning.
- RabbitMQ: Full message broker with delivery guarantees and complex routing. For “process these 500 emails in the background,” you’ll spend more time configuring vhosts and exchanges than writing worker logic.
- PostgreSQL LISTEN/NOTIFY: Lightweight pub/sub, not a job queue — messages don’t survive worker restarts, and there’s no retry logic built in.
The real issue isn’t the tool — it’s that most teams reach for distributed infrastructure when their data already lives in a database that can handle this natively.
Solutions Compared: Redis vs RabbitMQ vs PostgreSQL
Redis-Based Queues
Already running Redis for caching? Adding BullMQ or Celery is an easy bolt-on. The dequeue logic is atomic and fast. The downside: two data stores to back up, monitor, and keep in sync. Job data lives in Redis, business data lives in PostgreSQL — cross-referencing them during a 3am incident is genuinely painful.
RabbitMQ
For complex routing — fan-out, topic exchanges, priority queues — RabbitMQ earns its complexity. For “process these 500 emails in the background,” it’s overkill. You’ll spend more time configuring vhost permissions than writing actual worker logic.
PostgreSQL with SKIP LOCKED
This is the one that surprised me. PostgreSQL’s SELECT ... FOR UPDATE SKIP LOCKED — added in version 9.5 — was built for exactly this pattern. Multiple workers can safely dequeue jobs concurrently without claiming the same job twice.
Building the Queue: The PostgreSQL Approach
Step 1: Create the Jobs Table
CREATE TABLE jobs (
id BIGSERIAL PRIMARY KEY,
queue TEXT NOT NULL DEFAULT 'default',
payload JSONB NOT NULL,
status TEXT NOT NULL DEFAULT 'pending'
CHECK (status IN ('pending', 'running', 'done', 'failed')),
attempts INTEGER NOT NULL DEFAULT 0,
max_attempts INTEGER NOT NULL DEFAULT 3,
scheduled_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_jobs_queue_status_scheduled
ON jobs (queue, status, scheduled_at)
WHERE status = 'pending';The partial index on status = 'pending' is critical — workers query pending jobs constantly, and this index keeps it fast even with millions of rows in the table.
Step 2: The SKIP LOCKED Dequeue Pattern
-- Worker claims one job atomically
BEGIN;
UPDATE jobs
SET
status = 'running',
attempts = attempts + 1,
updated_at = NOW()
WHERE id = (
SELECT id
FROM jobs
WHERE queue = 'default'
AND status = 'pending'
AND scheduled_at <= NOW()
ORDER BY scheduled_at ASC
LIMIT 1
FOR UPDATE SKIP LOCKED
)
RETURNING *;
COMMIT;SKIP LOCKED is the key. Instead of blocking on a locked row — which causes worker pile-ups — a worker that encounters a locked job simply skips it and picks the next available one. Ten workers can run this exact query simultaneously. Each claims a different job. The database guarantees it.
Step 3: Python Worker Implementation
import psycopg2
import json
import time
DEQUEUE_SQL = """
UPDATE jobs
SET status = 'running', attempts = attempts + 1, updated_at = NOW()
WHERE id = (
SELECT id FROM jobs
WHERE queue = %s
AND status = 'pending'
AND scheduled_at <= NOW()
ORDER BY scheduled_at ASC
LIMIT 1
FOR UPDATE SKIP LOCKED
)
RETURNING id, payload, attempts, max_attempts;
"""
COMPLETE_SQL = "UPDATE jobs SET status = 'done', updated_at = NOW() WHERE id = %s"
FAIL_SQL = """
UPDATE jobs
SET status = CASE WHEN attempts >= max_attempts THEN 'failed' ELSE 'pending' END,
updated_at = NOW(),
scheduled_at = NOW() + (INTERVAL '1 minute' * POWER(2, attempts - 1))
WHERE id = %s
"""
def run_worker(dsn: str, queue: str = "default"):
conn = psycopg2.connect(dsn)
conn.autocommit = False
while True:
with conn.cursor() as cur:
cur.execute(DEQUEUE_SQL, (queue,))
row = cur.fetchone()
if row is None:
conn.commit()
time.sleep(2)
continue
job_id, payload, attempts, max_attempts = row
conn.commit()
try:
process_job(payload)
with conn.cursor() as cur:
cur.execute(COMPLETE_SQL, (job_id,))
conn.commit()
print(f"Job {job_id} done")
except Exception as e:
print(f"Job {job_id} failed: {e}")
with conn.cursor() as cur:
cur.execute(FAIL_SQL, (job_id,))
conn.commit()
def process_job(payload: dict):
job_type = payload.get("type")
if job_type == "send_email":
send_email(payload["to"], payload["subject"], payload["body"])
elif job_type == "resize_image":
resize_image(payload["image_id"], payload["width"], payload["height"])
else:
raise ValueError(f"Unknown job type: {job_type}")The retry logic in FAIL_SQL deserves a closer look. Failed jobs reschedule with true exponential backoff — 1 minute after the first failure, 2 minutes after the second, 4 minutes after the third. Once a job hits max_attempts, it flips to failed status — visible for inspection, not quietly discarded.
Enqueuing Jobs
def enqueue(conn, queue: str, payload: dict, delay_seconds: int = 0):
sql = """
INSERT INTO jobs (queue, payload, scheduled_at)
VALUES (%s, %s, NOW() + (%s * INTERVAL '1 second'))
RETURNING id
"""
with conn.cursor() as cur:
cur.execute(sql, (queue, json.dumps(payload), delay_seconds))
job_id = cur.fetchone()[0]
conn.commit()
return job_id
# Enqueue immediately
enqueue(conn, "default", {"type": "send_email", "to": "[email protected]",
"subject": "Order confirmed", "body": "..."})
# Enqueue with a 5-minute delay
enqueue(conn, "default", {"type": "resize_image", "image_id": 42,
"width": 800, "height": 600}, delay_seconds=300)When to Add Advisory Locks
Advisory Locks solve a different problem: ensuring only one instance of a specific operation runs at a time across your entire system — not per-row, but per-resource.
Here’s where it matters. Say you have a “sync user data” job. Two separate rows, queued minutes apart, could end up running simultaneously for the same user. SKIP LOCKED won’t prevent that — they’re different rows, both legitimately claimable. You need a higher-level lock on the user ID itself.
import hashlib
def process_with_advisory_lock(conn, resource_id: int, operation):
"""Only one worker can hold the lock for a given resource_id at a time."""
with conn.cursor() as cur:
# Returns False immediately if already locked — no blocking
cur.execute("SELECT pg_try_advisory_lock(%s)", (resource_id,))
acquired = cur.fetchone()[0]
if not acquired:
print(f"Resource {resource_id} already being processed, skipping")
return False
try:
operation()
return True
finally:
with conn.cursor() as cur:
cur.execute("SELECT pg_advisory_unlock(%s)", (resource_id,))
conn.commit()
# In your job processor
def process_job(conn, payload: dict):
if payload["type"] == "sync_user":
user_id = payload["user_id"]
# Use hashlib — Python's built-in hash() is randomized per process, not safe here
lock_key = int(hashlib.md5(f"sync_user:{user_id}".encode()).hexdigest(), 16) % (2**31)
process_with_advisory_lock(conn, lock_key, lambda: sync_user(user_id))Monitoring Your Queue
Debugging a PostgreSQL queue is refreshingly direct — standard SQL, no separate dashboard to log into.
-- Queue depth by status
SELECT queue, status, COUNT(*) AS count
FROM jobs
GROUP BY queue, status
ORDER BY queue, status;
-- Stuck jobs running longer than 10 minutes (worker crashed?)
SELECT id, payload, attempts, updated_at
FROM jobs
WHERE status = 'running'
AND updated_at < NOW() - INTERVAL '10 minutes';
-- Reset stuck jobs back to pending
UPDATE jobs
SET status = 'pending', updated_at = NOW()
WHERE status = 'running'
AND updated_at < NOW() - INTERVAL '10 minutes';I keep a small script that exports queue stats to CSV for weekly review. When I need to dig into failure patterns — grouping by error type or payload shape — I run that CSV through toolcraft.app’s CSV to JSON converter to get a format Python can load directly. It runs entirely in the browser, so job payloads containing user data never leave my machine.
The Best Approach: When to Use What
After several months running this in production, here’s the decision framework:
- PostgreSQL SKIP LOCKED: Default choice for most apps. Zero extra infrastructure, jobs survive database restarts, and you can JOIN job rows against your business data when debugging. Handles thousands of jobs per minute without issue.
- Add Advisory Locks: When you need per-resource mutual exclusion — preventing two workers from concurrently processing different job rows that target the same entity.
- Switch to Redis/BullMQ: When you’re processing tens of thousands of jobs per second, need sub-millisecond latency, or your team already runs Redis and the dashboard ecosystem justifies the dependency.
- RabbitMQ: Only when you genuinely need complex routing — fan-out to multiple consumer types, topic-based routing, or inter-service messaging at scale.
Most web apps — user signups, order processing, notification delivery, report generation — don’t need a message broker. PostgreSQL handles the queue cleanly, keeps the stack simple, and gives you SQL-powered observability that no Redis dashboard can match.
One practical caveat: PostgreSQL won’t push jobs to workers. Workers need to poll. A 2-second sleep between polls handles most workloads fine. If you need sub-second job pickup, pair SKIP LOCKED with LISTEN/NOTIFY. Add a trigger that fires a notification on INSERT, wake workers immediately on arrival, and keep the poll loop as a fallback for any notifications missed during restarts.
