Documentation
¶
Overview ¶
Package worker runs claimed jobs. A Pool starts one claim loop per queue that feeds a bounded set of handler goroutines, plus one heartbeater that extends the leases of everything in flight and fences jobs whose lease was lost or cancelled (docs/PLAN.md, step 7).
Index ¶
Constants ¶
const ( DefaultConcurrency = 16 DefaultBatchSize = 32 DefaultLeaseTTL = 30 * time.Second DefaultHeartbeatEvery = 10 * time.Second DefaultJobTimeout = 10 * time.Minute DefaultPollFloor = 200 * time.Millisecond )
Defaults applied by NewPool to zero Config fields.
Variables ¶
var ErrNoQueues = errors.New("worker: no queues configured")
ErrNoQueues is returned by Run when Config.Queues is empty.
Functions ¶
func DefaultOwner ¶
func DefaultOwner() string
DefaultOwner returns "<hostname>-<pid>-<4 hex>", unique per process start so a restarted process never confuses its own stale leases with live ones.
Types ¶
type Config ¶
type Config struct {
// Queues are the queues this pool claims from; Run fails without one.
Queues []string
// Concurrency bounds handler goroutines across all queues (DefaultConcurrency).
Concurrency int
// BatchSize is the most jobs one claim asks for (DefaultBatchSize).
BatchSize int
// LeaseTTL is how long a claim or heartbeat holds a lease (DefaultLeaseTTL).
LeaseTTL time.Duration
// HeartbeatEvery is the heartbeat period (DefaultHeartbeatEvery); it is
// clamped to LeaseTTL/3 when set higher than that.
HeartbeatEvery time.Duration
// JobTimeout bounds one handler call and is also the drain budget when Run
// stops (DefaultJobTimeout).
JobTimeout time.Duration
// Owner identifies this process in lease_owner (DefaultOwner()).
Owner string
// PollFloor is the claim loop's tick when the dispatcher stays silent
// (DefaultPollFloor).
PollFloor time.Duration
// Durations, when set, receives the wall time of every handler run
// (success or failure) so jobrun_job_duration_seconds can be exposed
// without the worker importing internal/admin. Optional.
Durations DurationObserver
}
Config tunes a Pool. Zero fields take the defaults listed on each.
type DurationObserver ¶
DurationObserver observes one handler run per call. *admin.Histogram satisfies it; cmd/jobrund passes it through Config.Durations.
type EventPublisher ¶
EventPublisher receives every transition the pool drives (queued -> leased, leased -> succeeded|queued|dead|cancelled). Publish must not block: the admin broker drops for slow subscribers.
type Pool ¶
type Pool struct {
// contains filtered or unexported fields
}
Pool claims jobs from the configured queues and runs their handlers.
Run may be called once; the pool is not reusable.
func NewPool ¶
func NewPool(s job.Store, d job.Dispatcher, reg *job.Registry, pub EventPublisher, log *slog.Logger, c Config) *Pool
NewPool wires a pool. d may be nil (a job.PollOnly ticker on PollFloor is used), pub may be nil (events are dropped), log may be nil (slog.Default).
func (*Pool) Run ¶
Run claims and executes jobs until ctx is done, then stops claiming, waits up to JobTimeout for in-flight handlers (heartbeating them meanwhile), cancels whatever is still running, waits for those goroutines to return and returns nil. Handler contexts are not cancelled by ctx: they end by finishing, by JobTimeout, by a lost lease or a cancel request, or by the drain budget.
type Stats ¶
type Stats struct {
// Claimed is the number of jobs handed to a handler goroutine.
Claimed int64
// Succeeded is the number of Complete calls that went through.
Succeeded int64
// Failed is the number of handler calls that returned an error (or panicked).
Failed int64
// Retried is the number of failures the store put back in the queue.
Retried int64
// Dead is the number of failures the store moved to dead.
Dead int64
// LeaseLost is the number of Complete/Fail calls rejected with ErrLeaseLost.
LeaseLost int64
// Fenced is the number of handlers cancelled because a heartbeat found
// their lease gone (or it could not be renewed in time); their result was
// dropped without calling Complete or Fail.
Fenced int64
// InFlight is the number of handlers running right now.
InFlight int64
}
Stats is a snapshot of the pool's counters since it was created.
Source Files