drover

README

drover

A PostgreSQL-backed task queue for Go, built from first principles — small enough to fully audit, honest about its delivery semantics, and documented down to every trade-off.

Status: pre-v0.1, under active development. The architecture is settled (ADRs), the roadmap is public (RFC-0001), and cycles ship as reviewable PRs.

Why another queue?

Because the reasoning is the product. Established Go queues are excellent — River for Postgres-native semantics, Asynq for observability — but their cores are large. Drover implements River-grade semantics (transactional enqueue, FOR UPDATE SKIP LOCKED claims, lease-based crash recovery, dead-letter retention) in a deliberately small codebase where every mechanism is explainable and provable, with each design decision recorded against the alternatives in docs/adr/ and grounded in published research.

Design at a glance

flowchart LR
    App[Your app] -->|"InsertTx(tx, args)"| PG[(PostgreSQL)]
    subgraph drover worker
        F[Fetcher<br/>SKIP LOCKED batch claim] --> CH[channel]
        CH --> W1[worker 1]
        CH --> W2[worker N]
        HB[Heartbeat<br/>extends leases] --> PG
        R[Rescuer<br/>requeues expired leases] --> PG
    end
    F --> PG
    W1 -->|complete / retry / dead| PG

• Transactional enqueue — jobs insert inside your own transaction; no ghost jobs, no outbox needed (ADR-0002).
• At-least-once, stated plainly — lease + heartbeat + rescuer; every duplicate source is named and bounded; handlers are idempotent by contract (ADR-0003).
• A real worker pool — a fixed, configurable number of goroutines claim and run jobs concurrently over a channel, and Stop drains in-flight work within a caller-supplied budget instead of leaving shutdown to lease expiry.
• Composable middleware — a func(Handler) Handler chain wraps every job, whatever its kind; the built-in Timeout bounds a handler's context and Logging reports each execution, and both compose with middleware you write yourself (ADR-0004).
• Scheduled, prioritized queues — InsertOpts delays a job to a future time and routes it to a named queue; queues are served from one shared worker pool by configurable weight, so a low-priority queue is slower, never starved (ADR-0003).
• Typed jobs — JobArgs + Worker[T] generics, no []byte payloads, no reflection.
• Stdlib-first — pgx + sqlc and the standard library; the dependency list stays short enough to read (ADR-0004).

Planned API (v0.1 target)

type SendEmail struct {
    To, Template string
}

func (SendEmail) Kind() string { return "send_email" }

// Worker
drover.Register(workers, &EmailWorker{}) // implements drover.Worker[SendEmail]

// Concurrency sizes the worker pool; it defaults to 10. A running job
// holds no connection, so it need not match the database pool's size.
// Queues share that one pool by weight; Middleware wraps every job,
// Logging installed outermost ahead of whatever you add.
client, err := drover.NewClient(pool, drover.Config{
    Workers:     workers,
    Concurrency: 8,
    Queues:      map[string]int{"default": 1, "bulk": 9},
    Middleware:  []drover.Middleware{drover.Timeout(30 * time.Second)},
})

// Enqueue atomically with your own writes
_, err = client.InsertTx(ctx, tx, SendEmail{To: user.Email, Template: "welcome"}, nil)

// Or delay a job and route it to a named queue; nil opts mean the
// "default" queue, runnable now.
_, err = client.Insert(ctx, SendEmail{To: user.Email, Template: "reminder"}, &drover.InsertOpts{
    Queue:       "bulk",
    ScheduledAt: time.Now().Add(24 * time.Hour),
})

// Start returns once the pool is running. Stop stops claiming, drains
// in-flight jobs within the given budget, and returns nil once every
// one of them has recorded its outcome — or an error naming how many
// did not finish and were returned to the queue instead.
err = client.Start(ctx)
err = client.Stop(shutdownCtx)

A full runnable version of this, including the retry path, a custom middleware, and shutdown on SIGINT, lives in examples/email.

Observability

Drover exposes Prometheus metrics on a dedicated ops port, separate from anything your application serves. Per-execution counters and histograms are recorded from the middleware chain; queue depth and oldest-job age are refreshed from the database on an interval, so scrape rate never multiplies database load (ADR-0005).

client, err := drover.NewClient(pool, drover.Config{
    Workers:         workers,
    Concurrency:     8,
    Queues:          map[string]int{"default": 1, "bulk": 9},
    OpsAddr:         "127.0.0.1:9090", // /metrics, /healthz, /readyz
    StatsInterval:   15 * time.Second, // how often depth/age gauges refresh
    MetricsRegistry: prometheus.NewRegistry(),
})

Leave OpsAddr empty to record metrics without serving them — pass MetricsRegistry to your own /metrics handler instead. Bind the ops port to a private interface; TLS and authentication are deployment concerns.

Ops endpoints

Path	Meaning
GET /metrics	Prometheus text format from this client's registry
GET /healthz	Process is alive — always 200
GET /readyz	Worker is started and the last gauge refresh succeeded within twice StatsInterval; otherwise 503 with a reason. Removes the worker from rotation on database loss without restarting the process.

Metrics

Name	Type	Labels	Meaning
drover_jobs_completed_total	Counter	queue	Executions that returned no error.
drover_jobs_failed_total	Counter	queue	Executions that returned an error, including recovered panics. Counts attempts, not jobs that reached dead — a job that fails four times and succeeds on the fifth increments this four times. Use drover_queue_depth{state="dead"} for permanent failures.
drover_job_duration_seconds	Histogram	queue	Wall-clock time one execution took, successful or not. Buckets span 5ms to 10 minutes.
drover_jobs_executing	Gauge	—	Executions currently inside the middleware chain.
drover_pool_concurrency	Gauge	—	Configured worker count; compare to drover_jobs_executing for saturation.
drover_queue_depth	Gauge	queue, state	Jobs held in one state on one queue. States: available, scheduled, retryable, running, dead. completed and cancelled are deliberately absent — those rows accumulate forever, and counting them would make this gauge's cost grow with history instead of backlog.
drover_oldest_job_age_seconds	Gauge	queue	Age in seconds of the oldest job claimable now (same predicate the fetch loop uses). 0 when no job is waiting — a configured-but-empty queue publishes zero, not a missing series.

Alerting

Recommended primary alert — a queue with work that is not moving:

max by (queue) (max_over_time(drover_oldest_job_age_seconds[5m])) > 300

drover_oldest_job_age_seconds is the primary alerting metric because it detects a stuck queue regardless of cause: workers dead, fetch broken, database slow, or handlers hanging. Counters tell you jobs failed; depth by state tells you where they are; oldest age tells you work is waiting longer than it should. Page on five minutes for most workloads; tune the threshold to your SLA.

Secondary signals: drover_queue_depth{state="dead"} for permanent failure accumulation; drover_jobs_executing / drover_pool_concurrency near 1.0 for saturation.

CLI

The drover binary is the operator surface for a live queue. Point it at Postgres with --database or $DATABASE_URL (the flag wins). The schema must already exist — the CLI does not run migrations. Add --json for machine-readable output on any command.

drover [--database URL] [--json] <command>

Command	What it does
stats	Per-queue depth and oldest-claimable age
jobs list	List jobs; optional --queue, --state, --limit (default 100, maximum 1000)
retry <id>	Redrive a dead job to available (attempt reset; prior errors kept)
cancel <id>	Cancel a waiting or dead job (running / terminal states are refused)
enqueue	Insert a job: --kind required; optional --queue, --args (JSON, default {})
version / --version	Print the embedded version (dev for local builds)

Library callers that need the same reads and operator writes without spawning the binary use drover.NewInspector(pool) — Stats, ListJobs, GetJob, Enqueue, CancelJob, and RetryJob on an Inspector. It is not a Client: there is no worker pool and no Start/Stop.

Releasing the binary

.goreleaser.yaml builds cmd/drover for linux, darwin, and windows on amd64 and arm64 with CGO_ENABLED=0, injects the tag via -X main.version={{.Version}}, and publishes archives plus checksums.txt (no Homebrew tap or Docker image). Cut a release from a version tag:

git tag v0.1.0
git push origin v0.1.0
goreleaser release --clean

Validate the config locally with goreleaser check. Snapshot builds without publishing: goreleaser release --snapshot --clean.

Roadmap

v0.1.0 = cycles A–F of RFC-0001: walking skeleton → retries/DLQ/rescue → worker pools + graceful shutdown → middleware + scheduled jobs → Prometheus observability → CLI + introspection. Then: benchmarks with published methodology, periodic jobs via advisory-lock leader election, and an optional server-rendered status page.

Documentation

• Architecture Decision Records — what was decided and why, including observability (ADR-0005)
• RFC-0001 roadmap — what ships when
• Research — the evidence behind the decisions (existing-system survey, storage mechanics, delivery semantics, conventions, scope)

License

MIT

Documentation

Overview

Package drover is a PostgreSQL-backed task queue: typed jobs are enqueued transactionally and executed by registered workers.

Delivery contract

Drover delivers each job at least once. A job whose worker crashes mid-execution is returned to the queue once its lease expires, so that job may run twice: duplicates are possible by design and Worker implementations must be idempotent. Exactly-once execution is not promised by drover or by any queue — see docs/adr/0003 in the repository for the full reasoning.

Enqueueing

Client.Insert persists a job in its own transaction. Client.InsertTx persists it inside a caller-owned pgx.Tx, so a domain write and its job commit or roll back atomically — no outbox required.

Failure and retries

A worker that returns an error, panics, or has no registered handler leaves its job queued for another attempt rather than killing it. Each attempt appends an entry to the job's errors array recording the attempt number, the time, the message, and a stack trace for panics. The job becomes dead only once it has used its max_attempts, and dead rows are retained for inspection rather than deleted.

Waits between attempts come from Config.RetryPolicy, which defaults to ExponentialRetryPolicy: attempt⁴ seconds, jittered by ±10% so that jobs failing against the same broken dependency do not retry in lockstep. Supply any RetryPolicy to replace that schedule.

A Worker can also decide its own outcome. Returning an error that wraps Cancel ends the job in the cancelled state without spending its remaining attempts, for work that can never succeed. Returning Snooze defers the job without consuming an attempt, for work that is not ready yet. Both are recognized through %w wrapping, so a worker may add its own context, and both are matchable with errors.Is against ErrCancelled and ErrSnoozed.

Crash recovery

Claiming a job takes a lease on it, and a running job's lease is renewed every Config.HeartbeatInterval for another Config.LeaseDuration. A worker that dies stops renewing, and a rescue sweep running every Config.RescueInterval returns any job whose lease has lapsed to the queue — or to the dead state if it has no attempts left. A rescue does not spend an attempt: the attempt that stranded the job was already spent.

The lease therefore bounds how long abandoned work sits idle after a SIGKILL or a lost node. Clean shutdown does not rely on it: Stop stops fetching and drains in-flight jobs first, keeping their leases alive until they finish.

A claim is held under a lease naming both the job and the attempt, and every state change is checked against it. A worker starved of heartbeats for longer than its lease — a long stop-the-world pause, a database outage — may find its job rescued and re-claimed elsewhere while it is still running; when it finishes, its write is refused rather than allowed to overwrite the attempt now in flight. The work may genuinely have run twice, which at-least-once delivery permits, but only the current holder records what happened.

Lease deadlines are computed by the database, not the client, so the lease means the same thing to every worker regardless of what their own clocks say.

Middleware

Config.Middleware wraps every job a Client runs, whatever its kind, as a chain of func(Handler) Handler — the same shape as an HTTP middleware chain, and composed the same way: index 0 is outermost, seeing a job first and its result last. The client always installs Logging outermost of everything, including a caller's own middleware, so per-job logging cannot be silently dropped by configuration; Timeout bounds how long a handler may run before its context is cancelled. Both are exported, ordinary middleware, so they double as the reference examples for writing another.

Queues and scheduling

InsertOpts, passed to Insert or InsertTx, chooses which named queue a job waits in and the earliest time it may run; nil, or a zero value, means the "default" queue, runnable as soon as a worker is free. Config.Queues maps every queue a Client works to a weight: the queues share one pool of Config.Concurrency workers rather than each getting a slice of it, and weight decides how often a queue is tried first in a fetch round, never whether it is tried at all, so no configured queue is starved.

Concurrency and shutdown

Jobs run on a fixed pool of Config.Concurrency goroutines, fed by a single fetch loop that claims no more rows than there are idle workers. The pool may safely be sized well above the database connection count: a job holds a connection only while it is claimed and while its outcome is recorded, not for the work in between.

Start(ctx) returns as soon as the pool is running rather than blocking for the client's lifetime. Calling it twice, or calling it again after Stop, returns ErrAlreadyStarted — a client's lifecycle runs once. Cancelling ctx begins the same shutdown Stop performs, with no deadline, so a client wired to a cancellable context still drains cleanly with nobody calling Stop.

Stop(ctx) stops claiming new work, then waits for everything already claimed to finish and record its outcome, returning nil once all of it has. ctx bounds that wait. When the budget runs out, Stop cancels the contexts running handlers were given, returns the jobs it could not finish to the queue so another worker can pick them up, and returns an error wrapping ErrDrainIncomplete naming how many there were. A non-nil Stop error is at-least-once delivery showing through at its most visible: those jobs are claimable again immediately — the requeue does not spend their attempt — and because Go offers no way to force a goroutine to stop, the handler this process abandoned may still be running, so the job can genuinely execute twice. Stop before Start returns ErrNotStarted; calling it again returns the first call's verdict without blocking.

Observability

Every client records Prometheus metrics on Config.MetricsRegistry, defaulting to a private registry per client. Per-execution counters and a duration histogram are recorded by a middleware installed inside Logging; queue depth and oldest-job age gauges are refreshed from the store on Config.StatsInterval (default fifteen seconds).

Config.OpsAddr binds a dedicated listener for GET /metrics, /healthz, and /readyz. An empty address records metrics without serving them. Until Start binds the address there is no listener, so probes see connection refused rather than an HTTP status. /healthz always returns 200; /readyz returns 503 while the client is draining or when the last gauge refresh is older than twice StatsInterval — typically because the database is unreachable.

drover_jobs_failed_total counts failed executions (attempts), not jobs that reached dead; use drover_queue_depth{state="dead"} for permanent failures. drover_queue_depth deliberately excludes completed and cancelled states. See the README observability section for every metric family, a compiling configuration snippet, and the recommended alerting expression over drover_oldest_job_age_seconds.

Example

Example wires the full path: migrate, register a typed worker, enqueue, work the queue on a pool, and shut down cleanly.

package main

import (
	"context"
	"log"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/augusto-dmh/drover"
)

type SendEmail struct {
	To       string `json:"to"`
	Template string `json:"template"`
}

func (SendEmail) Kind() string { return "send_email" }

type SendEmailWorker struct {
	drover.WorkerDefaults[SendEmail]
}

func (SendEmailWorker) Work(_ context.Context, job *drover.Job[SendEmail]) error {
	log.Printf("sending %s to %s", job.Args.Template, job.Args.To)
	return nil
}

func main() {
	ctx := context.Background()

	pool, err := pgxpool.New(ctx, "postgres://localhost:5432/app")
	if err != nil {
		log.Fatal(err)
	}
	if err := drover.Migrate(ctx, pool); err != nil {
		log.Fatal(err)
	}

	workers := drover.NewWorkers()
	drover.Register(workers, SendEmailWorker{})

	client, err := drover.NewClient(pool, drover.Config{
		Workers:     workers,
		Concurrency: 8,
		// Two named queues sharing the one pool above, "default" claimed
		// roughly four times as often as "bulk" — never exclusively, so
		// "bulk" is slower but not starved.
		Queues: map[string]int{"default": 4, "bulk": 1},
		// Timeout is one of the two built-in middleware; the client
		// always installs Logging outermost of whatever is configured
		// here, so job logging survives regardless.
		Middleware: []drover.Middleware{drover.Timeout(30 * time.Second)},
	})
	if err != nil {
		log.Fatal(err)
	}

	if _, err := client.Insert(ctx, SendEmail{To: "ada@example.com", Template: "welcome"}, nil); err != nil {
		log.Fatal(err)
	}

	// A nil *InsertOpts means the "default" queue, runnable now; here a
	// reminder is routed to "bulk" and held back for a day.
	reminder := SendEmail{To: "ada@example.com", Template: "reminder"}
	if _, err := client.Insert(ctx, reminder, &drover.InsertOpts{
		Queue:       "bulk",
		ScheduledAt: time.Now().Add(24 * time.Hour),
	}); err != nil {
		log.Fatal(err)
	}

	// Start returns as soon as the pool is running; the process stays
	// alive doing whatever else it does.
	if err := client.Start(ctx); err != nil {
		log.Fatal(err)
	}

	// On the way out, give the in-flight jobs a bounded chance to finish.
	// Whatever does not finish in time is returned to the queue, and Stop
	// says how much that was.
	shutdown, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	if err := client.Stop(shutdown); err != nil {
		log.Printf("drover: shutdown incomplete: %v", err)
	}
}

Index

• Variables
• func Cancel(reason error) error
• func Migrate(ctx context.Context, pool *pgxpool.Pool) error
• func Register[T JobArgs](ws *Workers, worker Worker[T])
• func Snooze(d time.Duration) error
• type Client
  • func NewClient(pool *pgxpool.Pool, cfg Config) (*Client, error)
  • func (c *Client) Insert(ctx context.Context, args JobArgs, opts *InsertOpts) (*JobRow, error)
  • func (c *Client) InsertTx(ctx context.Context, tx pgx.Tx, args JobArgs, opts *InsertOpts) (*JobRow, error)
  • func (c *Client) Start(ctx context.Context) error
  • func (c *Client) Stop(ctx context.Context) error
• type Config
• type ExponentialRetryPolicy
  • func (ExponentialRetryPolicy) NextRetry(_ context.Context, job *JobRow) time.Time
• type Handler
• type InsertOpts
• type Inspector
  • func NewInspector(pool *pgxpool.Pool) *Inspector
  • func (in *Inspector) CancelJob(ctx context.Context, id int64) (*JobRow, error)
  • func (in *Inspector) Enqueue(ctx context.Context, kind string, args json.RawMessage, opts *InsertOpts) (*JobRow, error)
  • func (in *Inspector) GetJob(ctx context.Context, id int64) (*JobRow, error)
  • func (in *Inspector) ListJobs(ctx context.Context, opts *ListJobsOpts) ([]*JobRow, error)
  • func (in *Inspector) RetryJob(ctx context.Context, id int64) (*JobRow, error)
  • func (in *Inspector) Stats(ctx context.Context) (*QueueStats, error)
• type Job
• type JobArgs
• type JobRow
• type JobState
• type ListJobsOpts
• type Middleware
  • func Logging(logger *slog.Logger) Middleware
  • func Timeout(d time.Duration) Middleware
• type QueueAge
• type QueueDepth
• type QueueStats
• type RetryPolicy
• type Worker
• type WorkerDefaults
• type Workers
  • func NewWorkers() *Workers

Examples

• Package
• Config (Observability)
• Middleware

Variables

var ErrAlreadyStarted = errors.New("drover: client already started")

ErrAlreadyStarted is returned by Start when the client is already running or has already been stopped. A client's lifecycle runs once.

var ErrCancelled = errors.New("drover: job cancelled by handler")

ErrCancelled is wrapped by every error Cancel returns. Workers do not return it directly; drover recognizes a cancellation with errors.Is, so a worker may add its own context with %w and still be understood.

var ErrDrainIncomplete = errors.New("drover: shutdown deadline reached")

ErrDrainIncomplete is wrapped by the error Stop returns when its budget ran out before every job finished. The jobs it names were returned to the queue and may run again elsewhere while the handlers this process abandoned are still running — the at-least-once contract showing through at the one moment it is most visible.

var ErrInvalidKind = errors.New("drover: job kind must be non-empty")

ErrInvalidKind is returned by Insert and InsertTx when args.Kind() returns an empty string.

var ErrInvalidTransition = errors.New("drover: invalid job state transition")

ErrInvalidTransition is returned by Inspector methods when a job is in a state that refuses the requested operator action.

var ErrNotFound = errors.New("drover: job not found")

ErrNotFound is returned by Inspector methods when the requested job id does not exist.

var ErrNotStarted = errors.New("drover: client not started")

ErrNotStarted is returned by Stop when the client was never started.

var ErrSnoozed = errors.New("drover: job snoozed by handler")

ErrSnoozed is wrapped by every error Snooze returns, so a caller can recognize a deferral with errors.Is the same way ErrCancelled identifies a cancellation. Prefer Snooze, which carries a duration; returning this sentinel on its own defers the job with no wait, so it is claimable again immediately.

Functions

func Cancel

func Cancel(reason error) error

Cancel reports that a job can never succeed. A worker returning it — bare, or wrapped in more context — ends the job in the cancelled state with reason recorded against the attempt, instead of spending the remaining attempts on work that is not going to start working.

func Migrate

func Migrate(ctx context.Context, pool *pgxpool.Pool) error

Migrate applies drover's embedded schema migrations to the database behind pool. It is idempotent: re-running it against an up-to-date schema is a no-op.

func Register

func Register[T JobArgs](ws *Workers, worker Worker[T])

Register adds worker as the handler for T's kind, taken from T's zero value. Registration happens at boot: registering the same kind twice is a programmer error and panics.

func Snooze

func Snooze(d time.Duration) error

Snooze defers a job by d without consuming an attempt. A worker returning it — bare, or wrapped in more context — leaves the job scheduled d from now, so a handler waiting on a precondition can keep asking without walking itself into the dead state. A zero or negative d makes the job claimable immediately rather than scheduling it in the past.

Types

type Client

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

Client enqueues jobs and runs the worker loop.

func NewClient

func NewClient(pool *pgxpool.Pool, cfg Config) (*Client, error)

NewClient returns a Client backed by the given PostgreSQL pool.

func (*Client) Insert

func (c *Client) Insert(ctx context.Context, args JobArgs, opts *InsertOpts) (*JobRow, error)

Insert enqueues a job in its own transaction. The job is available to workers as soon as Insert returns, unless opts delays it.

func (*Client) InsertTx

func (c *Client) InsertTx(ctx context.Context, tx pgx.Tx, args JobArgs, opts *InsertOpts) (*JobRow, error)

InsertTx enqueues a job inside the caller's transaction: the job exists if and only if tx commits, so a domain write and its job are atomic.

func (*Client) Start

func (c *Client) Start(ctx context.Context) error

Start begins working the queue and returns as soon as the pool is running — it does not block for the lifetime of the client. Jobs are executed by a fixed pool of Config.Concurrency goroutines fed by a single fetch loop, which claims only as many jobs as there are idle workers.

Shutdown is Stop's job. Cancelling ctx performs the same shutdown on its own, with no deadline, so a client wired to a cancellable context still drains cleanly without anyone calling Stop; what Stop adds is the ability to bound the wait and to learn what did not finish.

A client's lifecycle runs once. Calling Start on a client that is already running, or that has already been stopped, returns ErrAlreadyStarted.

A worker killed mid-job (crash, SIGKILL) leaves its job running until its lease expires and the rescuer returns it to the queue. That is the backstop for the paths no shutdown code survives; a clean shutdown loses nothing.

func (*Client) Stop

func (c *Client) Stop(ctx context.Context) error

Stop shuts the pool down and waits for it, in that order: it stops claiming new work, then waits for everything already claimed to finish and record its outcome. It returns nil when all of it did.

ctx bounds the wait. When that budget runs out, Stop cancels the contexts the running handlers were given, returns the jobs it could not finish to the queue so another worker can take them, and reports how many there were. Handlers that ignore cancellation keep running regardless — Go offers no way to stop a goroutine — which is why the answer is a count rather than a guarantee. Those jobs are at-least-once delivery working as documented: they were returned to the queue and may well run twice.

ctx bounds the waiting, not quite the whole call: handing the unfinished jobs back happens after that budget is spent, and each hand-back is allowed one HeartbeatInterval of its own. Budget the caller's side of a shutdown accordingly.

A ctx with no deadline waits as long as it takes. Stop before Start returns ErrNotStarted; calling it again returns the first call's verdict.

type Config

type Config struct {
	Workers      *Workers
	Logger       *slog.Logger
	PollInterval time.Duration
	RetryPolicy  RetryPolicy

	// Middleware wraps every job this client executes, whatever its
	// kind. The first element is outermost: it sees a job before the
	// others and its result after them.
	//
	// The chain is composed once, when the client is built, so appending
	// to the slice afterwards changes nothing about what the client runs.
	// A nil element is a programmer error and panics,
	// because the alternative is a middleware the caller believes is
	// running: a timeout that was silently dropped looks exactly like a
	// timeout that has not fired yet.
	Middleware []Middleware

	// Queues maps each queue this client works to its weight. An unset
	// or empty map means the single queue "default".
	//
	// Weight decides how often a queue is tried *first* in a fetch round,
	// not whether it is tried at all: every configured queue is visited
	// in every round, so no weighting can starve one. Giving "critical"
	// nine times the weight of "bulk" means critical jobs are usually
	// claimed before bulk ones — it does not mean bulk waits for critical
	// to empty.
	//
	// The queues share one pool of Concurrency workers rather than each
	// getting a slice of it, so a busy queue can use the whole pool while
	// the others are idle.
	//
	// A weight below one is corrected to one and reported, as is one
	// above the internal maximum — priority is a ratio, and an enormous
	// weight buys nothing a large one does not.
	//
	// An empty queue name panics: an empty Queue at enqueue time means
	// "default", so no caller could ever address it.
	//
	// Cost scales with the map. A round asks each queue in turn until the
	// idle workers are spoken for, so an idle client polls once per
	// configured queue per interval where a single-queue client polls
	// once. The queries are cheap — each is served by the fetch index and
	// one matching no rows does no write — but they are round trips, so
	// widen PollInterval as the map grows.
	Queues map[string]int

	// Concurrency is how many jobs this client executes at once. It
	// defaults to ten; a value of zero or less is treated as unset.
	//
	// It need not match the connection count of the pool the client was
	// built with. A running job holds no connection: drover takes one to
	// claim the job and one to record its outcome, and the handler's own
	// work happens in between, holding nothing.
	//
	// That is not the same as the pool size being irrelevant. Claims and
	// finalizations arrive in bursts — up to Concurrency workers can want
	// a connection at the same instant — and the heartbeat competes for
	// the same connections on a deadline of HeartbeatInterval. A renewal
	// that cannot get a connection in time lets a lease lapse, which is
	// how a slow database turns into a duplicated job. Leave the
	// connection pool some headroom above the fetch loop, the heartbeat
	// and the rescuer rather than sizing it to Concurrency exactly.
	//
	// The other cost of a high concurrency is handler-side: sockets,
	// memory, and load on whatever the handler talks to.
	Concurrency int

	// LeaseDuration is how long a claimed job may run before the rescuer
	// treats its worker as dead. It bounds how long work sits idle after
	// a crash, so a shorter lease recovers faster and a longer one
	// tolerates more heartbeat trouble.
	LeaseDuration time.Duration

	// HeartbeatInterval is how often a running job's lease is renewed.
	// It must be shorter than LeaseDuration or every job outliving one
	// lease would be rescued while still running; a value that is not is
	// replaced with a third of the lease.
	HeartbeatInterval time.Duration

	// RescueInterval is how often the client sweeps for jobs whose
	// workers died. Together with LeaseDuration it bounds how long
	// abandoned work sits idle. It defaults to LeaseDuration, so
	// shortening the lease to recover faster does not leave the sweep
	// running on the old, slower cadence.
	RescueInterval time.Duration

	// MetricsRegistry is where this client registers its metrics. Unset,
	// it gets a registry of its own.
	//
	// A private registry by default is what lets two clients exist in one
	// process: prometheus rejects the same collector twice, so a shared
	// registry would panic the second construction. For the same reason,
	// handing the *same* registry to two clients is a programmer error
	// and panics — give each its own, or give the second one nothing.
	//
	// Pass a registry to have drover's metrics gathered by an endpoint
	// you already serve; the client's own ops server is the alternative,
	// not a requirement.
	MetricsRegistry *prometheus.Registry

	// StatsInterval is how often this client refreshes its queue depth
	// and oldest-job-age gauges from the store. Zero (unset) takes the
	// default of fifteen seconds silently. A negative value is corrected
	// to the default and reported: the gauges would otherwise never move,
	// and an operator's alerts would go blind without a log line to
	// explain why.
	StatsInterval time.Duration

	// OpsAddr is the address the client binds for /metrics, /healthz, and
	// /readyz. Empty means no listener and no ops goroutine — the client
	// still records metrics on its registry; it just does not serve them.
	//
	// Bind failure fails Start and starts nothing: a worker that is
	// running but unreachable is the state this surface exists to remove.
	OpsAddr string
}

Config configures a Client. Zero values get defaults: slog.Default() for Logger, one second for PollInterval, an empty registry for Workers, ExponentialRetryPolicy for RetryPolicy, one minute for LeaseDuration, a third of the lease for HeartbeatInterval, the lease duration itself for RescueInterval, fifteen seconds for StatsInterval, and a registry of the client's own for MetricsRegistry.

Example (Observability)

ExampleConfig_observability mirrors the README observability snippet so a renamed or removed Config field fails `go test` instead of shipping a non-compiling example again.

package main

import (
	"time"

	"github.com/prometheus/client_golang/prometheus"

	"github.com/augusto-dmh/drover"
)

func main() {
	workers := drover.NewWorkers()
	_ = drover.Config{
		Workers:         workers,
		Concurrency:     8,
		Queues:          map[string]int{"default": 1, "bulk": 9},
		OpsAddr:         "127.0.0.1:9090", // /metrics, /healthz, /readyz
		StatsInterval:   15 * time.Second, // how often depth/age gauges refresh
		MetricsRegistry: prometheus.NewRegistry(),
	}
}

type ExponentialRetryPolicy

type ExponentialRetryPolicy struct{}

ExponentialRetryPolicy is the default schedule: the wait after attempt N is N⁴ seconds, jittered by ±10% so that jobs failing against the same broken dependency do not retry in lockstep. Attempt 1 waits about a second, attempt 10 about three hours, and attempt 25 about four and a half days.

func (ExponentialRetryPolicy) NextRetry

func (ExponentialRetryPolicy) NextRetry(_ context.Context, job *JobRow) time.Time

NextRetry implements RetryPolicy. It is safe for concurrent use: the only shared state is math/rand/v2's goroutine-safe global source.

type Handler

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

Handler executes one job and reports whether the attempt succeeded.

It is the unit a Middleware wraps, and it is also what Register builds from a typed Worker: by the time a Handler runs, the job's args are still raw JSON, because middleware is shared by every job kind and so cannot be generic over any one of them.

A Handler must not retain job beyond the call.

type InsertOpts

type InsertOpts struct {
	// Queue is which named queue the job waits in. Empty means
	// "default".
	//
	// A queue no running client is configured to work is not an error:
	// the process that enqueues a job is very often not the one that
	// runs it, so the job simply waits until something works that queue.
	Queue string

	// ScheduledAt is the earliest time the job may run. The zero value
	// means as soon as possible.
	//
	// It is a floor, not a promise: the job becomes claimable then, and
	// runs once a worker is free to take it. A time in the past is
	// treated as now rather than rejected, so a caller computing a delay
	// from a stale clock still enqueues a runnable job.
	ScheduledAt time.Time
}

InsertOpts are the per-job choices made at enqueue time. A nil *InsertOpts, or a zero value, means the defaults: the "default" queue, runnable immediately.

type Inspector

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

Inspector is the operator API over a drover queue store. It reads and mutates jobs without running workers: construct one with NewInspector and call methods directly — there is no Start/Stop lifecycle.

func NewInspector

func NewInspector(pool *pgxpool.Pool) *Inspector

NewInspector returns an Inspector backed by pool. It is ready for use without calling Start.

func (*Inspector) CancelJob

func (in *Inspector) CancelJob(ctx context.Context, id int64) (*JobRow, error)

CancelJob moves a waiting or dead job to cancelled. Running, completed, and already-cancelled jobs are refused with ErrInvalidTransition; a missing id is ErrNotFound.

func (*Inspector) Enqueue

func (in *Inspector) Enqueue(ctx context.Context, kind string, args json.RawMessage, opts *InsertOpts) (*JobRow, error)

Enqueue inserts a job with the given kind and raw JSON args. An empty kind or invalid JSON is refused without inserting. A nil opts uses the default queue and schedules the job immediately.

func (*Inspector) GetJob

func (in *Inspector) GetJob(ctx context.Context, id int64) (*JobRow, error)

GetJob returns the current row for id, or an error wrapping ErrNotFound when the id is unknown.

func (*Inspector) ListJobs

func (in *Inspector) ListJobs(ctx context.Context, opts *ListJobsOpts) ([]*JobRow, error)

ListJobs returns jobs matching the optional filters, newest id first, capped by Limit (default 100, maximum 1000).

func (*Inspector) RetryJob

func (in *Inspector) RetryJob(ctx context.Context, id int64) (*JobRow, error)

RetryJob redrives a dead job to available with attempt reset to 0, lease cleared, and prior errors retained. Any other state is ErrInvalidTransition; a missing id is ErrNotFound.

func (*Inspector) Stats

func (in *Inspector) Stats(ctx context.Context) (*QueueStats, error)

Stats returns per-queue depth counts for the published states and oldest-claimable ages, matching Driver.Stats semantics.

type Job

type Job[T JobArgs] struct {
	ID        int64
	Attempt   int
	CreatedAt time.Time
	Args      T
}

Job carries the decoded args payload plus row metadata to a Worker.

type JobArgs

type JobArgs interface {
	Kind() string
}

JobArgs is implemented by any type that can be enqueued as a job. Kind uniquely identifies the job type and links enqueued rows to the Worker registered for that kind. Args are serialized to JSON at enqueue time and decoded back before the Worker runs.

type JobRow

type JobRow struct {
	ID          int64
	Kind        string
	Queue       string
	Args        json.RawMessage
	State       JobState
	Attempt     int
	MaxAttempts int
	Errors      json.RawMessage
	ScheduledAt time.Time
	LeasedUntil *time.Time
	CreatedAt   time.Time
	FinalizedAt *time.Time
}

JobRow is the persisted representation of a job.

type JobState

type JobState string

JobState is the lifecycle state of a job row.

const (
	StateAvailable JobState = "available"
	StateScheduled JobState = "scheduled"
	StateRunning   JobState = "running"
	StateRetryable JobState = "retryable"
	StateCompleted JobState = "completed"
	StateCancelled JobState = "cancelled"
	StateDead      JobState = "dead"
)

All states exist in the schema from the start; Cycle A transitions only between Available, Running, Completed, and Dead. See AD-002.

type ListJobsOpts

type ListJobsOpts struct {
	Queue string
	State JobState
	Limit int
}

ListJobsOpts filters and bounds a ListJobs read. Empty Queue or State means no filter on that dimension. A Limit of zero or less defaults to 100. Limits above 1000 are refused.

type Middleware

type Middleware func(Handler) Handler

Middleware wraps a Handler in behaviour that applies to every job, whatever its kind — a timeout, a log record, a metric.

A Middleware is called once per job execution, with the next handler in the chain. It may inspect the job, derive a new context, run the next handler zero or more times, and return whatever it likes: the error the chain finally returns is the verdict on the attempt, so a middleware that swallows an error marks the job completed and one that returns an error without calling next fails the job without ever running its worker.

Example

ExampleMiddleware builds a chain by hand, the same way a Client builds Config.Middleware around its dispatch: the first middleware applied is outermost, so it sees the job first and its result last.

package main

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/augusto-dmh/drover"
)

func main() {
	var trace []string
	record := func(name string) drover.Middleware {
		return func(next drover.Handler) drover.Handler {
			return func(ctx context.Context, job *drover.JobRow) error {
				trace = append(trace, name+":start")
				err := next(ctx, job)
				trace = append(trace, name+":end")
				return err
			}
		}
	}

	base := func(ctx context.Context, job *drover.JobRow) error {
		trace = append(trace, "handler")
		return nil
	}

	// outer wraps inner wraps Timeout wraps base — outer runs first and
	// last, exactly as Config.Middleware's index 0 would.
	chain := record("outer")(drover.Timeout(time.Second)(record("inner")(base)))

	if err := chain(context.Background(), &drover.JobRow{ID: 1, Kind: "demo"}); err != nil {
		fmt.Println("error:", err)
	}
	fmt.Println(strings.Join(trace, " "))
}

Output:
outer:start inner:start handler inner:end outer:end

func Logging

func Logging(logger *slog.Logger) Middleware

Logging reports each job execution: one record when it starts and one when it ends, the latter carrying how long it took.

A failed execution is reported at WARN rather than ERROR. A handler returning an error is designed behaviour that the retry machinery expects and handles, so logging it at ERROR would fill an operator's error budget with jobs that are working exactly as intended. Whether anything is actually wrong is decided when the attempt is disposed of, and that is where the ERROR lives — on a job that has exhausted its attempts.

A nil logger falls back to slog.Default().

The client installs this itself, outermost, so job logging does not disappear the moment a caller configures a middleware of their own. It is exported because it is also the smallest complete example of writing one.

func Timeout

func Timeout(d time.Duration) Middleware

Timeout bounds how long a job's handler may run. When d elapses the context passed on to the rest of the chain is cancelled with context.DeadlineExceeded, which is the only way drover can ask a handler to stop: Go offers no way to halt a goroutine, so a handler that ignores its context runs to completion regardless.

The handler's own outcome is passed through untouched, including when it returns after its deadline expired. Substituting an error of the middleware's own would misreport a handler that ignored cancellation and genuinely succeeded — work that was really done, recorded as failed and then done again on the retry.

A d of zero or less applies no deadline, consistent with how every other duration drover takes treats a non-positive value.

The deadline covers the handler alone. The context under which the attempt's outcome is recorded is derived separately and is never cancelled, so a job cut off by its timeout still finalizes rather than sitting running until its lease lapses (AD-027).

type QueueAge

type QueueAge struct {
	Queue      string
	AgeSeconds float64
}

QueueAge is how long the oldest claimable job on one queue has been waiting, measured by the store's clock.

type QueueDepth

type QueueDepth struct {
	Queue string
	State string
	Count int64
}

QueueDepth is the number of jobs sitting in one state on one queue.

type QueueStats

type QueueStats struct {
	Depths []QueueDepth
	Oldest []QueueAge
}

QueueStats is one reading of what the queues hold: depths for the published states and oldest-claimable ages per queue.

type RetryPolicy

type RetryPolicy interface {
	NextRetry(ctx context.Context, job *JobRow) time.Time
}

RetryPolicy decides when a job that failed an attempt runs again. The whole row is in scope, so a policy may branch on kind, queue, attempt or the errors accumulated so far, and it answers with an absolute time — which expresses schedules ("retry at 03:00") that a plain duration cannot. An answer at or before the present means the job is claimable immediately.

Implementations must be safe for concurrent use: NextRetry is called from the worker loop and from the rescue sweep at the same time, and from every worker in the pool once pooled execution lands.

The context is the one the failing job was disposed of under, so a policy that consults a database or a remote configuration service can honour cancellation and deadlines. It is never nil.

type Worker

type Worker[T JobArgs] interface {
	Work(ctx context.Context, job *Job[T]) error
}

Worker executes jobs whose args decode to T. Implementations must be idempotent: drover delivers at least once, so the same job may run more than once after a crash or lease expiry.

type WorkerDefaults

type WorkerDefaults[T JobArgs] struct{}

WorkerDefaults is embedded by Worker implementations to keep them forward-compatible with optional Worker methods added in later versions.

type Workers

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

Workers maps job kinds to registered workers.

func NewWorkers

func NewWorkers() *Workers

NewWorkers returns an empty registry.