pgqueue

README

pgqueue

pgqueue is a Postgres-backed durable job queue and NOTIFY/LISTEN pub/sub module for Go, built on database/sql and github.com/lib/pq — jobs are claimed with FOR UPDATE SKIP LOCKED, failures retry with exponential backoff until they land in a dead-letter state, orphaned jobs are rescued after a crash, and delivery is at-least-once.

Installation

Use the Go CLI tool go to install pgqueue.

go get github.com/pascalallen/pgqueue

Usage

Apply the schema and enqueue jobs:

...

import (
    "github.com/pascalallen/pgqueue"
)

...

// pgqueue.Schema creates the pgqueue_jobs table and its indexes. Copy it
// into your application's migrations; for a quick start, execute it directly.
if _, err := db.ExecContext(ctx, pgqueue.Schema); err != nil {
    // Schema not applied; nothing below will work.
}

q := pgqueue.New(db)

id, err := q.Enqueue(ctx, "send_welcome_email", []byte(`{"user_id":"01J5..."}`))
if err != nil {
    // The job row was not inserted; nothing will run.
}

// EnqueueTx makes the job part of a business transaction: the job row and
// its wakeup NOTIFY are only visible/delivered if tx commits.
id, err = q.EnqueueTx(ctx, tx, "send_welcome_email", payload)
if err != nil {
    // Roll back tx; the job rides the transaction.
}

...

Run a worker:

...

w := pgqueue.NewWorker(db, pgqueue.WorkerConfig{
    ListenDSN: dsn, // dedicated LISTEN connection for prompt wakeups; empty means poll-only
})

w.Register("send_welcome_email", func(ctx context.Context, job pgqueue.Job) error {
    var p struct {
        UserID string `json:"user_id"`
    }
    if err := json.Unmarshal(job.Payload, &p); err != nil {
        // Retrying an unparseable payload is pointless: go straight to 'dead'.
        return pgqueue.Permanent(err)
    }

    // A plain error schedules a retry with backoff; nil completes the job.
    return sendWelcomeEmail(ctx, p.UserID)
})

go func() {
    if err := w.Start(ctx); err != nil {
        // Start only fails before the claim loop begins (already started, LISTEN error).
    }
}()

...

// Graceful shutdown: cancel Start's context to stop claiming, then wait for
// in-flight jobs to drain.
cancel()
if err := w.Stop(shutdownCtx); err != nil {
    // The worker did not drain before the deadline.
}

...

Fire-and-forget pub/sub over NOTIFY/LISTEN:

...

// Publish sends to every process currently LISTENing on the channel. Pass a
// *sql.Tx instead of db to defer delivery until commit.
if err := pgqueue.Publish(ctx, db, "orders_updated", []byte(`{"order_id":42}`)); err != nil {
    // The NOTIFY was not issued.
}

s := pgqueue.NewSubscriber(dsn, nil)
s.Handle("orders_updated", func(payload []byte) {
    // Callbacks run sequentially on the subscriber's goroutine — keep them
    // fast or hand off internally.
})

if err := s.Start(ctx); err != nil {
    // LISTEN failed; Start returns nil once ctx is canceled.
}

...

At-least-once delivery

• Workers claim batches with FOR UPDATE SKIP LOCKED, so any number of processes can consume the same queue without double-processing a claimed job.
• A failed job retries with exponential backoff (default: doubling from 2s to a ceiling of ~4m16s, plus up to 25% jitter) until MaxAttempts (default 5), then remains in the table with status = 'dead' for inspection. Wrap an error with pgqueue.Permanent to skip retries and dead-letter immediately.
• A rescue sweep re-queues jobs stuck in running longer than RescueAfter (default 5m) — orphans left by a crashed worker. This makes delivery at-least-once: handlers must tolerate being invoked more than once for the same job.

Schema ownership

pgqueue never runs migrations. The pgqueue.Schema constant creates the pgqueue_jobs table and its indexes, and pgqueue.SchemaDown drops everything it creates — copy them into your application's own migration files. If a release ever changes Schema, add a follow-up migration in the host application.

Pub/sub is not durable

Publish/Subscriber wrap Postgres NOTIFY/LISTEN, which is fire-and-forget: a subscriber that is disconnected (or mid-reconnect) when a notification fires never sees it, and Postgres caps NOTIFY payloads at ~8000 bytes. Use it as a real-time signal; when you need durability, use the job queue.

Testing

Run the test suite with the race detector and coverage:

go test -race -cover ./...

Integration tests need a Postgres database and are skipped when PGQUEUE_TEST_DSN is unset. To run them against a scratch Postgres:

docker run -d --rm --name pg -e POSTGRES_HOST_AUTH_METHOD=trust -p 5544:5432 postgres:16-alpine
PGQUEUE_TEST_DSN="host=localhost port=5544 user=postgres dbname=postgres sslmode=disable" go test -race -cover ./...

Create and view a coverage profile:

go test -covermode=count -coverprofile=coverage.out
go tool cover -html=coverage.out

Contributing

Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.

Please make sure to update tests as appropriate.

License

MIT

Documentation

Overview

Package pgqueue is a small Postgres-backed durable job queue and pub/sub layer built on database/sql and lib/pq.

Jobs are rows in the pgqueue_jobs table (see Schema). Producers enqueue with Queue.Enqueue or, to make a job part of a business transaction, Queue.EnqueueTx — the wakeup NOTIFY rides the caller's transaction, so workers are only woken for jobs that actually committed. Workers claim batches with FOR UPDATE SKIP LOCKED, so any number of processes can consume the same queue without double-processing. Failed jobs retry with backoff until MaxAttempts, then remain in the table with status 'dead' for inspection. A rescue sweep re-queues jobs orphaned by a crashed worker, which makes delivery at-least-once: handlers must tolerate being invoked more than once for the same job.

Publish and Subscriber wrap Postgres NOTIFY/LISTEN for fire-and-forget fan-out to every listening process. Notifications carry no durability: a subscriber that is disconnected when one fires never sees it.

Dependency policy: this package imports only the standard library and github.com/lib/pq, so it stays trivial to vendor or embed in any database/sql-based application.

Index

• Constants
• func DefaultBackoff(attempt int) time.Duration
• func IsPermanent(err error) bool
• func Permanent(err error) error
• func Publish(ctx context.Context, db DBTX, channel string, payload []byte) error
• type DBTX
• type EnqueueOption
  • func WithMaxAttempts(n int) EnqueueOption
  • func WithRunAt(t time.Time) EnqueueOption
• type Handler
• type Job
• type Logger
• type Option
  • func WithDefaultMaxAttempts(n int) Option
  • func WithLogger(l Logger) Option
  • func WithQueueName(name string) Option
• type Queue
  • func New(db *sql.DB, opts ...Option) *Queue
  • func (q *Queue) Enqueue(ctx context.Context, jobType string, payload []byte, opts ...EnqueueOption) (int64, error)
  • func (q *Queue) EnqueueTx(ctx context.Context, tx *sql.Tx, jobType string, payload []byte, ...) (int64, error)
• type Status
• type Subscriber
  • func NewSubscriber(dsn string, logger Logger) *Subscriber
  • func (s *Subscriber) Handle(channel string, fn func(payload []byte))
  • func (s *Subscriber) Start(ctx context.Context) error
• type Worker
  • func NewWorker(db *sql.DB, cfg WorkerConfig) *Worker
  • func (w *Worker) Register(jobType string, h Handler)
  • func (w *Worker) Start(ctx context.Context) error
  • func (w *Worker) Stop(ctx context.Context) error
• type WorkerConfig

Constants

const Schema = `` /* 889-byte string literal not displayed */

Schema creates the pgqueue_jobs table and its indexes. Hosts that manage their own migrations should copy this SQL into a migration; if a release ever changes Schema, add a follow-up migration in the host application.

const SchemaDown = `DROP TABLE IF EXISTS pgqueue_jobs;`

SchemaDown drops everything Schema creates.

Functions

func DefaultBackoff

func DefaultBackoff(attempt int) time.Duration

DefaultBackoff doubles from 2s up to a 5m cap, plus up to 25% jitter.

func IsPermanent

func IsPermanent(err error) bool

IsPermanent reports whether err is (or wraps) a Permanent error.

func Permanent

func Permanent(err error) error

Permanent wraps an error to tell the worker that retrying is pointless (e.g. an unparseable payload): the job goes straight to 'dead'.

func Publish

func Publish(ctx context.Context, db DBTX, channel string, payload []byte) error

Publish sends a fire-and-forget notification to every process currently LISTENing on channel. db may be a *sql.DB or, to defer delivery until commit, a *sql.Tx. Postgres caps NOTIFY payloads at ~8000 bytes.

Types

type DBTX

type DBTX interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}

DBTX is the subset of database/sql operations this package needs. It is satisfied by both *sql.DB and *sql.Tx.

type EnqueueOption

type EnqueueOption func(*enqueueSettings)

func WithMaxAttempts

func WithMaxAttempts(n int) EnqueueOption

func WithRunAt

func WithRunAt(t time.Time) EnqueueOption

WithRunAt schedules the job for a future time instead of immediately.

type Handler

type Handler func(ctx context.Context, job Job) error

Handler executes a claimed job. A nil return completes the job; an error schedules a retry (or moves the job to 'dead' once MaxAttempts is reached, immediately if the error is wrapped with Permanent). Delivery is at-least-once: handlers must tolerate re-execution of the same job.

type Job

type Job struct {
	ID          int64
	Queue       string
	Type        string
	Payload     json.RawMessage
	Attempts    int
	MaxAttempts int
	RunAt       time.Time
	LastError   string
}

Job is a claimed job as handed to a Handler. Attempts includes the attempt currently executing (it is 1 on the first delivery).

type Logger

type Logger interface {
	Debug(msg string, keyVals ...any)
	Info(msg string, keyVals ...any)
	Warn(msg string, keyVals ...any)
	Error(msg string, keyVals ...any)
}

Logger is the minimal structured-logging port used by this package. It is deliberately a subset of common structured logger interfaces so that host applications can pass their existing logger without an adapter.

type Option

type Option func(*Queue)

func WithDefaultMaxAttempts

func WithDefaultMaxAttempts(n int) Option

WithDefaultMaxAttempts changes the per-queue default (5) applied to jobs enqueued without WithMaxAttempts.

func WithLogger

func WithLogger(l Logger) Option

func WithQueueName

func WithQueueName(name string) Option

WithQueueName routes jobs to a queue other than "default".

type Queue

type Queue struct {
	// contains filtered or unexported fields
}

Queue enqueues jobs onto a single named queue.

func New

func New(db *sql.DB, opts ...Option) *Queue

func (*Queue) Enqueue

func (q *Queue) Enqueue(ctx context.Context, jobType string, payload []byte, opts ...EnqueueOption) (int64, error)

Enqueue inserts a job and notifies workers. payload must be valid JSON (nil is stored as {}).

func (*Queue) EnqueueTx

func (q *Queue) EnqueueTx(ctx context.Context, tx *sql.Tx, jobType string, payload []byte, opts ...EnqueueOption) (int64, error)

EnqueueTx is Enqueue inside the caller's transaction: the job row and its wakeup NOTIFY are only visible/delivered if the transaction commits.

type Status

type Status string

Status is the lifecycle state of a job row.

const (
	StatusPending   Status = "pending"
	StatusRunning   Status = "running"
	StatusCompleted Status = "completed"
	StatusDead      Status = "dead"
)

type Subscriber

type Subscriber struct {
	// contains filtered or unexported fields
}

Subscriber delivers NOTIFY payloads to registered callbacks over a single dedicated LISTEN connection that reconnects automatically.

func NewSubscriber

func NewSubscriber(dsn string, logger Logger) *Subscriber

NewSubscriber takes a lib/pq connection string for the dedicated LISTEN connection.

func (*Subscriber) Handle

func (s *Subscriber) Handle(channel string, fn func(payload []byte))

Handle registers a callback for one channel. Not safe to call after Start. Callbacks run sequentially on the subscriber's goroutine — keep them fast or hand off internally.

func (*Subscriber) Start

func (s *Subscriber) Start(ctx context.Context) error

Start listens until ctx is canceled. Notifications that fire while the connection is re-establishing are lost — Subscriber is a real-time signal, not a durable stream.

type Worker

type Worker struct {
	// contains filtered or unexported fields
}

Worker claims and executes jobs from one queue. Register handlers, then run Start in a goroutine; cancel its context to begin a graceful drain and use Stop to wait for the drain to finish.

func NewWorker

func NewWorker(db *sql.DB, cfg WorkerConfig) *Worker

func (*Worker) Register

func (w *Worker) Register(jobType string, h Handler)

Register maps a job type to its handler. Not safe to call after Start.

func (*Worker) Start

func (w *Worker) Start(ctx context.Context) error

Start runs the claim loop until ctx is canceled, then waits for in-flight jobs to finish before returning. Handlers receive a context that survives the cancellation so drains complete cleanly.

func (*Worker) Stop

func (w *Worker) Stop(ctx context.Context) error

Stop blocks until Start has fully drained and returned, or until ctx expires. It does not itself trigger the shutdown — cancel Start's context for that.

type WorkerConfig

type WorkerConfig struct {
	// QueueName selects which queue to consume. Default "default".
	QueueName string
	// Concurrency is the number of jobs executed simultaneously. Default 2.
	Concurrency int
	// PollInterval is the safety-net poll cadence; LISTEN provides prompt
	// wakeups when ListenDSN is set. Default 5s.
	PollInterval time.Duration
	// ListenDSN is a lib/pq connection string for the dedicated LISTEN
	// connection that wakes the worker as soon as a job is enqueued. Empty
	// means poll-only.
	ListenDSN string
	// Backoff maps the just-failed attempt number (1-based) to the delay
	// before the next attempt. Default DefaultBackoff.
	Backoff func(attempt int) time.Duration
	// RescueAfter re-queues jobs stuck in 'running' longer than this —
	// orphans left by a crashed process. It must comfortably exceed the
	// longest handler runtime. Default 5m.
	RescueAfter time.Duration
	Logger      Logger
}