worker

Documentation

Overview

Package worker implements anchor's consumer: the loop that claims runs from the queue, re-executes them under replay to their frontier, and persists exactly one outcome per claim. The worker is a loop, not a service — it exposes no listener and accepts no inbound calls. Every arrow points outward: to the ledger for truth, to the queue for flow. That is what lets any number of workers run anywhere without coordination beyond the two interfaces.

Index

• type Option
  • func WithBackoff(fn func(attempt int) time.Duration) Option
  • func WithCodec(c run.Codec) Option
  • func WithConcurrency(n int) Option
  • func WithHeartbeatEvery(d time.Duration) Option
  • func WithJanitorEvery(d time.Duration) Option
  • func WithLeaseTTL(d time.Duration) Option
  • func WithLogger(l *slog.Logger) Option
  • func WithRegistry(r *run.Registry) Option
• type Worker
  • func New(lgr ledger.Ledger, q queue.Queue, opts ...Option) *Worker
  • func (w *Worker) Run(ctx context.Context) error

Types

type Option

type Option func(*options)

Option configures a Worker.

func WithBackoff

func WithBackoff(fn func(attempt int) time.Duration) Option

WithBackoff sets the retry delay as a function of the failure count just reached. The default is exponential — 250ms·2^(n−1) capped at 30s — with ±20% jitter so a synchronized fleet spreads its retries.

func WithCodec

func WithCodec(c run.Codec) Option

WithCodec overrides the default JSON codec. It must match the codec the submitting client used.

func WithConcurrency

func WithConcurrency(n int) Option

WithConcurrency sets the number of claim slots (default GOMAXPROCS).

func WithHeartbeatEvery

func WithHeartbeatEvery(d time.Duration) Option

WithHeartbeatEvery sets the lease renewal interval (default LeaseTTL/3).

func WithJanitorEvery

func WithJanitorEvery(d time.Duration) Option

WithJanitorEvery sets the repair-loop interval (default 30s); 0 disables the janitor on this worker.

func WithLeaseTTL

func WithLeaseTTL(d time.Duration) Option

WithLeaseTTL sets the claim lease duration (default 30s). Heartbeats renew it; a worker that stops heartbeating loses the run after at most this long.

func WithLogger

func WithLogger(l *slog.Logger) Option

WithLogger sets the logger for warnings (default slog.Default()).

func WithRegistry

func WithRegistry(r *run.Registry) Option

WithRegistry sets the registry workflow names are resolved against (default run.DefaultRegistry).

type Worker

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

Worker consumes runs. Each claim is driven through a fixed protocol — claim → load → replay to frontier → exactly one of Complete+Settle, Settle (park), or Release (retry) — with lease expiry covering the paths where the worker itself dies. Create with New, start with Run.

func New

func New(lgr ledger.Ledger, q queue.Queue, opts ...Option) *Worker

New composes a worker over any Ledger and Queue implementation. The queue's DeltaLog capability is discovered by type assertion; without it, attempt dividers are skipped and streams are simply not produced.

func (*Worker) Run

func (w *Worker) Run(ctx context.Context) error

Run blocks, consuming runs until ctx is done. It starts Concurrency claim loops plus, unless disabled, one janitor goroutine. On cancellation it stops claiming, finishes or releases every in-flight run, waits for all goroutines, and returns ctx.Err().