retry

Documentation

Overview

Package retry implements the bounded-retry envelope used to wrap the long-lived retry loops in the Parti runtime (source watcher restart, handoff watcher restart, assignment watcher, dynamic consumer recovery). The envelope replaces forever-retry loops with a bounded attempt budget, exponential backoff with a hard ceiling, jitter to avoid thundering herds, and a one-shot escalation callback on exhaustion so the worker can take itself out of rotation rather than generate infinite API load against a vanished resource.

The package is internal — callers wire their site-specific Work, Classify, and OnPermanent functions through the Config struct.

Index

• Variables
• type Class
• type Config
• type Envelope
  • func New(cfg Config) *Envelope
  • func (e *Envelope) Run(ctx context.Context) error

Variables

var ErrExhausted = errors.New("retry envelope: attempt budget exhausted")

ErrExhausted is returned by Run when the attempt budget is exhausted (after firing OnPermanent) or when Classify returned GiveUp.

Types

type Class

type Class int

Class categorises an error returned by the work function.

const (
	// Transient: retry after backoff (subject to the attempt budget).
	Transient Class = iota
	// GiveUp: stop immediately; the resource is gone and waiting longer
	// will not help. Fires the permanent-failure callback.
	GiveUp
	// Fatal: bubble the error up to the caller without retry and without
	// firing the permanent-failure callback. Use this when the caller
	// must inspect the error directly (e.g. configuration mistakes).
	Fatal
)

type Config

type Config struct {
	// Work is invoked on each attempt. nil is not permitted.
	Work func(ctx context.Context) error
	// Classify maps a Work error to a Class. If nil, every error is
	// treated as Transient.
	Classify func(err error) Class
	// OnPermanent is invoked at most once when the envelope gives up
	// (attempt budget exhausted OR Classify returned GiveUp). The
	// caller's last error is passed in. The callback runs synchronously
	// on the envelope's goroutine — it must be non-blocking.
	OnPermanent func(err error)
	// OnProgress is invoked after each failed attempt with the (1-based)
	// attempt number and the error returned. Optional.
	OnProgress func(attempt int, err error)
	// BaseBackoff is the delay before the second attempt; doubles each
	// step up to MaxBackoff. Required.
	BaseBackoff time.Duration
	// MaxBackoff caps the per-attempt delay. Required.
	MaxBackoff time.Duration
	// MaxAttempts is the total attempt budget per Run invocation. After
	// the Nth failure (counting from 1) the envelope fires OnPermanent
	// and returns ErrExhausted. Must be > 0.
	MaxAttempts int
	// Jitter is the ± fraction applied to each backoff delay
	// (0..1 reasonable; 0 disables). Avoids synchronized reconnect
	// thundering herds across worker fleets.
	Jitter float64
	// Clock and Sleep are optional indirections for tests; production
	// uses time.Now and time.After.
	Now   func() time.Time
	Sleep func(ctx context.Context, d time.Duration) error
}

Config configures a bounded-retry envelope.

type Envelope

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

Envelope is the bounded-retry runner.

func New

func New(cfg Config) *Envelope

New builds an Envelope from the given Config. Panics on invalid configuration (Work nil, MaxAttempts <= 0, MaxBackoff <= 0, BaseBackoff <= 0). These are programmer errors, not user errors — fail loudly.

func (*Envelope) Run

func (e *Envelope) Run(ctx context.Context) error

Run drives the envelope. Returns:

• nil on success
• ctx.Err() if the context cancels at any point (no OnPermanent fired)
• the original Work error on a Fatal classification (no OnPermanent fired)
• ErrExhausted on attempt-budget exhaustion or GiveUp (OnPermanent fired exactly once with the most recent Work error)

Concurrency: Run is single-shot; one Envelope per logical retry loop. Reusing the same Envelope concurrently is not supported.