loop

Documentation

Overview

Package loop runs background work as periodic workflow passes.

A Ticker drives one pass, an interfaces.Workflow[struct{}, struct{}] the composition root has already decorated (recovery, tracing, metrics, logging), on a schedule: once per Interval, or back to back when Interval is zero. A failed pass is logged and retried after ErrorBackoff; the loop ends only when its context does. A Group starts a service's loops together, keeps one loop's failure from stopping the others, and waits for every loop on shutdown; it is an interfaces.Lifecycle, so a lifecycle.Manager can own it. Pass adapts a func(context.Context) error, such as a RunOnce method, into the workflow a Ticker runs.

Index

• func Pass(fn func(ctx context.Context) error) interfaces.Workflow[struct{}, struct{}]
• type Config
  • func (c Config) Validate() error
• type Decorator
• type Group
  • func NewGroup(logger interfaces.Logger) *Group
  • func NewGroupFromSpecs(specs []Spec, decorate Decorator, logger interfaces.Logger) (*Group, error)
  • func (g *Group) Add(name string, r Runner)
  • func (g *Group) Start(ctx context.Context) error
  • func (g *Group) Stop(ctx context.Context) error
  • func (g *Group) Wait()
• type Runner
• type RunnerFunc
  • func (f RunnerFunc) Run(ctx context.Context) error
• type Spec
• type Ticker
  • func NewTicker(name string, pass interfaces.Workflow[struct{}, struct{}], cfg Config, ...) (*Ticker, error)
  • func (t *Ticker) Run(ctx context.Context) error

Functions

func Pass

func Pass(fn func(ctx context.Context) error) interfaces.Workflow[struct{}, struct{}]

Pass adapts fn, one iteration of a loop, into the workflow a Ticker runs, so the composition root can decorate it like any other workflow. A nil fn panics.

Types

type Config

type Config struct {
	// Interval is the time between passes. Zero runs passes back to back, for a pass
	// that already blocks on its source (a stream read with a block timeout).
	Interval time.Duration
	// ErrorBackoff is the wait after a failed pass before the loop continues. Zero
	// continues at once; with a zero Interval that suits only a pass that blocks on its
	// source, since a failing source would otherwise turn the loop into a hot spin.
	ErrorBackoff time.Duration
	// RunAtStart runs one pass as soon as the loop starts instead of after the first
	// Interval.
	RunAtStart bool
}

Config schedules a Ticker.

func (Config) Validate

func (c Config) Validate() error

Validate rejects a negative Interval or ErrorBackoff with CodeInvalidInput.

type Decorator

type Decorator func(
	op string,
	pass interfaces.Workflow[struct{}, struct{}],
) interfaces.Workflow[struct{}, struct{}]

Decorator wraps one loop pass in the composition root's workflow decorator stack under op, so each pass is logged, counted, traced and recovered like any other workflow.

type Group

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

Group starts a service's loops together and waits for all of them.

func NewGroup

func NewGroup(logger interfaces.Logger) *Group

NewGroup returns an empty Group that logs to logger. A nil logger panics.

func NewGroupFromSpecs

func NewGroupFromSpecs(
	specs []Spec,
	decorate Decorator,
	logger interfaces.Logger,
) (*Group, error)

NewGroupFromSpecs builds one Ticker per spec, its pass wrapped by decorate (a nil decorate leaves passes undecorated), into a Group that starts and stops them together, in spec order. An invalid schedule is CodeInvalidInput; a nil logger or Run panics, as a wiring bug.

func (*Group) Add

func (g *Group) Add(name string, r Runner)

Add registers r under name. Add after Start panics, as a wiring bug.

func (*Group) Start

func (g *Group) Start(ctx context.Context) error

Start runs every runner on its own goroutine under a context derived from ctx and returns at once. ctx is the loops' lifetime, not a startup deadline: when it ends, every loop stops as it does on Stop, so pass the service's run context. A runner that returns an error while that context is live is logged at error with its name; the others keep running. One that returns after the shutdown began is logged at debug, or not at all when it returned the context's own error. A second Start is CodeConflict.

func (*Group) Stop

func (g *Group) Stop(ctx context.Context) error

Stop cancels the runners and waits for them to return. If ctx ends first, Stop returns its error coded CodeTimeout or CodeCanceled. Stop on a group that was never started returns nil.

func (*Group) Wait

func (g *Group) Wait()

Wait blocks until every runner has returned. It returns at once on a group that was never started.

type Runner

type Runner interface {
	// Run executes the loop until ctx ends (returning nil) or it cannot continue
	// (returning the error).
	Run(ctx context.Context) error
}

Runner is one background loop: Run blocks until ctx ends or the loop fails for good.

interface-composition exemption — narrow consumer-side port: a single Run(ctx) → error loop contract, not an id-CRUD data-access surface, so it embeds no foundation generic.

type RunnerFunc

type RunnerFunc func(ctx context.Context) error

RunnerFunc adapts a function into a Runner.

func (RunnerFunc) Run

func (f RunnerFunc) Run(ctx context.Context) error

Run calls f.

type Spec

type Spec struct {
	// Run is one iteration of the loop.
	Run func(ctx context.Context) error

	// Name labels the loop in logs and in the Group.
	Name string

	// Op is the workflow op the pass is decorated under.
	Op string

	// Schedule paces the loop.
	Schedule Config
}

Spec declares one background loop: the pass it runs, the op that pass is decorated under, and its schedule. A service lists its loops as Specs and the composition root turns them into one Group with NewGroupFromSpecs, so the service never builds a Ticker or a decorator itself.

type Ticker

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

Ticker runs one workflow pass on a schedule until its context ends.

func NewTicker

func NewTicker(
	name string,
	pass interfaces.Workflow[struct{}, struct{}],
	cfg Config,
	logger interfaces.Logger,
) (*Ticker, error)

NewTicker returns a Ticker that runs pass on cfg's schedule, logging failures to logger under name. An invalid cfg is CodeInvalidInput; a nil pass or logger panics, as a wiring bug.

func (*Ticker) Run

func (t *Ticker) Run(ctx context.Context) error

Run executes passes until ctx ends, then returns nil. With a zero Interval passes run back to back; otherwise one runs per Interval, plus one at start when RunAtStart is set. A pass that outlasts Interval delays the next one rather than queueing the missed ticks, so passes never overlap. Run satisfies Runner.