pgqueue

package module
v1.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 16, 2026 License: MIT Imports: 11 Imported by: 0

README

pgqueue

Go Reference GitHub go.mod Go version Go Report Card GitHub Workflow Status (with branch) GitHub GitHub code size in bytes

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
    JobTimeout: 2 * time.Minute, // bound each handler; zero leaves handlers unbounded
    Retain:     7 * 24 * time.Hour, // prune completed/dead jobs older than a week; zero never prunes
    Middleware: []pgqueue.Middleware{ /* metrics, tracing — first entry is outermost */ },
})

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: ErrAlreadyStarted, or
        // the LISTEN connection could not be established before ctx was canceled.
    }
}()

...

// 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 could not be established before ctx was canceled, or Start was
    // called twice; once listening, Start returns nil when 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. A rescued job that has already used all of its attempts (one that crashed the process every time, say) is dead-lettered instead of re-queued.
  • Terminal writes are fenced on the claimed attempt: if a slow handler outlives RescueAfter and the job is rescued and re-run, the stale run's result is dropped rather than overwriting the newer run's. Set RescueAfter comfortably above your longest handler runtime so this stays a safety net, not the norm.
Retention

Finished jobs stay in pgqueue_jobs for inspection until pruned. Set WorkerConfig.Retain to have the worker delete its queue's completed and dead jobs older than that during its maintenance sweep, or call Queue.Prune(ctx, olderThan) on your own schedule. Nothing is pruned by default.

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.

Logging

pgqueue.Logger is the four-method subset (Debug/Info/Warn/Error with key-value pairs) that *slog.Logger already satisfies, so pass slog.Default() or your own structured logger; the default is silent.

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 (or dead-letters them once their attempts are used up), which makes delivery at-least-once: handlers must tolerate being invoked more than once for the same job. Terminal writes are fenced on the claimed attempt, so a stale run can never overwrite the outcome of the run that superseded it.

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

Examples

Constants

View Source 
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.

View Source 
const SchemaDown = `DROP TABLE IF EXISTS pgqueue_jobs;`

SchemaDown drops everything Schema creates.

Variables

View Source 
var ErrAlreadyStarted = errors.New("pgqueue: already started")

ErrAlreadyStarted is returned by Worker.Start and Subscriber.Start when the receiver has already been started.

Functions

func DefaultBackoff 

func DefaultBackoff(attempt int) time.Duration

DefaultBackoff doubles from 2s (attempt 1) to a ceiling of 256s (attempt 8 and beyond), plus up to 25% jitter — so the longest possible delay is 320s.

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.

Example 
package main

import (
	"context"
	"database/sql"
	"log"
	"log/slog"

	"github.com/pascalallen/pgqueue"
)

func main() {
	const dsn = "postgres://localhost/app?sslmode=disable"
	db, err := sql.Open("postgres", dsn)
	if err != nil {
		log.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	sub := pgqueue.NewSubscriber(dsn, slog.Default())
	sub.Handle("orders_updated", func(payload []byte) {
		// Runs on the subscriber goroutine: keep it fast or hand off.
		log.Printf("order update: %s", payload)
	})
	go func() {
		if err := sub.Start(ctx); err != nil {
			log.Println("subscriber stopped:", err)
		}
	}()

	// Every process currently LISTENing on the channel receives this; 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 {
		log.Fatal(err)
	}
}

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
	CreatedAt   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 — *slog.Logger satisfies it as-is — so that host applications can pass their existing logger without an adapter.

type Middleware added in v1.1.0 

type Middleware func(next Handler) Handler

Middleware wraps a Handler — for metrics, tracing, logging, or per-job setup — without the library taking on those dependencies itself.

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 {}).

Example 
package main

import (
	"context"
	"database/sql"
	"log"
	"log/slog"
	"time"

	"github.com/pascalallen/pgqueue"
)

func main() {
	db, err := sql.Open("postgres", "postgres://localhost/app?sslmode=disable")
	if err != nil {
		log.Fatal(err)
	}
	ctx := context.Background()

	// pgqueue.Schema creates the pgqueue_jobs table; copy it into your own
	// migrations. Executing it directly is fine for a quick start.
	if _, err := db.ExecContext(ctx, pgqueue.Schema); err != nil {
		log.Fatal(err)
	}

	q := pgqueue.New(db, pgqueue.WithLogger(slog.Default()))

	id, err := q.Enqueue(ctx, "send_welcome_email", []byte(`{"user_id":"01J5..."}`),
		pgqueue.WithMaxAttempts(3),
		pgqueue.WithRunAt(time.Now().Add(time.Minute)),
	)
	if err != nil {
		log.Fatal(err)
	}
	log.Println("enqueued job", id)
}

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.

Example 
package main

import (
	"context"
	"database/sql"
	"log"

	"github.com/pascalallen/pgqueue"
)

func main() {
	db, err := sql.Open("postgres", "postgres://localhost/app?sslmode=disable")
	if err != nil {
		log.Fatal(err)
	}
	ctx := context.Background()
	q := pgqueue.New(db)

	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		log.Fatal(err)
	}
	defer tx.Rollback()

	// ... business writes on tx ...

	// The job row and its wakeup NOTIFY become visible only if tx commits.
	if _, err := q.EnqueueTx(ctx, tx, "send_welcome_email", []byte(`{"user_id":"01J5..."}`)); err != nil {
		log.Fatal(err)
	}
	if err := tx.Commit(); err != nil {
		log.Fatal(err)
	}
}

func (*Queue) Prune added in v1.1.0 

func (q *Queue) Prune(ctx context.Context, olderThan time.Duration) (int64, error)

Prune deletes this queue's completed and dead jobs whose last update is older than olderThan and reports how many rows went. Call it from your own schedule, or set WorkerConfig.Retain to have the worker do it.

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. It panics if called 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.

Example 
package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"log"
	"log/slog"
	"time"

	"github.com/pascalallen/pgqueue"
)

func main() {
	const dsn = "postgres://localhost/app?sslmode=disable"
	db, err := sql.Open("postgres", dsn)
	if err != nil {
		log.Fatal(err)
	}

	w := pgqueue.NewWorker(db, pgqueue.WorkerConfig{
		Concurrency: 4,
		ListenDSN:   dsn, // dedicated LISTEN connection; empty means poll-only
		Logger:      slog.Default(),
	})

	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: dead-letter now.
			return pgqueue.Permanent(err)
		}
		// A plain error schedules a retry with backoff; nil completes the job.
		return sendWelcomeEmail(ctx, p.UserID)
	})

	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		if err := w.Start(ctx); err != nil {
			log.Println("worker stopped:", err)
		}
	}()

	// ... on shutdown: stop claiming, then wait for in-flight jobs to drain.
	cancel()
	stopCtx, stopCancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer stopCancel()
	if err := w.Stop(stopCtx); err != nil {
		log.Println("worker did not drain in time:", err)
	}
}

func sendWelcomeEmail(context.Context, string) error { return nil }

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. It panics if called after Start: the handler map is read without locking by running jobs.

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 (JobTimeout, when set). Default 5m.
	RescueAfter time.Duration
	// JobTimeout bounds each handler invocation: the handler's context is
	// canceled once it elapses and the attempt fails with the context error.
	// Zero (the default) leaves handlers unbounded, so a hung handler blocks
	// drain forever — set it unless every handler already bounds itself.
	JobTimeout time.Duration
	// Retain, when set, makes the worker prune this queue's completed and
	// dead jobs whose last update is older than Retain (see Queue.Prune)
	// during its maintenance sweep. Zero (the default) never prunes.
	Retain time.Duration
	// Middleware wraps every registered handler; the first entry is the
	// outermost. Applied at Register time, so it must be set before then.
	Middleware []Middleware
	Logger     Logger
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.