Build a Job Queue With pgx and SKIP LOCKED in Postgres
Build a production shaped Postgres job queue with pgx: batch SKIP LOCKED claims, retries with backoff, leases, a reaper, LISTEN wake-ups and graceful shutdown
A job queue in PostgreSQL is a table, one statement and a loop. The table holds jobs; the statement claims ready ones for a worker without letting two workers claim the same row; the loop runs them. An earlier article built the core and measured what SKIP LOCKED buys. This one builds the version you would put behind a real service, with pgx: claims in batches, retries that back off, a lease that long jobs renew, a reaper for dead workers, LISTEN wake-ups so idle workers react at once, and a shutdown that does not leave jobs stuck. It runs on Go 1.25.6 with pgx v5.11.0 and PostgreSQL 18.6, and each behaviour is demonstrated by the lab.
Tested with Go 1.25.6, pgx v5.11.0, PostgreSQL 18.6 in Docker, macOS on Apple Silicon (arm64)
It belongs to the database cornerstone. It uses the locking from SELECT FOR UPDATE, the wake-up pattern from pgx LISTEN and NOTIFY, the transaction rules from unit of work, and it complements the outbox pattern, which is a queue of this kind.
The design
One table, jobs682, with a status that is queued, running, done or dead. A job has a kind (which handler runs it), a JSON payload, an attempt count and a maximum, a run_at time (so a retry can be delayed), a locked_until lease and the last error. Two partial indexes keep the hot queries small whatever the table's size: one over queued jobs ordered by run_at and id, one over running jobs by lease. A trigger sends NOTIFY jobs682 after each insert statement, so workers wake when work arrives.
package main
import (
"context"
"encoding/json"
"fmt"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
const schema = `
drop table if exists jobs682;
create table jobs682 (
id bigserial primary key,
kind text not null,
payload jsonb not null default '{}',
status text not null default 'queued' check (status in ('queued', 'running', 'done', 'dead')),
attempts int not null default 0,
max_attempts int not null default 3,
run_at timestamptz not null default now(),
locked_until timestamptz,
last_error text,
done_at timestamptz
);
create index jobs682_ready on jobs682 (run_at, id) where status = 'queued';
create index jobs682_running on jobs682 (locked_until) where status = 'running';
create or replace function jobs682_notify() returns trigger language plpgsql as $$
begin perform pg_notify('jobs682', ''); return new; end $$;
create trigger jobs682_notify after insert on jobs682 for each statement execute function jobs682_notify();
`
type Job struct {
ID int64
Kind string
Payload json.RawMessage
Attempts int
Max int
}
type Queue struct{ Pool *pgxpool.Pool }
// EnqueueBatch adds many jobs in one round trip. Passing a pgx.Tx instead of the pool would add them in the caller's transaction.
func (q Queue) EnqueueBatch(ctx context.Context, kinds []string, payloads []any) error {
b := &pgx.Batch{}
for i, k := range kinds {
b.Queue("insert into jobs682 (kind, payload) values ($1, $2)", k, payloads[i])
}
return q.Pool.SendBatch(ctx, b).Close()
}
// Claim takes up to n ready jobs and leases them. skip locked lets many workers claim at once without waiting for each other.
func (q Queue) Claim(ctx context.Context, n int, lease time.Duration) ([]Job, error) {
rows, err := q.Pool.Query(ctx, `
update jobs682 set status = 'running', attempts = attempts + 1, locked_until = now() + $2::interval
where id in (
select id from jobs682 where status = 'queued' and run_at <= now()
order by run_at, id limit $1 for update skip locked)
returning id, kind, payload, attempts, max_attempts`, n, fmt.Sprintf("%d milliseconds", lease.Milliseconds()))
if err != nil {
return nil, err
}
return pgx.CollectRows(rows, func(r pgx.CollectableRow) (Job, error) {
var j Job
err := r.Scan(&j.ID, &j.Kind, &j.Payload, &j.Attempts, &j.Max)
return j, err
})
}
func (q Queue) Done(ctx context.Context, id int64) error {
_, err := q.Pool.Exec(ctx, "update jobs682 set status = 'done', done_at = now(), locked_until = null where id = $1", id)
return err
}
// Fail puts the job back with an exponential delay, or buries it when it has used all its attempts.
func (q Queue) Fail(ctx context.Context, j Job, cause error) error {
_, err := q.Pool.Exec(ctx, `
update jobs682 set last_error = $2, locked_until = null,
status = case when attempts >= max_attempts then 'dead' else 'queued' end,
run_at = now() + (interval '20 milliseconds' * power(2, attempts))
where id = $1`, j.ID, cause.Error())
return err
}
// Release gives a claimed job back untouched, which is how a worker that is shutting down returns what it has not started.
func (q Queue) Release(ctx context.Context, id int64) error {
_, err := q.Pool.Exec(ctx, "update jobs682 set status = 'queued', attempts = attempts - 1, locked_until = null where id = $1 and status = 'running'", id)
return err
}
// Extend renews the lease of a job that is taking longer than expected.
func (q Queue) Extend(ctx context.Context, id int64, lease time.Duration) error {
_, err := q.Pool.Exec(ctx, "update jobs682 set locked_until = now() + $2::interval where id = $1 and status = 'running'", id, fmt.Sprintf("%d milliseconds", lease.Milliseconds()))
return err
}
// Reap returns jobs whose lease ran out, because their worker died, to the queue.
func (q Queue) Reap(ctx context.Context) (int64, error) {
tag, err := q.Pool.Exec(ctx, "update jobs682 set status = 'queued', locked_until = null where status = 'running' and locked_until < now()")
return tag.RowsAffected(), err
}
func (q Queue) Counts(ctx context.Context) (map[string]int, error) {
rows, err := q.Pool.Query(ctx, "select status, count(*)::int from jobs682 group by status")
if err != nil {
return nil, err
}
out := map[string]int{}
for rows.Next() {
var s string
var n int
if err := rows.Scan(&s, &n); err != nil {
return nil, err
}
out[s] = n
}
return out, rows.Err()
}Enqueue
EnqueueBatch queues many inserts in one pgx batch, one round trip for hundreds of jobs. In a service the same code takes a pgx.Tx (the DB interface both the pool and a transaction satisfy), so a job can be inserted in the same transaction as the data it belongs to: a rolled back request never leaves a job behind, and a committed one always has its job. That property is the strongest reason to put the queue in the database.
Claim
Claim(ctx, n, lease) is one statement:
update jobs682 set status = 'running', attempts = attempts + 1, locked_until = now() + $2 where id in (select id from jobs682 where status = 'queued' and run_at <= now() order by run_at, id limit $1 for update skip locked) returning ...
The inner select picks the next n ready jobs and locks them, skipping rows other workers have locked, so any number of workers can claim concurrently without waiting for each other or taking the same job. The outer update marks them running, counts the attempt and sets the lease, and returning hands the rows back. Claiming in batches of 10 means one round trip and one lock acquisition per ten jobs, which is most of the speed difference between a toy queue and a usable one. The cost of a large batch is fairness and recovery: ten jobs claimed by a slow worker wait behind each other.
Complete, fail, extend, reap
Done sets done. Fail records the error and decides between a retry and the grave in SQL: the status becomes dead when attempts >= max_attempts and queued otherwise, and run_at moves into the future by 20ms * 2^attempts, an exponential backoff computed by the database, so a failing job does not hammer a failing dependency. Extend renews a lease. Reap returns every running job whose lease has expired to the queue, which is how work claimed by a crashed worker is recovered. Release gives a claimed job back untouched, used on shutdown.
The workers
package main
import (
"context"
"errors"
"sync"
"time"
"github.com/jackc/pgx/v5"
)
type Handler func(ctx context.Context, j Job) error
type Workers struct {
Q Queue
Handlers map[string]Handler
Count int
Batch int
Lease time.Duration
DSN string
}
// Run starts the workers and a listener that wakes them, and returns when ctx ends and every running job has finished.
func (w Workers) Run(ctx context.Context) {
wake := make(chan struct{}, 1)
go w.listen(ctx, wake)
var wg sync.WaitGroup
for i := 0; i < w.Count; i++ {
wg.Add(1)
go func() {
defer wg.Done()
w.loop(ctx, wake)
}()
}
// The reaper returns the jobs of dead workers to the queue.
wg.Add(1)
go func() {
defer wg.Done()
t := time.NewTicker(200 * time.Millisecond)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
_, _ = w.Q.Reap(context.Background())
}
}
}()
wg.Wait()
}
func (w Workers) listen(ctx context.Context, wake chan<- struct{}) {
for ctx.Err() == nil {
conn, err := pgx.Connect(ctx, w.DSN)
if err != nil {
time.Sleep(100 * time.Millisecond)
continue
}
if _, err := conn.Exec(ctx, "listen jobs682"); err == nil {
for {
if _, err := conn.WaitForNotification(ctx); err != nil {
break
}
select {
case wake <- struct{}{}:
default:
}
}
}
conn.Close(context.Background())
}
}
func (w Workers) loop(ctx context.Context, wake <-chan struct{}) {
for ctx.Err() == nil {
jobs, err := w.Q.Claim(ctx, w.Batch, w.Lease)
if err != nil {
if ctx.Err() != nil {
return
}
time.Sleep(50 * time.Millisecond)
continue
}
if len(jobs) == 0 {
// Nothing ready: wait for a notification, with a poll as the safety net for delayed retries.
select {
case <-ctx.Done():
return
case <-wake:
case <-time.After(100 * time.Millisecond):
}
continue
}
for _, j := range jobs {
if ctx.Err() != nil {
// Shutting down: do not start more work, hand the unstarted job back.
_ = w.Q.Release(context.Background(), j.ID)
continue
}
w.run(j)
}
}
}
// run executes one job to the end even if ctx has been cancelled, which is what a graceful shutdown means.
func (w Workers) run(j Job) {
bg := context.Background()
h, ok := w.Handlers[j.Kind]
if !ok {
_ = w.Q.Fail(bg, j, errors.New("no handler for "+j.Kind))
return
}
jobCtx, cancel := context.WithCancel(bg)
defer cancel()
// A job that runs longer than half its lease renews it.
go func() {
t := time.NewTicker(w.Lease / 2)
defer t.Stop()
for {
select {
case <-jobCtx.Done():
return
case <-t.C:
_ = w.Q.Extend(bg, j.ID, w.Lease)
}
}
}()
if err := h(jobCtx, j); err != nil {
_ = w.Q.Fail(bg, j, err)
return
}
_ = w.Q.Done(bg, j.ID)
}Run starts N worker goroutines, a listener and a reaper, and returns when the context ends and the workers have finished. Each worker loops: claim a batch; if nothing is ready, wait for a notification, a 100 millisecond timer or the context. The timer is the safety net for two things: notifications can be lost (they are not stored), and delayed retries become ready without any insert to announce them. The listener holds its own connection outside the pool, as the listen article recommends, and sends a signal on a one slot channel so many notifications collapse into one wake up.
run executes a single job. Two details carry the design. It runs the handler under a background context, not the worker's, so that a shutdown signal does not cancel a job in the middle: finishing running jobs is what graceful means. And a small goroutine renews the lease at half its length while the handler runs, so a job that takes longer than the lease is not mistaken for a dead one and handed to a second worker.
What the run shows
package main
import (
"context"
"errors"
"fmt"
"log"
"sync/atomic"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
const dsn = "postgres://postgres:[email protected]:55432/lab?sslmode=disable"
func must(err error) {
if err != nil {
log.Fatal(err)
}
}
func main() {
ctx := context.Background()
pool, err := pgxpool.New(ctx, dsn+"&pool_max_conns=12")
must(err)
defer pool.Close()
_, err = pool.Exec(ctx, schema)
must(err)
q := Queue{pool}
var ran atomic.Int64
handlers := map[string]Handler{
"email": func(ctx context.Context, j Job) error { ran.Add(1); time.Sleep(time.Millisecond); return nil },
// fails on its first attempt and succeeds on the second
"flaky": func(ctx context.Context, j Job) error {
ran.Add(1)
if j.Attempts == 1 {
return errors.New("temporary failure")
}
return nil
},
// never works: it should end up dead after three attempts
"broken": func(ctx context.Context, j Job) error { ran.Add(1); return errors.New("always fails") },
// takes longer than its lease and keeps it alive by renewal
"slow": func(ctx context.Context, j Job) error { ran.Add(1); time.Sleep(900 * time.Millisecond); return nil },
}
// 500 jobs of four kinds.
var kinds []string
var payloads []any
for i := 0; i < 500; i++ {
k := "email"
switch {
case i%50 == 7:
k = "flaky"
case i%125 == 3:
k = "broken"
case i == 100:
k = "slow"
}
kinds = append(kinds, k)
payloads = append(payloads, map[string]int{"n": i})
}
must(q.EnqueueBatch(ctx, kinds, payloads))
counts, _ := q.Counts(ctx)
fmt.Println("1. enqueued 500 jobs in one batch:", counts)
w := Workers{Q: q, Handlers: handlers, Count: 4, Batch: 10, Lease: 400 * time.Millisecond, DSN: dsn}
runCtx, stop := context.WithCancel(ctx)
done := make(chan struct{})
start := time.Now()
go func() { w.Run(runCtx); close(done) }()
// Wait until nothing is queued or running.
for {
time.Sleep(50 * time.Millisecond)
c, _ := q.Counts(ctx)
if c["queued"] == 0 && c["running"] == 0 {
break
}
if time.Since(start) > 20*time.Second {
log.Fatal("did not finish")
}
}
elapsed := time.Since(start)
stop()
<-done
counts, _ = q.Counts(ctx)
fmt.Printf("2. four workers, claiming 10 at a time, finished in %v: %v\n", elapsed.Round(10*time.Millisecond), counts)
fmt.Println(" handler calls in total:", ran.Load(), "(500 jobs plus the retries)")
var maxAttempts, flakyDone, brokenDead, slowAttempts int
var lastErr string
must(pool.QueryRow(ctx, "select max(attempts) from jobs682").Scan(&maxAttempts))
must(pool.QueryRow(ctx, "select count(*) from jobs682 where kind = 'flaky' and status = 'done' and attempts = 2").Scan(&flakyDone))
must(pool.QueryRow(ctx, "select count(*), max(last_error) from jobs682 where kind = 'broken' and status = 'dead' and attempts = 3").Scan(&brokenDead, &lastErr))
must(pool.QueryRow(ctx, "select attempts from jobs682 where kind = 'slow'").Scan(&slowAttempts))
fmt.Printf("3. flaky jobs that succeeded on the second attempt: %d; broken jobs buried after three attempts: %d (last error %q)\n", flakyDone, brokenDead, lastErr)
fmt.Printf(" the slow job ran for 900ms on a 400ms lease and was attempted %d time, because it renewed its lease\n", slowAttempts)
// A worker that dies: jobs are claimed and never finished, then the reaper returns them.
must(q.EnqueueBatch(ctx, []string{"email", "email", "email"}, []any{1, 2, 3}))
claimed, err := q.Claim(ctx, 3, 100*time.Millisecond)
must(err)
time.Sleep(200 * time.Millisecond)
n, err := q.Reap(ctx)
must(err)
fmt.Printf("4. a worker claimed %d jobs and died; after the lease expired the reaper returned %d to the queue\n", len(claimed), n)
// Graceful shutdown: cancel while jobs are running and see that the running ones finish.
must(q.EnqueueBatch(ctx, []string{"slow", "slow", "slow", "slow", "email", "email"}, []any{1, 2, 3, 4, 5, 6}))
runCtx2, stop2 := context.WithCancel(ctx)
done2 := make(chan struct{})
go func() { w.Run(runCtx2); close(done2) }()
time.Sleep(300 * time.Millisecond)
stop2()
t0 := time.Now()
<-done2
counts, _ = q.Counts(ctx)
fmt.Printf("5. shutdown with four slow jobs running: Run returned after %v, running now %d, queued again %d\n", time.Since(t0).Round(50*time.Millisecond), counts["running"], counts["queued"])
}1. enqueued 500 jobs in one batch: map[queued:500]
2. four workers, claiming 10 at a time, finished in 970ms: map[dead:4 done:496]
handler calls in total: 522 (500 jobs plus the retries)
3. flaky jobs that succeeded on the second attempt: 10; broken jobs buried after three attempts: 4 (last error "always fails")
the slow job ran for 900ms on a 400ms lease and was attempted 1 time, because it renewed its lease
4. a worker claimed 3 jobs and died; after the lease expired the reaper returned 3 to the queue
5. shutdown with four slow jobs running: Run returned after 600ms, running now 0, queued again 8Throughput. 500 jobs of four kinds were enqueued in one batch. Four workers claiming 10 at a time cleared them in 970ms, ending with done:496 and dead:4. The handlers ran 522 times: 500 jobs plus the retries. The jobs here do almost no work, so the number measures the queue's overhead: about two milliseconds per job across four workers, with a claim, a status update and a notification path for each.
Retries. The flaky handler fails on its first attempt, and the ten such jobs all succeeded on the second, so each used exactly two attempts. The broken handler always fails, and the four such jobs ended dead after three attempts with the last error, always fails, stored in the row. A dead job is visible and queryable: that is the dead letter queue, and the place to look when something is wrong.
Leases. The slow job sleeps 900 milliseconds while the lease is 400. It was attempted once: the worker renewed the lease every 200 milliseconds while it ran, so the reaper, which runs every 200 milliseconds, never saw an expired lease for it.
A dead worker. The lab claims three jobs with a 100 millisecond lease and then does nothing with them, as a crashed process would. After 200 milliseconds the reaper returned all 3 to the queue. This is the at least once guarantee at work, and the reason handlers must be idempotent: a worker that is merely slow, not dead, can finish after its jobs were handed to another.
Shutdown. The lab enqueues four slow jobs and two quick ones, starts the workers, cancels the context 300 milliseconds in, and times how long Run takes. The first version of this lab finished every job in each claimed batch before stopping, so one worker holding four slow jobs took 3.35 seconds to exit. The code above checks the context before starting each claimed job and, if it is cancelled, calls Release to put the job back. Run then returned after about 600 milliseconds, the time for the one job already running to finish, with nothing left running and the unstarted jobs back in the queue. A batch claim makes this step necessary: the larger the batch, the more a shutdown has to give back.
Operating it
- Idempotent handlers. Delivery is at least once. A job may run twice after a lease expiry or a worker crash after the work but before
Done. Use a unique key for side effects, or record completion in the same transaction as the effect. - Pool sizing. Each worker holds a connection only during a statement, and the listener holds one. A pool of
workers + a fewis enough;ClaimandDoneare short. Do not hold a transaction open across a handler. - Lease length. Longer than the longest normal job's renewal interval, shorter than the time you can wait to recover from a crash. With renewal, a lease of ten to thirty seconds is typical.
- Vacuum. Every job is updated at least twice, so the table churns. Tune autovacuum for it, and delete or partition done jobs on a schedule; the partial indexes keep queued and running jobs fast regardless.
- Monitoring. Alert on the age of the oldest queued job, the count of dead jobs, and the reaper's activity. A rising reaper count means workers are dying or leases are too short.
- Priorities and queues. Add
queue textandpriority intcolumns, order the claim bypriority desc, run_at, id, and give each worker pool a list of queues, with a composite partial index to match. - Deduplication. A
unique_keycolumn with a unique index andon conflict do nothingstops the same logical job being queued twice.
Testing a queue
Test it the way the lab exercises it, against a real database, and cover the failure paths, since the happy path is the easy part. The cases worth a test each: a job that fails and is retried with the right delay; a job that fails past its maximum and ends up dead with the error recorded; two workers claiming from the same table never receive the same job (run a hundred jobs through eight workers and assert every id was handled exactly once, counting handler calls per id); a lease that expires returns the job to the queue; a worker that is cancelled mid run returns unstarted jobs; and a handler that panics does not take the worker down, which the code above does not yet guard against and a real worker should, by recovering in run and treating the panic as a failure. Add the race detector, since the interesting bugs are in the interplay of workers, the reaper and the listener.
Time based behaviour is the hard part of those tests. Keep delays short (milliseconds) in tests, make the intervals fields of the Workers struct rather than constants, and wait for conditions with a polling helper that fails after a deadline, never with a fixed sleep that is right on your machine and wrong on a loaded CI runner.
Scheduling and uniqueness
Two features that people add early deserve a note. Delayed and scheduled jobs need nothing new: run_at in the future makes a job wait, and the claim already filters on run_at <= now(). A recurring job is a handler that enqueues its next run, ideally with a unique key such as the job kind plus the scheduled time, so that two workers finishing at once cannot queue it twice. Uniqueness in general, "only one job of this kind for this user at a time", is a partial unique index on the columns that identify the job, restricted to the statuses queued and running, so a finished job does not block the next one: create unique index on jobs682 (kind, (payload->>'user')) where status in ('queued', 'running'). An insert that conflicts becomes an on conflict do nothing, and the caller learns from the row count that a job was already pending.
Visibility
A queue that cannot be inspected is a source of dread, and the table makes inspection easy. The queries you will run most: the count by status and kind; the oldest queued job and its age; the dead jobs with their errors; the jobs whose lease will expire soonest; and a histogram of attempts. Put the first three on a dashboard and alert on them. Add a started_at column and you can measure queue latency (claim time minus run_at) and run time per kind, which is the number that tells you whether you need more workers or faster handlers. Keep done jobs for a while, a day or a week, for debugging, then delete them in batches, or partition the table by day and drop old partitions, which is instant and leaves nothing for vacuum.
When to use a library instead
This is about 200 lines and covers most needs of a service with modest volume. Libraries that package the same design, with migrations, scheduling, a web interface, uniqueness and more, such as River, are worth evaluating before you maintain your own; they were not run for this article. Move to a dedicated broker when you need fan out to many consumers, rates beyond what a database should carry, or replay of history. Until then the queue next to your data, with transactional enqueue and plain SQL to inspect it, is hard to beat. The locking clause is documented in the PostgreSQL manual's page on SELECT locking, and the pgx features used here in the pgx reference.
