A job queue decouples the moment work is requested from the moment it executes. Instead of reaching for a dedicated broker like RabbitMQ or Redis, many production systems use PostgreSQL itself as the queue backend. This approach trades raw throughput for operational simplicity: you get ACID guarantees, transactional enqueue (enqueue a job in the same transaction that creates the entity it processes), and zero additional infrastructure to monitor.
Libraries like Gue (Go), Graphile Worker (Node), and PGMQ implement the queue abstraction on top of PostgreSQL advisory locks and FOR UPDATE SKIP LOCKED. In a typical production application, PostgreSQL-backed queues drive webhook delivery, data synchronization, and email notification pipelines where reliability matters more than sub-millisecond latency.
Understanding how to build and tune these queues is essential for any team that already runs PostgreSQL and wants to avoid the operational burden of a separate message broker for moderate-throughput workloads (up to tens of thousands of jobs per second).
SELECT ... FOR UPDATE SKIP LOCKED to avoid contention.FOR UPDATE SKIP LOCKED is the secret weapon of PostgreSQL job queues. It lets multiple workers race for jobs without blocking each other: each worker atomically claims the next available row, skipping any row already locked by another worker.
Transactional enqueue ensures that if the business transaction rolls back, the job is never enqueued. This eliminates phantom jobs and the need for compensation logic that plagues two-system designs.
A job moves through a well-defined state machine: available → running → completed (or failed → available on retry). When a worker picks up a job, it sets the status to running and records a heartbeat timestamp. If the worker crashes, a reaper process detects stale heartbeats and returns the job to available.
The reaper pattern (a background process that reclaims orphaned jobs) is critical for fault tolerance. Without it, a crashed worker’s jobs remain locked forever, creating a silent queue stall.
Workers use FOR UPDATE SKIP LOCKED within a transaction. The worker fetches a batch of N jobs, processes them, and commits. If processing fails, the transaction rolls back and the jobs become available again. This provides natural at-least-once delivery without distributed coordination.
Jobs can carry a priority column and a run_at timestamp. The polling query orders by priority descending and filters run_at <= NOW(), enabling delayed execution and priority lanes. High-priority webhooks can jump ahead of bulk sync jobs without separate queues.
Instead of polling on a fixed interval, PostgreSQL’s LISTEN/NOTIFY mechanism can wake workers immediately when a new job is inserted. The enqueue trigger sends a notification, and idle workers react within milliseconds rather than waiting for the next poll cycle.
dead_jobs table for manual inspection rather than silently dropping them.SELECT count(*) FROM jobs WHERE status = 'available'.func EnqueueWebhookDelivery(ctx context.Context, tx *sql.Tx, eventID string, payload json.RawMessage) error {
_, err := tx.ExecContext(ctx, `
INSERT INTO jobs (queue, type, payload, run_at, max_retries)
VALUES ('webhooks', 'deliver', $1, NOW(), 5)
`, payload)
return err
}
// Usage: enqueue in the same tx that records the event
func HandleStripeEvent(ctx context.Context, db *sql.DB, event StripeEvent) error {
tx, _ := db.BeginTx(ctx, nil)
defer tx.Rollback()
if err := RecordEvent(ctx, tx, event); err != nil {
return err
}
if err := EnqueueWebhookDelivery(ctx, tx, event.ID, event.Data); err != nil {
return err
}
return tx.Commit()
}func (w *Worker) Poll(ctx context.Context) error {
tx, _ := w.db.BeginTx(ctx, nil)
defer tx.Rollback()
rows, err := tx.QueryContext(ctx, `
SELECT id, type, payload FROM jobs
WHERE queue = $1
AND status = 'available'
AND run_at <= NOW()
ORDER BY priority DESC, run_at ASC
LIMIT $2
FOR UPDATE SKIP LOCKED
`, w.queue, w.batchSize)
if err != nil {
return err
}
defer rows.Close()
var jobs []Job
for rows.Next() {
var j Job
rows.Scan(&j.ID, &j.Type, &j.Payload)
jobs = append(jobs, j)
}
for _, j := range jobs {
if err := w.process(ctx, tx, j); err != nil {
w.scheduleRetry(ctx, tx, j, err)
} else {
w.markComplete(ctx, tx, j)
}
}
return tx.Commit()
}CREATE TABLE jobs (
id BIGSERIAL PRIMARY KEY,
queue TEXT NOT NULL DEFAULT 'default',
type TEXT NOT NULL,
payload JSONB NOT NULL DEFAULT '{}',
status TEXT NOT NULL DEFAULT 'available',
priority INT NOT NULL DEFAULT 0,
run_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
attempts INT NOT NULL DEFAULT 0,
max_retries INT NOT NULL DEFAULT 3,
last_error TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_jobs_poll ON jobs (queue, status, priority DESC, run_at)
WHERE status = 'available';status = 'available', the polling query scans the entire (growing) table and degrades over time.FOR UPDATE lock while doing slow I/O (HTTP calls, file processing) blocks the row for the entire duration. Claim the job, commit, then process externally.LISTEN/NOTIFY or back off when the queue is empty.DELETE or archive operations, and run VACUUM to reclaim space.FOR UPDATE SKIP LOCKED enables lock-free concurrent job claiming across multiple workers.LISTEN/NOTIFY reduces job pickup latency from poll-interval to near-instant.