TL;DR — SELECT ... FOR UPDATE SKIP LOCKED plus LISTEN/NOTIFY gives you a correct, low‑latency job queue built directly into Postgres. You get exactly‑once claiming, no phantom jobs, and no extra infrastructure. The trade‑off: every dequeue is a row update that bloats the table and churns the WAL, so you’re capped by autovacuum and max_connections long before you’d saturate a dedicated broker. If your system does under a few thousand jobs per second and the jobs are tied to existing transactions, this is the right default. When you’ve measured that you need more throughput or fan‑out, then you add Redis or RabbitMQ.
The queue problem people over‑complicate
You sign up a user, commit the row, and now you need to send a welcome email. The common reflex is to push a message to Redis or RabbitMQ. Now you have a second system that can disagree with your database.
- Enqueue after commit → network blip, email never sent.
- Enqueue inside the DB transaction → rollback after the broker write, phantom job.
When the job is derived from the same data you just committed, there’s no reason to add a second state store. Postgres already is a state store, and it already has row locking and push notifications. Use them and you remove a whole class of consistency bugs.
The core pattern: SELECT ... FOR UPDATE SKIP LOCKED
The fear that killed Postgres‑backed queues for years was lock‑wait pile‑up. A naive worker does:
DELETE FROM jobs WHERE id = (SELECT id FROM jobs ORDER BY id LIMIT 1);
Two workers target the same row, one blocks on the other’s row lock, and suddenly your “queue” is a serial bottleneck.
SKIP LOCKED fixes that. SELECT … FOR UPDATE SKIP LOCKED acquires row‑level locks and skips rows already locked by concurrent transactions. Two workers each run the same query, each gets a different unlocked row, and neither blocks waiting. That’s the whole magic — parallel dequeue with zero lock contention, no external broker required.
A working job queue in ~40 lines of SQL
Let’s build one. First, the jobs table. I use clock_timestamp() because it returns the actual current time when the statement executes — it is more accurate for job scheduling than now(), which holds the transaction start time.
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','processing','failed','completed')),
run_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
locked_at TIMESTAMPTZ,
attempts INT NOT NULL DEFAULT 0,
max_attempts INT NOT NULL DEFAULT 5,
created_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp()
);
CREATE INDEX idx_jobs_dequeue ON jobs (queue, status, run_at)
WHERE status = 'pending';
locked_at serves as a heartbeat for visibility‑timeout handling. The partial index makes dequeue scans fast: only pending rows are indexed.
Enqueue helper
CREATE OR REPLACE FUNCTION enqueue_job(
_queue TEXT,
_payload JSONB,
_run_at TIMESTAMPTZ DEFAULT clock_timestamp(),
_max_attempts INT DEFAULT 5
) RETURNS BIGINT AS $$
INSERT INTO jobs (queue, payload, run_at, max_attempts)
VALUES (_queue, _payload, _run_at, _max_attempts)
RETURNING id;
$$ LANGUAGE sql;
Dequeue — one statement, no polling
Claim a job, set its status to processing, stamp locked_at, and return the row — all in a single atomic statement with SKIP LOCKED:
WITH next_job AS (
SELECT j.id
FROM jobs j
WHERE j.queue = 'default'
AND j.status = 'pending'
AND j.run_at <= clock_timestamp()
ORDER BY j.run_at
LIMIT 1
FOR UPDATE SKIP LOCKED
)
UPDATE jobs j
SET status = 'processing',
locked_at = clock_timestamp(),
attempts = attempts + 1,
updated_at = clock_timestamp()
FROM next_job nj
WHERE j.id = nj.id
RETURNING j.*;
No two‑phase, no separate claim step. Two concurrent sessions running the exact same query see different rows, with no blocking.
Demo: two workers, no lock pile‑up
Open two psql sessions. Session 1:
postgres=# SELECT enqueue_job('default', '{"task":"send_welcome","user_id":42}');
enqueue_job
-------------
1
postgres=# SELECT enqueue_job('default', '{"task":"send_welcome","user_id":99}');
enqueue_job
-------------
2
Now run the dequeue query in both sessions before either commits:
Session 1:
id | queue | payload | status | attempts | locked_at
----+---------+----------------------------+-----------+----------+------------------------------
1 | default | {"task":"send_welcome",...} | processing| 1 | 2025-03-15 14:32:01.123456+00
Session 2, same exact query:
id | queue | payload | status | attempts | locked_at
----+---------+----------------------------+-----------+----------+------------------------------
2 | default | {"task":"send_welcome",...} | processing| 1 | 2025-03-15 14:32:01.123789+00
Neither blocked. They picked different rows. That’s SKIP LOCKED at work.
Push instead of poll: LISTEN/NOTIFY
Polling burns CPU and imposes a latency floor of half your polling interval. Postgres’s built‑in pub‑sub can push a notification the instant a job is committed.
Create a trigger that only fires when a new row is inserted with status pending:
CREATE OR REPLACE FUNCTION notify_job_insert()
RETURNS trigger AS $$
BEGIN
PERFORM pg_notify('job_channel', NEW.id::text);
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER trg_job_insert
AFTER INSERT ON jobs
FOR EACH ROW
WHEN (NEW.status = 'pending')
EXECUTE FUNCTION notify_job_insert();
NOTIFY is transactional: if the inserting
Top comments (0)