Build a Job Queue on Postgres in Go With SKIP LOCKED

Most services eventually need to do something later: send an email, resize an image, call a slow API, retry a webhook. The usual answer is a queue, and the usual first reaction is to add a broker. For a service that already runs PostgreSQL, a table is often enough. It gives you transactions with your data (a job enqueued in the same transaction as the order it belongs to cannot exist without it), nothing new to operate, and plain SQL to inspect it. It stops being enough at high rates, and this article measures where it works. The mechanism that makes it possible is FOR UPDATE SKIP LOCKED, available since PostgreSQL 9.5.

We build the whole thing in one Go program, step by step, on Go 1.25.6 with pgx and PostgreSQL 18.6: enqueue, claim, complete, retry with delay, a dead state and a reaper for dead workers. It builds on locking with FOR UPDATE, uses the transaction rules from unit of work, and pairs with the outbox pattern, which is a queue of this kind. For a pgx specific version see a job queue with pgx.

What a Postgres queue promises

Be clear about the guarantees before writing it.

• At least once. A job can run more than once, if a worker dies after doing the work and before recording it. Jobs must be idempotent.
• Transactional enqueue. insert into jobs inside the caller's transaction means a rolled back request never leaves a job behind. That is the feature a broker cannot offer without extra machinery.
• Rough ordering. Jobs are claimed in run_at order, but with several workers they finish in any order.
• Thousands per second at the high end, not hundreds of thousands. Each job is updated at least twice, so the table churns. The measurements below are for a laptop; yours will differ.

The schema

create table jobs613 (
    id           bigserial primary key,
    payload      text not null,
    status       text not null default 'queued',
    attempts     int not null default 0,
    run_at       timestamptz not null default now(),
    locked_until timestamptz,
    done_at      timestamptz
);
create index jobs613_ready on jobs613 (run_at) where status = 'queued';

The status is queued, running, done or dead. run_at lets a job be delayed, which is also how retries back off. locked_until is a lease, explained below. The partial index covers only queued rows, so it stays small however many finished jobs the table holds; without it the claim query would scan the table.

The program

package main

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log"
	"sort"
	"sync"
	"time"

	_ "github.com/jackc/pgx/v5/stdlib"
)

const dsn = "postgres://postgres:[email protected]:55432/lab?sslmode=disable"

// Queue is a job queue in one table.
type Queue struct{ db *sql.DB }

type Job struct {
	ID      int64
	Payload string
	Attempt int
}

func (q Queue) Setup(ctx context.Context) error {
	for _, s := range []string{
		"drop table if exists jobs613",
		`create table jobs613 (
			id bigserial primary key,
			payload text not null,
			status text not null default 'queued',
			attempts int not null default 0,
			run_at timestamptz not null default now(),
			locked_until timestamptz,
			done_at timestamptz
		)`,
		"create index jobs613_ready on jobs613 (run_at) where status = 'queued'",
	} {
		if _, err := q.db.ExecContext(ctx, s); err != nil {
			return err
		}
	}
	return nil
}

func (q Queue) Enqueue(ctx context.Context, n int) error {
	_, err := q.db.ExecContext(ctx, "insert into jobs613 (payload) select 'job ' || i from generate_series(1, $1) i", n)
	return err
}

var ErrEmpty = errors.New("queue is empty")

// Claim takes one ready job and leases it for the given time. SKIP LOCKED keeps workers off each other's rows.
func (q Queue) Claim(ctx context.Context, lease time.Duration, skipLocked bool) (Job, error) {
	lock := "for update"
	if skipLocked {
		lock += " skip locked"
	}
	var j Job
	err := q.db.QueryRowContext(ctx, `
		update jobs613 set status = 'running', attempts = attempts + 1, locked_until = now() + $1::interval
		where id = (select id from jobs613 where status = 'queued' and run_at <= now() order by run_at, id limit 1 `+lock+`)
		returning id, payload, attempts`, fmt.Sprintf("%d milliseconds", lease.Milliseconds())).Scan(&j.ID, &j.Payload, &j.Attempt)
	if errors.Is(err, sql.ErrNoRows) {
		return Job{}, ErrEmpty
	}
	return j, err
}

func (q Queue) Done(ctx context.Context, id int64) error {
	_, err := q.db.ExecContext(ctx, "update jobs613 set status = 'done', done_at = now(), locked_until = null where id = $1", id)
	return err
}

// Retry puts a failed job back with a delay, or buries it after three attempts.
func (q Queue) Fail(ctx context.Context, j Job) error {
	if j.Attempt >= 3 {
		_, err := q.db.ExecContext(ctx, "update jobs613 set status = 'dead', locked_until = null where id = $1", j.ID)
		return err
	}
	_, err := q.db.ExecContext(ctx, "update jobs613 set status = 'queued', run_at = now() + interval '50 milliseconds', locked_until = null where id = $1", j.ID)
	return err
}

// Reap returns jobs whose lease ran out (a worker died) to the queue.
func (q Queue) Reap(ctx context.Context) (int64, error) {
	res, err := q.db.ExecContext(ctx, "update jobs613 set status = 'queued', locked_until = null where status = 'running' and locked_until < now()")
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

func (q Queue) Counts(ctx context.Context) map[string]int {
	out := map[string]int{}
	rows, _ := q.db.QueryContext(ctx, "select status, count(*) from jobs613 group by status")
	defer rows.Close()
	for rows.Next() {
		var s string
		var n int
		_ = rows.Scan(&s, &n)
		out[s] = n
	}
	return out
}

func format(m map[string]int) string {
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	s := ""
	for _, k := range keys {
		s += fmt.Sprintf("%s=%d ", k, m[k])
	}
	return s
}

// holdStyle is the other common design: take the row lock in a transaction and keep it while the job runs.
func holdStyle(ctx context.Context, db *sql.DB, q Queue, skip bool) (done, earlyQuits int, elapsed time.Duration) {
	_ = q.Setup(ctx)
	_ = q.Enqueue(ctx, 200)
	lock := "for update"
	if skip {
		lock += " skip locked"
	}
	var mu sync.Mutex
	var wg sync.WaitGroup
	start := time.Now()
	for w := 0; w < 8; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				tx, err := db.BeginTx(ctx, nil)
				if err != nil {
					log.Fatal(err)
				}
				var id int64
				err = tx.QueryRowContext(ctx, "select id from jobs613 where status = 'queued' order by id limit 1 "+lock).Scan(&id)
				if errors.Is(err, sql.ErrNoRows) {
					tx.Rollback()
					var left int
					_ = db.QueryRowContext(ctx, "select count(*) from jobs613 where status = 'queued'").Scan(&left)
					if left > 0 {
						mu.Lock()
						earlyQuits++
						mu.Unlock()
					}
					return
				}
				if err != nil {
					log.Fatal(err)
				}
				time.Sleep(10 * time.Millisecond) // the job runs while the row is locked
				if _, err := tx.ExecContext(ctx, "update jobs613 set status = 'done' where id = $1", id); err != nil {
					log.Fatal(err)
				}
				_ = tx.Commit()
				mu.Lock()
				done++
				mu.Unlock()
			}
		}()
	}
	wg.Wait()
	return done, earlyQuits, time.Since(start)
}

func main() {
	ctx := context.Background()
	db, err := sql.Open("pgx", dsn)
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()
	db.SetMaxOpenConns(20)
	q := Queue{db}

	// Run 1: 8 workers, 1000 jobs, every job takes 2ms. A job named "job 7" fails twice, then succeeds.
	// Run 2: the same with a job that always fails and a worker that dies holding a lease.
	for _, skip := range []bool{true, false} {
		if err := q.Setup(ctx); err != nil {
			log.Fatal(err)
		}
		if err := q.Enqueue(ctx, 1000); err != nil {
			log.Fatal(err)
		}
		var mu sync.Mutex
		processed := map[int64]int{}
		failures := map[string]int{}
		var wg sync.WaitGroup
		early := 0
		start := time.Now()
		for w := 0; w < 8; w++ {
			wg.Add(1)
			go func() {
				defer wg.Done()
				for {
					j, err := q.Claim(ctx, 5*time.Second, skip)
					if errors.Is(err, ErrEmpty) {
						var left int
						_ = db.QueryRowContext(ctx, "select count(*) from jobs613 where status = 'queued' and run_at <= now()").Scan(&left)
						if left > 10 {
							mu.Lock()
							early++
							mu.Unlock()
						}
						return
					}
					if err != nil {
						log.Fatal(err)
					}
					time.Sleep(2 * time.Millisecond)
					mu.Lock()
					processed[j.ID]++
					failed := j.Payload == "job 7" && j.Attempt < 3
					if failed {
						failures[j.Payload]++
					}
					mu.Unlock()
					if failed {
						_ = q.Fail(ctx, j)
						continue
					}
					_ = q.Done(ctx, j.ID)
				}
			}()
		}
		wg.Wait()
		// jobs that were retried are ready again after their delay, so drain once more
		time.Sleep(80 * time.Millisecond)
		for {
			j, err := q.Claim(ctx, 5*time.Second, skip)
			if errors.Is(err, ErrEmpty) {
				break
			}
			processed[j.ID]++
			_ = q.Done(ctx, j.ID)
		}
		elapsed := time.Since(start)
		twice := 0
		for _, n := range processed {
			if n > 1 {
				twice++
			}
		}
		fmt.Printf("skip locked=%-5v %s| distinct jobs run %d, run more than once %d (job 7 failed %d times), workers that quit with jobs left %d, %v\n",
			skip, format(q.Counts(ctx)), len(processed), twice, failures["job 7"], early, elapsed.Round(10*time.Millisecond))
	}

	// The lock-while-working design, 200 jobs of 10ms and 8 workers: ideal time is 250ms.
	for _, skip := range []bool{true, false} {
		done, early, elapsed := holdStyle(ctx, db, q, skip)
		var left int
		_ = db.QueryRowContext(ctx, "select count(*) from jobs613 where status = 'queued'").Scan(&left)
		fmt.Printf("hold the lock, skip locked=%-5v done %d, still queued %d, workers that quit with jobs left %d, %v\n",
			skip, done, left, early, elapsed.Round(10*time.Millisecond))
	}

	// A worker that claims a job and dies: the lease expires and the reaper returns it to the queue.
	_ = q.Setup(ctx)
	_ = q.Enqueue(ctx, 3)
	j, _ := q.Claim(ctx, 100*time.Millisecond, true)
	fmt.Printf("worker claimed job %d, then died. counts: %s\n", j.ID, format(q.Counts(ctx)))
	n, _ := q.Reap(ctx)
	fmt.Printf("reap right away returned %d jobs\n", n)
	time.Sleep(150 * time.Millisecond)
	n, _ = q.Reap(ctx)
	j2, _ := q.Claim(ctx, 5*time.Second, true)
	fmt.Printf("reap after the lease ran out returned %d job; the next claim got job %d on attempt %d\n", n, j2.ID, j2.Attempt)
}

skip locked=true  done=1000 | distinct jobs run 1000, run more than once 1 (job 7 failed 2 times), workers that quit with jobs left 0, 1.3s
skip locked=false done=1000 | distinct jobs run 1000, run more than once 1 (job 7 failed 2 times), workers that quit with jobs left 0, 930ms
hold the lock, skip locked=true  done 200, still queued 0, workers that quit with jobs left 2, 360ms
hold the lock, skip locked=false done 200, still queued 0, workers that quit with jobs left 0, 2.63s
worker claimed job 1, then died. counts: queued=2 running=1 
reap right away returned 0 jobs
reap after the lease ran out returned 1 job; the next claim got job 1 on attempt 2

Queue has six methods, each one statement.

Enqueue inserts rows. In a real service it takes the caller's transaction, so the job commits with the data it belongs to.

Claim is the heart. One statement finds the next ready row, marks it running, bumps the attempt count, sets the lease and returns the row:

update jobs613 set status = 'running', attempts = attempts + 1, locked_until = now() + $1::interval
where id = (select id from jobs613 where status = 'queued' and run_at <= now()
            order by run_at, id limit 1 for update skip locked)
returning id, payload, attempts

The subquery picks one row and locks it, skipping any that another worker has locked. The outer update changes it. Because it is a single statement in autocommit mode, the lock lasts only for the statement, and the row is then running, which takes it out of every other worker's view. If no row is ready, returning produces nothing and database/sql reports sql.ErrNoRows, which Claim turns into ErrEmpty.

Done sets done and clears the lease. Fail puts the job back with run_at fifty milliseconds ahead, a stand-in for exponential backoff, or marks it dead after three attempts so a person can look at it. Reap returns jobs whose lease expired to the queue. Counts groups by status, which is also your first monitoring query.

A thousand jobs, eight workers

The first two output lines run 1000 jobs, each taking 2 milliseconds, on eight workers. One job, "job 7", is built to fail on its first two attempts and succeed on the third.

Both runs finished with done=1000, with 1000 distinct jobs run and exactly one run more than once, which was job 7 retried: it failed 2 times, as designed. No job was claimed by two workers at the same moment. That is the property that matters, and the lease style claim provides it with or without skip locked, because the single update statement sets the status before the next worker's subquery runs.

Look at the times: 1.3s with skip locked and 930ms without. SKIP LOCKED did not make this faster; it was slower in this run, and I did not investigate why. Probably workers scanning past rows that others had locked cost more than the short wait they avoided. The honest reading is that when the claim is one quick statement, the row lock lasts for microseconds, and the clause adds no throughput. It still has a role: it stops a worker from ever waiting behind another, and removes the case where a waiting worker rechecks a changed row and returns nothing, a behaviour described in the locking article.

The other design: lock while working

The second pair of lines tests the style found in most tutorials. A worker begins a transaction, selects a ready row with for update, does the job while the transaction is open, marks it done and commits. There is no lease because the row lock itself is the claim. If the worker dies, the connection drops and PostgreSQL releases the lock, so the row is ready again. The cost is a transaction open for the length of the job.

Here the clause decides everything. 200 jobs of 10 milliseconds on eight workers, ideal time 250ms:

• With skip locked: 360ms, all 200 done.
• Without it: 2.63s, all 200 done. Every worker picked the same first row and queued behind its lock, so the work ran nearly one at a time: 200 jobs times 10ms is two seconds, and the rest is overhead.

The "workers that quit with jobs left 2" in the first line is not lost work. At the very end, a worker found nothing unlocked while peers still held the last rows in their open transactions, so it exited. Those rows were then completed by their holders, and the final count is done 200, still queued 0.

Which design to use:

For anything longer than a few hundred milliseconds, use the lease. A long open transaction holds back vacuum and uses a connection that the rest of the service needs; see connection pool settings.

Dead workers and the reaper

The last three lines show the lease at work. A worker claims job 1 with a 100 millisecond lease and then "dies": it never calls Done. The counts show queued=2 running=1. A Reap straight away returned 0 jobs, because the lease had not expired. After 150 milliseconds the reaper returned 1, and the next claim got job 1 on attempt 2.

Set the lease longer than any job should take, and have long jobs extend it with a heartbeat update. A lease that is too short causes duplicate runs while the first worker is still alive, and one that is too long delays recovery. Run the reaper on a timer in every worker, which is safe because the statement is idempotent. This is also why jobs must be idempotent: a slow worker whose lease expired can finish after the job has been given to another.

Running it for real

A few details separate the demo from a service.

1. Poll with backoff and wake on notification. An idle worker that polls every 100ms adds an average 50ms of latency and a steady trickle of queries. A NOTIFY after enqueue wakes workers immediately, with the poll kept as a safety net, since notifications can be lost; see LISTEN and NOTIFY from Go.
2. Batch the claim when jobs are tiny. limit 10 with returning takes ten at once and cuts round trips.
3. Vacuum. Each job is updated at least twice, leaving dead rows. Tune autovacuum for the table (a lower scale factor, such as autovacuum_vacuum_scale_factor = 0.01) and delete or archive done rows on a schedule. A queue table that is never cleaned slows its own claim query.
4. Monitor the count by status and the age of the oldest queued job. A growing age is the signal that workers are down or too few.
5. Put a context deadline on every job and on every claim, so a stuck worker releases its connection; see query timeouts.
6. Cap attempts and keep dead jobs visible, with the last error stored in a column.

Extending it

The core above is about forty lines of SQL. Most real needs are a column or a clause away.

Enqueue in the caller's transaction. Make Enqueue take a DBTX, the two method interface from unit of work, so the job is inserted with the data it belongs to. If the request rolls back, the job never existed. This is the single biggest reason to use a database queue, and it is easy to lose by giving the queue its own connection.

Deduplicate with a key. Add a nullable unique_key column with a unique index and insert with on conflict (unique_key) do nothing. "Send the welcome email for user 42" can then be enqueued twice and exist once. Use it for jobs that are triggered by events that can repeat.

Priorities and several queues. A queue text column and a priority int column, with the claim ordering by priority desc, run_at, id and filtering by queue, give you named queues for free. Add the columns to the partial index, and give each worker pool the list of queues it serves, so slow jobs cannot starve quick ones.

Rate limits and concurrency per key. Some jobs must not run in parallel for the same customer. A concurrency_key column plus a not exists (select 1 from jobs where status = 'running' and concurrency_key = j.concurrency_key) clause in the claim does it, at the cost of a more expensive query. For a handful of keys, an advisory lock on the key inside the worker is simpler; see advisory locks.

Graceful shutdown. On a termination signal, stop claiming, let running jobs finish within a deadline, and exit. A job still running at the deadline keeps its lease and is picked up by another worker after it expires. Passing a context derived from the signal into Claim makes the worker stop asking at once, and a separate context with a timeout bounds the drain.

Backoff. The lab uses a fixed fifty milliseconds. In practice use an exponential delay with jitter, such as run_at = now() + (interval '5 seconds' * power(2, attempts)), capped at an hour, and store last_error so the dead jobs explain themselves.

Test it with real workers. The lab is also the shape of the test: start several workers against the real database, enqueue a known number of jobs, and assert on the final counts and on how many times each job ran. Add a case that kills a worker mid job and one that makes a job fail repeatedly. A queue test with a mock database proves nothing about locking, so run it against PostgreSQL, as described in testing against PostgreSQL.

When to use a real queue instead

A table queue is the right size when jobs number in the thousands per second or fewer, when they must commit with your data, and when you would rather not run another system. Move to a broker or a purpose built library when you need fan out to many consumers, very high rates, long retention with replay, or delivery across services you do not control. Go libraries such as River package this same pattern, with migrations, retries and a dashboard, and are worth evaluating before you extend the code here; they were not run for this article. The feature is documented in the PostgreSQL manual under the locking clause, which describes SKIP LOCKED and warns that it returns an inconsistent view, acceptable for a queue and not for general queries. The tuning of autovacuum for a table with this churn is covered in the manual's chapter on routine vacuuming.