jobq

jobq

A Postgres-backed background job queue for Go, built step by step to learn Go concurrency and distributed-systems design. Its only dependency is pgx.

• At-least-once delivery, with handlers retried using exponential backoff and jitter, then discarded after MaxAttempts.
• Safe under concurrency: many workers and processes share a queue, and each claim takes a different job (FOR UPDATE SKIP LOCKED).
• Crash-safe: leases, heartbeats, and a rescuer recover jobs from workers that die; attempt-number fencing stops stalled workers from overwriting results.
• Graceful shutdown, LISTEN/NOTIFY for millisecond pickup, leader election, periodic jobs, unique jobs, and transactional enqueue.
• About 10,000 jobs/second on a single CPU core (see DESIGN.md).

Delivery guarantee: at-least-once. A job can run more than once, for example if a worker crashes after finishing a job but before recording it. Handlers must be idempotent.

Usage

pool, _ := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
store.Migrate(ctx, pool) // safe to call on every startup
client, _ := jobq.NewClient(pool)

// Enqueue, from anywhere in your app.
client.Enqueue(ctx, "send_email", EmailArgs{To: "user@example.com"}, nil)

// Or as part of your own transaction: the job exists only if it commits.
client.EnqueueTx(ctx, tx, "send_email", EmailArgs{To: user.Email}, nil)

// Run a worker.
worker := jobq.NewWorker(client, &jobq.WorkerConfig{Concurrency: 20})
jobq.Handle(worker, "send_email", func(ctx context.Context, job *jobq.Job, args EmailArgs) error {
	if args.To == "" {
		return jobq.Permanent(errors.New("no recipient")) // don't retry
	}
	return mailer.Send(ctx, args.To) // an error means: retry with backoff
})
worker.AddPeriodic(jobq.PeriodicJob{Name: "nightly-report", Every: 24 * time.Hour, Kind: "report", Args: ReportArgs{}})

ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
worker.Run(ctx) // returns after a graceful shutdown

cmd/demo is a complete runnable example. Options per job are in EnqueueOpts (queue, priority, scheduled time, max attempts, unique key); worker settings are in WorkerConfig, and every zero value means a sensible default.

Operating it

jobqctl stats              # per queue: ready, scheduled, running, done, and lag
jobqctl list -state discarded
jobqctl show 42            # a job and its full error history
jobqctl retry 42           # run a finished job again
jobqctl cancel 43          # discard a job that hasn't started

The leader worker deletes completed jobs after 24 hours and discarded jobs after 7 days (configurable). Logs go through log/slog.

Metrics. jobq doesn't depend on a metrics library; export Worker.Stats() and Client.QueueStats() with whatever you use. With Prometheus:

lag := prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: "jobq_queue_lag_seconds"}, []string{"queue"})
go func() {
	for range time.Tick(15 * time.Second) {
		stats, err := client.QueueStats(ctx)
		if err != nil {
			continue
		}
		for _, q := range stats {
			lag.WithLabelValues(q.Queue).Set(q.Lag.Seconds()) // alert when this keeps growing
		}
	}
}()

Development

Requires Go 1.25+ and Docker.

make db-up       # start Postgres
make check       # go vet + tests with the race detector (what CI runs)
make bench       # load test against a scratch database
make psql        # poke around the database

Each test gets its own throwaway database (see internal/testdb), so tests run in parallel. The chaos tests start real worker processes and kill -9 them; go test -short skips the slow ones.

Layout

client.go, enqueue.go      Client, Enqueue, EnqueueTx, unique jobs
worker.go                  Worker: fetch loop, backpressure, shutdown, logging
claim.go, complete.go      claiming jobs and recording outcomes (with fencing)
completer.go               batched completions
lease.go, leader.go        heartbeats, rescuer, leader election, cleaner
notify.go, periodic.go     LISTEN/NOTIFY, periodic jobs
handler.go, errors.go      typed handlers, Permanent errors
backoff.go, admin.go       retry backoff; inspection and admin API
store/                     embedded migrations
cmd/jobqctl, cmd/demo, cmd/jobqbench
DESIGN.md                  every decision, why it was made, and what the tests showed

Roadmap

• 0. Setup: repo, Postgres, migrations, CI
• 1. Simplest loop: enqueue, fetch, run, complete
• 2. Concurrency: executor pool, backpressure, typed handlers
• 3. Failure handling: retries, backoff, discard
• 4. Crash safety: leases, heartbeats, fencing, rescuer
• 5. Graceful shutdown
• 6. Multiple processes + LISTEN/NOTIFY
• 7. Leader election, periodic jobs, unique jobs
• 8. Transactional enqueue
• 9. Logging, stats, admin CLI, cleanup
• 10. Load test, profiling, write-up