Documentation
¶
Overview ¶
Package jobs is the agent's durable work queue: a port for work that must survive a restart and be retried until it succeeds or is exhausted. Scheduled automations, delayed and retried side effects, and fan-out work are enqueued here rather than run inline, so a crash between "decide to do X" and "X done" loses nothing.
Like state, spine, and observe, the queue is a PORT with a zero-dependency default and heavy backends as opt-in adapters. The default is SQLite-backed (durable, single file, no setup); a fast in-memory queue is the reference for tests and ephemeral use. River (Postgres) and NATS/JetStream (cross-process, fleet, k8s) attach later as adapters held to the same jobstest conformance suite. Workers claim jobs under a lease, so a crashed worker's in-flight jobs return to the queue when the lease expires rather than being lost.
The queue is distinct from its neighbours: the bus is at-most-once ambient signalling, the spine is the durable ordered truth log. The queue is durable, at-least-once, retried work that the bus and spine are not.
Index ¶
- Constants
- Variables
- func Backoff(attempt int, base, ceiling int64) int64
- func ClaimDefaults(p ClaimParams) (queue string, limit int)
- func Claimable(j Job, now int64) bool
- func ExpiredExhausted(j Job, now int64) bool
- func MarkClaimed(j *Job, now, leaseFor int64)
- func MarkDone(j *Job, now int64)
- func MarkFailed(j *Job, errMsg string, retryAt, now int64)
- func MarkRecovered(j *Job, now int64)
- func MarkTimedOut(j *Job, now int64)
- func Notify(ch chan<- struct{})
- type ClaimParams
- type EnqueueParams
- type Job
- type MemoryQueue
- func (q *MemoryQueue) Claim(_ context.Context, p ClaimParams) ([]Job, error)
- func (q *MemoryQueue) Close() error
- func (q *MemoryQueue) Complete(_ context.Context, id string) error
- func (q *MemoryQueue) Enqueue(_ context.Context, p EnqueueParams) (Job, error)
- func (q *MemoryQueue) Fail(_ context.Context, id, errMsg string, retryAt int64) error
- func (q *MemoryQueue) Get(_ context.Context, id string) (Job, error)
- func (q *MemoryQueue) Ready() <-chan struct{}
- func (q *MemoryQueue) Recover(context.Context) (int, error)
- type Option
- type Queue
- type State
- type Waker
Constants ¶
const DefaultMaxAttempts = 25
DefaultMaxAttempts is the retry ceiling applied when EnqueueParams.MaxAttempts is unset. The job runs once and is retried up to this many times in total.
const DefaultQueue = "default"
DefaultQueue is the lane a job lands in when EnqueueParams.Queue is empty.
Variables ¶
var ( // ErrNotFound is returned by Get/Complete/Fail for an unknown job ID. ErrNotFound = fault.New(fault.Terminal, "job_not_found", "job not found") // ErrNotRunning is returned by Complete/Fail when the job is not currently // leased (already done, dead, or never claimed). ErrNotRunning = fault.New(fault.Terminal, "job_not_running", "job is not running") // ErrInvalidJob is returned by Enqueue for params that cannot describe a job // (an empty Kind). ErrInvalidJob = fault.New(fault.Terminal, "job_invalid", "invalid job") )
Sentinel errors, fault-classified so callers branch on the class rather than on string matching.
Functions ¶
func Backoff ¶
Backoff returns the delay before the attempt-th retry (1-based), doubling from base and capped at ceiling. It is the default exponential policy a worker applies when computing Fail's retryAt; callers may substitute their own. attempt <= 0 is treated as 1.
func ClaimDefaults ¶
func ClaimDefaults(p ClaimParams) (queue string, limit int)
ClaimDefaults resolves a ClaimParams to a concrete queue and limit.
func Claimable ¶
Claimable reports whether a job may be claimed at time now: a pending job whose RunAt has arrived, or a running job whose lease has expired and still has attempts left (crash recovery). Each claim is an attempt, so a job whose lease expires after its final attempt is not reclaimed (it is timed out, see ExpiredExhausted) rather than retried past MaxAttempts.
func ExpiredExhausted ¶
ExpiredExhausted reports whether a running job has timed out on its final attempt: its lease has expired and no attempts remain. Such a job is reaped to dead by Claim rather than left as a zombie or retried forever.
func MarkClaimed ¶
MarkClaimed transitions a job into a fresh lease: running, one more attempt, and a lease ending leaseFor nanos from now.
func MarkFailed ¶
MarkFailed transitions a running job after a failed attempt: back to pending (RunAt = retryAt) while attempts remain, or dead once they are exhausted.
func MarkRecovered ¶
MarkRecovered expires a running job's lease so the next Claim reclaims it at once, leaving the attempt count untouched. Startup crash recovery uses it: under the single-instance lock a job still running belongs to a dead predecessor, so its lease need not be waited out. A job whose attempts are already spent is not revived; the next Claim reaps it to dead (see ExpiredExhausted).
func MarkTimedOut ¶
MarkTimedOut transitions a job that exhausted its attempts by lease expiry to dead, recording why. Used by Claim's reap step.
func Notify ¶
func Notify(ch chan<- struct{})
Notify sends the coalesced, non-blocking ready signal a Waker delivers: if the channel's buffered slot is free the signal is queued, otherwise one is already pending and this one merges into it. Shared so every backend coalesces identically and none can block a writer on a slow worker.
Types ¶
type ClaimParams ¶
type ClaimParams struct {
Queue string // empty means DefaultQueue
Limit int // max jobs to claim; <= 0 means 1
LeaseFor int64 // lease duration in nanos; the claim is held this long
}
ClaimParams describes a worker's request to lease ready jobs.
type EnqueueParams ¶
type EnqueueParams struct {
Queue string
Kind string
Payload []byte
Scope state.Scope
MaxAttempts int // <= 0 uses DefaultMaxAttempts
RunAt int64 // unix nanos; 0 means as soon as possible
}
EnqueueParams describes a job to enqueue. Only Kind is required; the queue fills in safe defaults (DefaultQueue, MaxAttempts, RunAt = now).
type Job ¶
type Job struct {
ID string
Queue string
Kind string // selects the handler; the queue itself never interprets it
Payload []byte
Scope state.Scope
State State
Attempt int // attempts started so far (0 until first claim)
MaxAttempts int // attempts allowed before the job goes dead
LastError string // fault class + message from the most recent failure
RunAt int64 // unix nanos: earliest time the job may be claimed
LeaseExpires int64 // unix nanos: when a running job's claim lapses (0 if not running)
OriginInstanceID string // instance that enqueued the job, for fleet attribution
CreatedAt int64
UpdatedAt int64
}
Job is one unit of durable work. Payload is opaque bytes so the queue is agnostic to encoding; the typed-handler layer marshals it.
func BuildJob ¶
func BuildJob(p EnqueueParams, now int64, id, instanceID string) Job
BuildJob constructs the job an Enqueue stores, applying defaults (DefaultQueue, DefaultMaxAttempts, RunAt = now). The caller validates that p.Kind is non-empty and supplies the assigned id, now (unix nanos), and origin instance.
type MemoryQueue ¶
type MemoryQueue struct {
// contains filtered or unexported fields
}
MemoryQueue is the in-process reference Queue: a map guarded by a mutex, no persistence. It is the fast default for tests and ephemeral runs, and the reference semantics the durable SQLite backend must match (both run jobstest.RunSuite). Lease expiry is honoured, so it exercises the same crash-recovery path as a durable backend, just without surviving a restart.
func NewMemory ¶
func NewMemory(opts ...Option) *MemoryQueue
NewMemory constructs an in-process Queue ready to use with zero configuration.
func (*MemoryQueue) Claim ¶
func (q *MemoryQueue) Claim(_ context.Context, p ClaimParams) ([]Job, error)
Claim implements Queue.
func (*MemoryQueue) Complete ¶
func (q *MemoryQueue) Complete(_ context.Context, id string) error
Complete implements Queue.
func (*MemoryQueue) Enqueue ¶
func (q *MemoryQueue) Enqueue(_ context.Context, p EnqueueParams) (Job, error)
Enqueue implements Queue.
func (*MemoryQueue) Ready ¶
func (q *MemoryQueue) Ready() <-chan struct{}
Ready implements Waker: the channel signalled after Enqueue and after a Fail that returns a job to pending.
func (*MemoryQueue) Recover ¶
func (q *MemoryQueue) Recover(context.Context) (int, error)
Recover implements Queue: it expires the lease of every job left running, so the next Claim reclaims (or, if attempts are spent, reaps) it at once. Only live jobs are scanned; terminal jobs already left the active index.
type Option ¶
type Option func(*MemoryQueue)
Option configures a MemoryQueue.
func WithClock ¶
WithClock sets the time source (default: clock.System). Tests and deterministic replay pass a clock.Manual.
func WithIDGenerator ¶
WithIDGenerator sets the job-ID source (default: a generator on the queue's clock). A seeded generator makes enqueued IDs reproducible.
func WithInstanceID ¶
WithInstanceID sets the instance stamped onto enqueued jobs (default "local").
type Queue ¶
type Queue interface {
// Enqueue persists a new job and returns it with its assigned ID and defaults
// applied.
Enqueue(ctx context.Context, p EnqueueParams) (Job, error)
// Claim leases up to Limit ready jobs from a queue: jobs that are pending with
// RunAt in the past, or running jobs whose lease has expired (a crashed
// worker's work). Claimed jobs move to StateRunning with their Attempt
// incremented and a fresh lease, and are returned to the caller.
Claim(ctx context.Context, p ClaimParams) ([]Job, error)
// Complete marks a claimed job done. It is an error to complete a job that is
// not running.
Complete(ctx context.Context, id string) error
// Fail records a failed attempt. If attempts remain the job returns to pending
// with RunAt = retryAt (the caller computes backoff); otherwise it goes dead. A
// negative retryAt fails the job permanently (dead now, no further attempts),
// for a cause the caller knows is not retryable.
Fail(ctx context.Context, id, errMsg string, retryAt int64) error
// Get returns a job by ID.
Get(ctx context.Context, id string) (Job, error)
// Recover makes jobs left running by a crashed worker immediately claimable,
// returning how many were reset. Startup calls it once under the single-instance
// lock, where no live worker holds a lease, so a run interrupted mid-step is
// re-dispatched at once instead of waiting out its lease (up to several minutes). It
// does not change a job's attempt count; a job that already exhausted its attempts
// is reaped to dead by the next Claim rather than retried.
Recover(ctx context.Context) (int, error)
// Close releases the queue's resources.
Close() error
}
Queue is the durable work-queue port. Implementations must be safe for concurrent use and must claim each ready job to at most one worker at a time.
type State ¶
type State string
State is a job's lifecycle position.
const ( // StatePending is ready to run at or after RunAt, awaiting a worker. StatePending State = "pending" // StateRunning is claimed by a worker under an unexpired lease. StateRunning State = "running" // StateDone is completed successfully; terminal. StateDone State = "done" // StateDead is failed and out of attempts; terminal, retained for inspection. StateDead State = "dead" )
type Waker ¶
type Waker interface {
// Ready returns the channel signalled after a job is enqueued or returned to
// pending for retry. Every call returns the same channel.
Ready() <-chan struct{}
}
Waker is an optional Queue capability: a queue that can observe its own in-process writes exposes a ready channel, so a worker wakes the moment work is enqueued instead of discovering it on its next idle poll. The signal is advisory and coalesced (one buffered slot): receiving one does not guarantee a claimable job, and a missed one costs nothing because the worker's poll remains the fallback. The poll is also the only path for scheduled RunAt arrivals and for writes from other processes, which a queue cannot signal.
Source Files
Directories