Documentation
¶
Overview ¶
Package jobs is the DB-backed job queue: enqueue, workers with retry and exponential backoff, and a coarse cron scheduler.
Status lifecycle: queued → running → done, or on error back to queued with run_at = now + min(2^attempts, 300)s, and dead once attempts reaches max_attempts. A handler that returns an error after the worker's ctx was cancelled (shutdown) is requeued without counting an attempt. "failed" is a virtual status for Stats and List: queued jobs that already failed at least once (awaiting retry). attempts counts failed runs only.
Index ¶
- Variables
- func AttemptFromContext(ctx context.Context) (attempt, max int)
- func ContextWithAttempt(ctx context.Context, attempt, max int) context.Context
- func Migrations() fs.FS
- type Handler
- type Job
- type Option
- type ParkedCounts
- type Queue
- func (q *Queue) Count(ctx context.Context) (int, error)
- func (q *Queue) Cron(name string, every time.Duration, fn func(ctx context.Context) error)
- func (q *Queue) Enqueue(ctx context.Context, kind string, payload any, opts ...Option) (string, error)
- func (q *Queue) EnqueueTx(ctx context.Context, tx *sql.Tx, kind string, payload any, opts ...Option) (string, error)
- func (q *Queue) Get(ctx context.Context, id string) (Job, error)
- func (q *Queue) InFlight(ctx context.Context, key string) (string, bool, error)
- func (q *Queue) Kinds() []string
- func (q *Queue) List(ctx context.Context, status string, limit int) ([]Job, error)
- func (q *Queue) Park(ctx context.Context, kind, reason string) (int64, error)
- func (q *Queue) Parked(ctx context.Context) (ParkedCounts, error)
- func (q *Queue) Prune(ctx context.Context, now time.Time, r Retention) (int64, error)
- func (q *Queue) Register(kind string, h Handler, maxAttempts int)
- func (q *Queue) Retry(ctx context.Context, id string) error
- func (q *Queue) RunOne(ctx context.Context) (bool, error)
- func (q *Queue) Start(ctx context.Context, workers int)
- func (q *Queue) Stats(ctx context.Context) (Stats, error)
- func (q *Queue) Unpark(ctx context.Context, kind string) (int64, error)
- func (q *Queue) Unregistered(ctx context.Context) (ParkedCounts, error)
- func (q *Queue) Wait()
- type Retention
- type Stats
Constants ¶
This section is empty.
Variables ¶
var DefaultRetention = Retention{Done: 7 * 24 * time.Hour, Dead: 30 * 24 * time.Hour, Keyed: 90 * 24 * time.Hour}
DefaultRetention keeps done jobs 7 days, dead jobs 30 days and jobs with an idempotency key 90 days.
var ErrNotFound = errors.New("jobs: not found")
ErrNotFound is returned by Retry when no retryable job has the given id.
Functions ¶
func AttemptFromContext ¶
AttemptFromContext returns, inside a job handler, the 1-based number of the attempt being run and the job's max attempts (attempt == max is the final attempt). Outside a handler it returns 0, 0.
func ContextWithAttempt ¶
ContextWithAttempt returns ctx carrying attempt/max for AttemptFromContext. The worker sets it for every handler run; tests and code that invoke a handler directly can set it themselves.
func Migrations ¶
Migrations returns the jobs table migrations ("migrations/NNNN_x.sql").
Types ¶
type Handler ¶
type Handler func(ctx context.Context, payload json.RawMessage) error
Handler processes one job payload. A non-nil error schedules a retry.
type Job ¶
type Job struct {
ID string `json:"id"`
Kind string `json:"kind"`
Payload json.RawMessage `json:"payload"`
Status string `json:"status"`
RunAt string `json:"run_at"`
Attempts int `json:"attempts"`
MaxAttempts int `json:"max_attempts"`
LastError string `json:"last_error"`
LockedBy string `json:"locked_by"`
LockedAt string `json:"locked_at"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
IdempotencyKey string `json:"idempotency_key"`
ParkedReason string `json:"parked_reason" doc:"Non-empty while the job is held for a disabled or missing handler"`
}
Job is a row of the jobs table.
type Option ¶
type Option func(*enqueueOptions)
Option configures Enqueue.
func Idempotency ¶
Idempotency dedupes on key: enqueueing an existing key returns the existing job's id instead of inserting a new job.
type ParkedCounts ¶ added in v0.15.0
ParkedCounts reports queued jobs that are parked, by kind.
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue is the job queue. Register handlers and crons before Start.
func (*Queue) Cron ¶
Cron runs fn roughly every `every`, checked on the scheduler tick (30s), so intervals shorter than the tick are rounded up to it. The first tick after Start runs every cron. Register crons before Start.
func (*Queue) Enqueue ¶
func (q *Queue) Enqueue(ctx context.Context, kind string, payload any, opts ...Option) (string, error)
Enqueue stores a job and returns its id. payload is marshalled to JSON.
Enqueue writes on its own pool connection: never call it inside db.InTx (it would wait on the transaction's lock for busy_timeout and fail). Use EnqueueTx inside a transaction, or call Enqueue after commit.
func (*Queue) EnqueueTx ¶
func (q *Queue) EnqueueTx(ctx context.Context, tx *sql.Tx, kind string, payload any, opts ...Option) (string, error)
EnqueueTx is Enqueue inside tx: the job exists only if tx commits.
func (*Queue) InFlight ¶ added in v0.15.0
InFlight reports whether the job with idempotency key is still queued or running, returning its id when so. Done and dead jobs are not in flight.
func (*Queue) List ¶
List returns up to limit jobs, newest first. status "" means all; "failed" means queued jobs awaiting retry; "queued" includes those too.
func (*Queue) Park ¶ added in v0.15.0
Park marks every queued job of kind as parked with reason. Running, done and dead jobs are not touched. It returns how many rows changed.
func (*Queue) Parked ¶ added in v0.15.0
func (q *Queue) Parked(ctx context.Context) (ParkedCounts, error)
Parked counts parked queued jobs by kind.
func (*Queue) Prune ¶
Prune deletes done and dead jobs older than r (relative to now) and returns how many rows it removed. Queued, running and failed jobs are never touched.
func (*Queue) RunOne ¶ added in v0.15.0
RunOne claims and runs one due job on the caller's goroutine, reporting whether a job ran. It is for tests and one-shot tools that drive the queue without Start; a live worker process should not call it.
func (*Queue) Start ¶
Start requeues jobs abandoned by a previous process (see recoverStale), then launches workers goroutines and the cron scheduler, which repeats the recovery on every tick. It returns immediately; cancel ctx to stop and call Wait to block until every goroutine has exited.
func (*Queue) Unpark ¶ added in v0.15.0
Unpark releases parked jobs of kind back to the queue, due now, with their payload, attempts and idempotency key unchanged.
func (*Queue) Unregistered ¶ added in v0.15.0
func (q *Queue) Unregistered(ctx context.Context) (ParkedCounts, error)
Unregistered counts queued, unparked jobs whose kind has no handler in this process. They are never claimed; the count tells operators that a disabled or missing module left work waiting.
type Retention ¶
Retention says how long finished jobs are kept, by their last update (updated_at, the time they finished). Keyed applies to jobs with an idempotency key: they guard against duplicates (e.g. a webhook event id), so they are kept for the longer of Keyed and their status window.
type Stats ¶
type Stats struct {
Queued int `json:"queued"`
Running int `json:"running"`
Failed int `json:"failed"`
Dead int `json:"dead"`
Done int `json:"done"`
Parked int `json:"parked" doc:"Queued jobs held for a disabled or missing handler; counted in neither queued nor failed"`
}
Stats counts jobs by status. Queued excludes jobs awaiting retry, which are counted in Failed.
Source Files