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 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) List(ctx context.Context, status string, limit int) ([]Job, 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) Start(ctx context.Context, workers int)
- func (q *Queue) Stats(ctx context.Context) (Stats, 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"`
}
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 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) List ¶
List returns up to limit jobs, newest first. status "" means all; "failed" means queued jobs awaiting retry; "queued" includes those too.
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) 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.
Source Files