worker

Documentation

Overview

Package worker implements the straddler job processing pool.

Index

• func Backoff(attempt int, base, max time.Duration) time.Duration
• type Pool
  • func NewPool(concurrency int, staleTimeout time.Duration, cfg WorkerConfig, q *db.Queue, ...) *Pool
  • func (p *Pool) Run(ctx context.Context) error
• type WorkerCallbacks
• type WorkerConfig

Functions

func Backoff

func Backoff(attempt int, base, max time.Duration) time.Duration

Backoff computes a retry delay using exponential backoff with ±10% jitter.

delay = clamp(base * 2^(attempt-1), 0, max) ± 10% jitter

attempt is 1-indexed — pass job.AttemptCount directly after it has been incremented by ClaimNextJob.

The jitter prevents a thundering herd: if N workers all fail at the same moment they won't all wake up and retry at exactly the same instant.

Types

type Pool

type Pool struct {
	// contains filtered or unexported fields
}

Pool manages a fixed number of concurrent workers and a background reaper.

func NewPool

func NewPool(concurrency int, staleTimeout time.Duration, cfg WorkerConfig, q *db.Queue, r registry.Client) *Pool

NewPool creates a Pool. concurrency is the number of parallel copy workers. staleTimeout is how long a job can stay in_progress before the reaper resets it back to pending (handles crashed workers).

func (*Pool) Run

func (p *Pool) Run(ctx context.Context) error

Run starts all workers and the reaper, blocking until ctx is cancelled or (when cfg.ExitWhenDone is true) the queue is fully drained. All goroutines complete before Run returns.

type WorkerCallbacks

type WorkerCallbacks struct {
	// OnComplete is called after a job is successfully marked complete in the DB.
	OnComplete func(sourceRef, destRef string, duration time.Duration)

	// OnFailed is called after a job permanently exhausts all retry attempts.
	// Not called for transient failures that will be retried.
	OnFailed func(sourceRef, destRef string, errMsg string)

	// OnHeartbeat is called every heartbeatInterval while a copy is in flight.
	OnHeartbeat func(sourceRef string, elapsed time.Duration)
}

WorkerCallbacks holds optional event hooks called during job processing. All functions must be safe for concurrent use from multiple worker goroutines. Nil functions are silently skipped.

type WorkerConfig

type WorkerConfig struct {
	// WorkerID identifies this worker in the claimed_by column and log fields.
	// The pool appends a numeric suffix to the base ID for each goroutine.
	WorkerID     string
	PollInterval time.Duration
	MaxAttempts  int
	BaseBackoff  time.Duration
	MaxBackoff   time.Duration

	// ExitWhenDone makes workers return nil (instead of polling forever) when the
	// queue has no claimable jobs AND no jobs are in_progress. Used by the `run`
	// command to exit automatically without a Ctrl+C.
	ExitWhenDone bool

	// Callbacks are optional event hooks. Zero value (all nil) is valid.
	Callbacks WorkerCallbacks
}

WorkerConfig holds parameters for a worker. All workers in a pool share one WorkerConfig. Named WorkerConfig (not Config) to avoid ambiguity with internal/config.Config at call sites.