Documentation
¶
Overview ¶
Package work runs the persistent work queue (decision 0033): a pool of workers that lease due items, run the handler registered for each item's kind, and complete, retry or park it. Work is asked for with Enqueue inside the transaction of the change that needs it; readers that must see the work done call EnsureFresh first.
Index ¶
- Constants
- Variables
- func Backoff(attempt int) time.Duration
- type Handler
- type Options
- type Queue
- func (q *Queue) Enqueue(ctx context.Context, kind, key string, payload []byte) error
- func (q *Queue) EnsureFresh(ctx context.Context, kind, key string) error
- func (q *Queue) Handle(kind string, h Handler) error
- func (q *Queue) HandleEvery(kind string, every time.Duration, h Handler) error
- func (q *Queue) Run(ctx context.Context) error
- func (q *Queue) RunDue(ctx context.Context, kind string) (int, error)
- func (q *Queue) WaitIdle(ctx context.Context) error
Constants ¶
const ( DefaultWorkers = 4 DefaultLease = 2 * time.Minute DefaultMaxAttempts = 10 DefaultPoll = time.Second )
Defaults for Options left zero.
Variables ¶
var ( // ErrParked reports that the work a reader waits for keeps failing and // is parked until it is enqueued again. ErrParked = errors.New("work: parked") // ErrNoHandler reports a kind with no registered handler. ErrNoHandler = errors.New("work: no handler") // ErrOptions reports options New cannot use. ErrOptions = errors.New("work: invalid options") // ErrDuplicateHandler reports a second handler for one kind. ErrDuplicateHandler = errors.New("work: kind already has a handler") // ErrPanic reports a handler that panicked. ErrPanic = errors.New("work: handler panicked") )
Functions ¶
Types ¶
type Handler ¶
Handler does the work of one item. It must be idempotent: an item can run more than once, for example when a worker stops after the work but before recording it. A cancelled ctx means the queue is stopping or the lease was lost; the handler should return promptly.
type Options ¶
type Options struct {
// Workers is how many items run at once on this node.
Workers int
// Lease is how long a run holds its item before another worker may
// take it over; a running item's lease is extended every half lease.
Lease time.Duration
// MaxAttempts is how many runs an item gets before it is parked.
MaxAttempts int
// Poll is how long an idle worker waits before asking again, when no
// enqueue on this node wakes it sooner.
Poll time.Duration
// Holder names this node's leases; empty uses the host name, the
// process id and a random suffix.
Holder string
// Backoff is the wait before an item's next run after its attempt'th
// failed; nil doubles from a second up to ten minutes.
Backoff func(attempt int) time.Duration
Now func() time.Time
Log *slog.Logger
// Meters exports the queue's metrics (decision 0016); nil exports none.
Meters metric.MeterProvider
}
Options configures a Queue.
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue runs the persistent work queue for one node.
func (*Queue) Enqueue ¶
Enqueue asks for kind's work on key. Call it with the context of the transaction that makes the change needing the work, so the work is asked for exactly when the change commits; this node's workers are woken once it has.
func (*Queue) EnsureFresh ¶
EnsureFresh brings the work owed for kind and key up to date before the caller reads what it derives: it runs a waiting item now, in the caller's context, or waits for a run already in progress elsewhere. It returns nil when nothing is owed, the handler's error when the run it made fails, and ErrParked when the item is parked.
func (*Queue) Handle ¶
Handle registers h for kind. Register every handler before Run: Run leases only the kinds registered when it starts. A second handler for one kind is an error.
func (*Queue) HandleEvery ¶
HandleEvery registers h for kind as periodic work, for work that Artifactory does on a timer rather than as it is asked for, such as its snapshot cleanup every five minutes. An item of kind is not run as soon as it is due: it waits in the queue, so a restart keeps it, for the next tick of a timer every interval on each node, which runs every item of kind then due (RunDue). EnsureFresh still runs one at once. Register it before Run, as Handle.
func (*Queue) Run ¶
Run leases and runs items until ctx ends, then hands back the leases of runs it cuts short and returns once every run has settled. Periodic kinds run on their own timers, outside the workers.
func (*Queue) RunDue ¶
RunDue runs every item of kind that is due now, one after another, and returns how many runs it made. A periodic kind's timer calls it; so may a caller that need not wait for the timer, such as a test. A run that fails is retried after its backoff, never in the same pass, and an item asked for again during its run waits for the next pass: it is due again from the end of its run, after every item that was due before it, so meeting it again means the pass has taken everything else.
func (*Queue) WaitIdle ¶
WaitIdle returns once no item of a registered kind is pending or leased and no run is in progress on this node, for tests. Parked items do not count; an item that keeps failing keeps it waiting until ctx ends. A pending item of a periodic kind does not count either, since it waits for its timer by design; RunDue runs it.
Source Files