Documentation
¶
Overview ¶
Package worker provides the job execution engine — a Runner that orchestrates a single job attempt through middleware and an exec.Executor, and a Pool that manages concurrent worker goroutines polling for jobs.
Index ¶
- Constants
- type Executordeprecated
- type Pool
- type PoolOption
- func WithDefaultLeaseTTL(d time.Duration) PoolOption
- func WithHeartbeatInterval(d time.Duration) PoolOption
- func WithMaxPollInterval(d time.Duration) PoolOption
- func WithPollInterval(d time.Duration) PoolOption
- func WithPoolConcurrency(n int) PoolOption
- func WithPoolQueues(queues []string) PoolOption
- func WithQueueManager(m QueueManager) PoolOption
- func WithReapInterval(d time.Duration) PoolOption
- func WithResourceManager(m resource.Manager) PoolOption
- func WithStaleJobThreshold(d time.Duration) PoolOption
- func WithStoreCallTimeout(d time.Duration) PoolOption
- func WithTelemetry(t *telemetry.Telemetry) PoolOption
- func WithWorkerCustomKeys(keys []string) PoolOption
- type QueueManager
- type Runner
- func (r *Runner) Close() error
- func (r *Runner) Execute(ctx context.Context, j *job.Job) error
- func (r *Runner) Reclaim(ctx context.Context, workerID id.WorkerID) error
- func (r *Runner) WithArtifacts(svc *artifact.Service, scratchRoot string) *Runner
- func (r *Runner) WithTelemetry(t *telemetry.Telemetry) *Runner
- type Settings
Constants ¶
const DefaultReapInterval = 15 * time.Second
DefaultReapInterval is how often the pool scans for expired leases.
It is deliberately independent of any lease TTL. The reaper used to tick at the stale-job threshold, so a five-minute threshold meant a dead job could sit for ten minutes before anyone looked; and with per-definition TTLs there is no single threshold left to tick at.
const DefaultReclaimBatch = 100
DefaultReclaimBatch caps how many expired leases one pass reclaims, so a backlog after an outage drains over several ticks instead of one statement that locks a large slice of the table.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Executor
deprecated
type Executor = Runner
Executor is the former name of Runner.
The type was renamed because it orchestrates an attempt — middleware, retry, DLQ, state, events — and was never the thing that invokes the handler. That is now exec.Executor. This alias keeps existing code compiling.
Deprecated: use Runner.
type Pool ¶
type Pool struct {
// contains filtered or unexported fields
}
Pool manages a set of concurrent worker goroutines fed by a single fetcher that polls the store for jobs and executes them through the Executor.
func NewPool ¶
func NewPool( store job.Store, executor *Executor, extensions *ext.Registry, logger log.Logger, opts ...PoolOption, ) *Pool
NewPool creates a worker pool.
func (*Pool) Stop ¶
Stop signals all workers to stop and waits for them to finish. If the context has a deadline, active jobs are cancelled when time runs out.
func (*Pool) Wake ¶
func (p *Pool) Wake()
Wake nudges the fetcher to poll for jobs immediately, resetting any idle backoff. It is non-blocking and safe to call from any goroutine, including before Start. Call it after enqueuing a job in-process so the job is picked up without waiting out the idle poll interval.
type PoolOption ¶
type PoolOption func(*Pool)
PoolOption configures a Pool.
func WithDefaultLeaseTTL ¶
func WithDefaultLeaseTTL(d time.Duration) PoolOption
WithDefaultLeaseTTL sets how far each renewal pushes a job's lease expiry when the job declares no LeaseTTL of its own.
This is a liveness window, not a time limit: it should be a small multiple of the heartbeat interval regardless of how long the work takes. When unset the pool uses the configured stale-job threshold, so an existing deployment sees the same reclamation timing it had before leases existed.
func WithHeartbeatInterval ¶
func WithHeartbeatInterval(d time.Duration) PoolOption
WithHeartbeatInterval sets how often the pool sends heartbeats for active jobs. A zero value disables heartbeats.
func WithMaxPollInterval ¶
func WithMaxPollInterval(d time.Duration) PoolOption
WithMaxPollInterval caps the fetcher's idle backoff. Empty polls double the poll interval up to this value; dequeued work or Wake resets it.
func WithPollInterval ¶
func WithPollInterval(d time.Duration) PoolOption
WithPollInterval sets how often the pool's single fetcher (fetchLoop) polls the store for new jobs.
func WithPoolConcurrency ¶
func WithPoolConcurrency(n int) PoolOption
WithPoolConcurrency sets the number of concurrent worker goroutines.
func WithPoolQueues ¶
func WithPoolQueues(queues []string) PoolOption
WithPoolQueues sets the queues the pool will poll.
func WithQueueManager ¶
func WithQueueManager(m QueueManager) PoolOption
WithQueueManager sets the queue manager for rate limiting and concurrency control.
func WithReapInterval ¶
func WithReapInterval(d time.Duration) PoolOption
WithReapInterval sets how often the reaper scans for expired leases / stale jobs.
This is the scan cadence, not the lease duration — it does not control how long a lease survives without renewal. For that, see WithDefaultLeaseTTL and job.WithLeaseTTL. A zero value leaves DefaultReapInterval in place; it has no effect when WithStaleJobThreshold is zero, since that still disables reaping entirely.
func WithResourceManager ¶
func WithResourceManager(m resource.Manager) PoolOption
WithResourceManager makes the pool resource-aware.
With a manager installed the fetcher offers its free capacity as the dequeue budget, so the store never hands this worker a job it cannot run, and every claimed job holds a lease for as long as it executes. Without one the pool passes an unbounded DequeueOpts and takes no leases — every backend skips its fit predicate and behaviour is identical to a pool that predates the resource model.
Prefer engine.WithResourceManager to calling this directly. A manager makes admit able to stall the fetcher for up to one pollInterval per batch while it holds claimed, running-state, not-yet-heartbeating jobs, which puts pollInterval and staleJobThreshold into a relationship the pool cannot police: it is handed both as already-decided values and does not own the policy. engine.Build validates them together (checkReaperMargin) before it constructs a pool. Construct one here instead and that check does not run — a staleJobThreshold inside the stall lets the reaper reclaim a job this fetcher still holds, and the job runs twice.
func WithStaleJobThreshold ¶
func WithStaleJobThreshold(d time.Duration) PoolOption
WithStaleJobThreshold sets the threshold after which running jobs without a heartbeat are considered stale and reaped, on the legacy SELECT-then-UPDATE path a backend implementing only job.Store falls back to (reapStaleJobsLegacy). It has no effect on any first-party backend: every one of them (memory, mongo, postgres, redis, sqlite) implements job.LeaseStore, so reapStaleJobs routes to reclaimExpiredLeases instead, which reclaims purely by lease expiry (job.LeaseTTL / WithDefaultLeaseTTL) and never reads this value. A zero value disables reaping outright, on either path — see reapStaleJobs and reclaimExpiredLeases.
func WithStoreCallTimeout ¶
func WithStoreCallTimeout(d time.Duration) PoolOption
WithStoreCallTimeout caps a single store roundtrip. Pass a positive duration to override defaultStoreCallTimeout, zero to leave the default in place, or a negative value to disable the timeout entirely. Disabling is only intended for unit tests.
func WithTelemetry ¶
func WithTelemetry(t *telemetry.Telemetry) PoolOption
WithTelemetry sets where the pool records its spans and metrics: one root "process job" span per attempt, and the poll, requeue, lease and reclaim counters. engine.Build passes the engine's own; without it the pool records no metrics and starts spans on the global tracer provider.
func WithWorkerCustomKeys ¶
func WithWorkerCustomKeys(keys []string) PoolOption
WithWorkerCustomKeys sets the custom resource keys this worker offers at dequeue, overriding the keys derived from the resource manager's capacity.
The derived default is usually what you want; this exists to narrow it, so a worker draining a device can stop attracting work for it without being reconfigured. Keep the list a subset of the manager's custom capacity: dequeue matches custom resources by key and never by quantity, so a key advertised here that the manager has no capacity for will pass the store's filter and then be refused locally, and the job will bounce back to pending on every attempt.
Setting this without a resource manager makes the dequeue bounded on its own, which is a deliberate opt-in: the worker then claims only jobs whose custom keys it offers, with no quantity accounting at all.
The slice is copied, so the caller keeps no handle on pool state.
type QueueManager ¶
type QueueManager interface {
// Acquire checks rate limits and concurrency for the queue/tenant
// combination. Returns true if the job is allowed to proceed.
Acquire(queue, tenantID string) bool
// Release decrements the active count for the queue/tenant pair.
Release(queue, tenantID string)
}
QueueManager controls per-queue and per-tenant rate limiting and concurrency. The worker pool calls Acquire before executing a dequeued job and Release after execution completes.
type Runner ¶
type Runner struct {
// contains filtered or unexported fields
}
Runner executes a single job attempt: it selects an executor from the job's policy, runs the attempt through the middleware chain, then handles retry logic, DLQ push, state updates, and lifecycle events.
Runner orchestrates the attempt. It does not itself invoke the handler — that is exec.Executor's job, which is what lets the same attempt run in-process or in a pod without this file changing.
func NewExecutor
deprecated
func NewExecutor( registry *job.Registry, extensions *ext.Registry, store job.Store, dlqService *dlq.Service, bo backoff.Strategy, logger log.Logger, mws ...middleware.Middleware, ) *Runner
NewExecutor creates a Runner with no executor registry, so handlers are called directly in-process exactly as before.
Deprecated: use NewRunner, which takes an *exec.Registry.
func NewRunner ¶
func NewRunner( registry *job.Registry, extensions *ext.Registry, store job.Store, dlqService *dlq.Service, bo backoff.Strategy, executors *exec.Registry, logger log.Logger, mws ...middleware.Middleware, ) *Runner
NewRunner creates a Runner with the given dependencies.
A nil executors registry means handlers are called directly, which is the behaviour the deprecated NewExecutor preserves.
func (*Runner) Close ¶
Close releases every configured executor's own resources.
An engine closes its own registry when it stops, since it outlives the pool; this is the equivalent for a caller that assembles a Runner itself. Call it after the pool has finished its in-flight attempts.
func (*Runner) Execute ¶
Execute runs a job through the middleware chain and its executor, then routes the outcome through handleFailure/handleSuccess. The terminal write depends on which of several shapes the outcome takes:
- Success: marks completed, emits JobCompleted.
- A launch failure (the handler never ran) with launch attempts remaining: sets StatePending with a backoff delay and emits nothing — see requeueAfterLaunchFailure. It does not consume the job's own retry budget.
- A launch failure that exhausts maxLaunchAttempts, or any failure marked permanent (dispatch.ErrPermanent, or exec.Error.Permanent for a rung that cannot carry a Go error chain): DLQs immediately — on the FIRST attempt if that is when permanence was discovered, not only "after exhausting retries." Marks failed, pushes to DLQ, emits JobFailed + JobDLQ.
- An ordinary failure with retries remaining: marks retrying with backoff, emits JobRetrying.
- An ordinary failure with retries exhausted: marks failed, pushes to DLQ, emits JobFailed + JobDLQ, same as the permanent case.
- Any of the terminal writes above losing the race to job.ErrLeaseLost, or the pool having cancelled the attempt with that cause: abandonLostLease writes nothing to the store at all — the current lease holder's own write must stand untouched — and only emits JobCancelled when the row says an operator cancelled it, or JobFailed otherwise, so extensions still observe the loss.
func (*Runner) Reclaim ¶
Reclaim asks every configured executor to release sandboxes this worker leaked across a restart, and removes stale scratch output directories a previous process left behind. The pool calls it once at startup.
Failures are joined rather than fatal: a rung that cannot sweep should not stop the worker from running the jobs it can still execute.
func (*Runner) WithArtifacts ¶
WithArtifacts configures the artifact plane an out-of-process rung commits its outputs through, and the root directory its scratch OutputDir is created under. It returns r so callers can chain it onto NewRunner.
Never calling this, or passing a nil or disabled svc, leaves output committing off: an out-of-process attempt still gets a fresh, empty OutputDir that is removed once the attempt ends, but PriorOutputs stays empty and nothing the sandbox wrote is committed — exactly Runner's behaviour before this existed. An empty scratchRoot defaults to os.TempDir().
func (*Runner) WithTelemetry ¶
WithTelemetry sets where the runner records attempt and dead letter metrics, and returns r so callers can chain it onto NewRunner. Its spans (one per terminal path, plus "commit job outputs") nest under whatever span the attempt's context carries, the pool's "process job" span in practice, and are not started without one.
type Settings ¶
type Settings struct {
Concurrency int
Queues []string
PollInterval, MaxPollInterval, HeartbeatInterval, StaleJobThreshold time.Duration
ReapInterval, DefaultLeaseTTL, StoreCallTimeout time.Duration
StoreCallsBounded, ReapingEnabled, LeasesEnabled bool
}
Settings is the effective configuration of this process's worker pool.
Source Files