resilience

type Backoff ¶

type Backoff func(attempt int) time.Duration

Backoff returns the delay before the specified one-based retry attempt.

func ExponentialBackoff(base, max time.Duration, jitter float64) (Backoff, error)

type Breaker ¶

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

Breaker is a circuit breaker: it stops calling a resource that keeps failing, and after OpenFor lets a few calls through to find out whether it has recovered. Fail fast is the whole point, so a rejected call returns ErrOpen immediately; nothing in a Breaker waits.

Opening schedules the move to HalfOpen through the Clock's AfterFunc, so State and Hooks.OnStateChange reflect the elapsed OpenFor on their own rather than at the next call. Every operation also applies any due transition itself, so a late timer changes nothing.

package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/apsis-io/velocity/resilience"
)

func main() {
	breaker, _ := resilience.NewBreaker(resilience.BreakerPolicy{
		Trip:    resilience.ConsecutiveFailures(2),
		OpenFor: time.Minute,
		Failure: func(err error) bool { return !errors.Is(err, context.Canceled) },
	})
	unreachable := func(context.Context) (string, error) { return "", errors.New("connection refused") }

	for range 3 {
		_, err := breaker.Do(context.Background(), unreachable)
		fmt.Println(breaker.State(), errors.Is(err, resilience.ErrOpen))
	}
}

Output:
closed false
open false
open true

package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/apsis-io/velocity/resilience"
)

func main() {
	breaker, _ := resilience.NewBreaker(resilience.BreakerPolicy{
		Trip:    resilience.ConsecutiveFailures(2),
		OpenFor: time.Minute,
	})
	calls := 0
	_, err := resilience.Retry(context.Background(), resilience.Policy{
		MaxAttempts: 10,
		Retryable:   func(err error) bool { return !errors.Is(err, resilience.ErrOpen) },
	}, func(ctx context.Context) (string, error) {
		return breaker.Do(ctx, func(context.Context) (string, error) {
			calls++
			return "", errors.New("connection refused")
		})
	})
	fmt.Println(calls, errors.Is(err, resilience.ErrOpen))
}

Output:
2 true

func NewBreaker(policy BreakerPolicy) (*Breaker, error)

func (b *Breaker) Allow() (report func(error), err error)

func (b *Breaker) Counts() Counts

func (b *Breaker) Do[T any](ctx context.Context, fn func(context.Context) (T, error)) (T, error)

func (b *Breaker) Reset()

func (b *Breaker) State() State

type BreakerError ¶

type BreakerError struct {
	State      State
	RetryAfter time.Duration
}

BreakerError is a call the Breaker refused to make. State is Open, with RetryAfter the time until probes are admitted, or HalfOpen when every probe slot is taken, in which case RetryAfter is zero because the answer depends on the probes in flight rather than on the clock.

func (e *BreakerError) Error() string

func (e *BreakerError) Is(target error) bool

type BreakerHooks ¶

type BreakerHooks struct {
	// OnStateChange reports each transition with the counts that ended the
	// previous state.
	OnStateChange func(from, to State, counts Counts)
}

BreakerHooks lets a caller observe transitions without the package owning metrics state. Nil callbacks are skipped. Hooks run outside the breaker's lock, on the goroutine whose call caused the transition.

type BreakerPolicy ¶

type BreakerPolicy struct {
	// Trip decides when a Closed breaker opens.
	Trip Trip
	// OpenFor is how long the breaker stays Open before admitting probes.
	OpenFor time.Duration
	// Interval, when positive, resets Counts while Closed each time it
	// elapses, so a ratio Trip judges recent behaviour rather than the whole
	// history. Zero never resets until a transition.
	Interval time.Duration
	// MaxProbes bounds concurrent calls while HalfOpen; zero means one.
	MaxProbes int
	// SuccessesToClose is how many probes must succeed, with no failure in
	// between, to close again; zero means one.
	SuccessesToClose int
	// Failure decides whether an error counts against the breaker. Nil counts
	// every error, including the caller's own cancellation; most callers want
	// to exclude that, since it says nothing about the protected resource:
	//
	//	Failure: func(err error) bool { return !errors.Is(err, context.Canceled) }
	Failure Classifier
	Clock   Clock
	Hooks   BreakerHooks
}

BreakerPolicy configures a Breaker. Trip and OpenFor are required.

type Classifier ¶

type Classifier func(error) bool

Classifier decides whether an error should be retried. Nil retries all errors.

type Clock ¶

type Clock interface {
	Now() time.Time
	Sleep(context.Context, time.Duration) error
	// AfterFunc runs f on its own goroutine once d has elapsed, unless the
	// returned Timer is stopped first.
	AfterFunc(d time.Duration, f func()) Timer
}

Clock supplies time to Retry and Breaker: the current reading, cancellation-aware sleeping, and deferred execution. ManualClock implements it for tests.

func RealClock() Clock

type Counts ¶

type Counts struct {
	Requests             int
	Successes            int
	Failures             int
	ConsecutiveSuccesses int
	ConsecutiveFailures  int
}

Counts are the outcomes recorded since the last state change or Interval reset. They describe completed calls only; an admitted call that has not reported yet is in none of them.

type HedgeBudget ¶ added in v0.4.0

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

HedgeBudget bounds the extra load hedging creates, as a fraction of the traffic that justified it. Each execution credits the budget once, and each speculative attempt spends one credit, so a ratio of 0.1 permits a hedge on roughly one call in ten however the delays fall.

Without one, a dependency that has become slow enough to trigger hedging receives several times its usual load at exactly the wrong moment. Share one budget per dependency across every policy that hedges it.

A nil *HedgeBudget permits every hedge, which is the right choice only when the operation is cheap or the caller is the only client.

func NewHedgeBudget(ratio float64, burst int) (*HedgeBudget, error)

func (b *HedgeBudget) Tokens() float64

type HedgeHooks ¶ added in v0.4.0

type HedgeHooks struct {
	// OnAttempt fires as each attempt starts. hedge is false for the first
	// attempt and true for every speculative one.
	OnAttempt func(attempt int, hedge bool)
	// OnAttemptComplete fires as each attempt finishes, including attempts
	// that lose the race and those cancelled after a winner is chosen. won
	// reports whether this attempt's value was the one returned.
	OnAttemptComplete func(attempt int, duration time.Duration, err error, won bool)
	// OnDiscard fires for each losing result handed to Discard, with
	// whatever Discard returned. It is the only report of a disposal
	// failure: the caller already has their value, and a hedge that could
	// not clean up after itself is an operational event, not a call error.
	OnDiscard func(attempt int, err error)
}

HedgeHooks lets a caller observe hedged execution without the package owning metrics state. Nil callbacks are skipped.

type HedgePolicy ¶ added in v0.4.0

type HedgePolicy[T any] struct {
	// MaxAttempts bounds attempts in total, the first included. It must be
	// at least 2; one attempt is not a hedge.
	MaxAttempts int
	// Delay reports how long to wait before starting attempt n while the
	// attempts already running have neither succeeded nor failed. It is the
	// same Backoff that Retry uses, so a fixed delay is
	// func(int) time.Duration { return d } and a widening one is
	// ExponentialBackoff.
	Delay Backoff
	// Budget bounds how much extra load hedging is allowed to create. A nil
	// Budget hedges whenever the policy says to, which multiplies load on
	// the very dependency that is already slow.
	Budget *HedgeBudget
	// Discard disposes the result of an attempt that arrived after the
	// winner, which is otherwise leaked: a hedge produces N results and
	// returns one. Nil means T needs no disposal. For an owned resource,
	// this is its Drop.
	Discard func(T) error
	// Retryable decides whether a failed attempt should be followed by
	// another. Nil hedges past every error. A non-retryable error ends the
	// execution immediately and is returned as is.
	Retryable Classifier
	Clock     Clock
	Hooks     HedgeHooks
}

HedgePolicy configures Hedge. MaxAttempts and Delay are required.

type LatencyDelay ¶ added in v0.4.0

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

LatencyDelay derives a hedge delay from observed latency instead of a guess. A static delay has to be chosen for a latency distribution the caller does not yet know and that shifts under load; this hedges an attempt once it has taken longer than the given quantile of recent successful ones, so "too slow" is defined by what the dependency has actually been doing.

latency, err := resilience.NewLatencyDelay(0.95, 256, 20, time.Second)
policy := resilience.HedgePolicy[T]{
    MaxAttempts: 2,
    Delay:       latency.Delay,
    Hooks:       resilience.HedgeHooks{OnAttemptComplete: latency.Observe},
}

The two halves are wired separately because Hooks has one slot: a caller with their own instrumentation calls Observe from inside it. Only successful attempts are recorded — a failure's duration says how fast the dependency broke, not how long its work takes.

It is safe for concurrent use; share one per dependency, as the samples are only meaningful across executions.

func NewLatencyDelay(quantile float64, window, minSamples int, warmup time.Duration) (*LatencyDelay, error)

func (d *LatencyDelay) Delay(int) time.Duration

func (d *LatencyDelay) Observe(_ int, duration time.Duration, err error, _ bool)

func (d *LatencyDelay) Samples() int

type ManualClock ¶

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

ManualClock is a Clock that moves only when told to, for deterministic tests of Retry and Breaker. Sleep advances the clock by the requested delay instead of waiting, and honours a context that is already done. AfterFunc callbacks run synchronously, in due order, on the goroutine that advances the clock past them.

clock := resilience.NewManualClock(time.Unix(0, 0))
breaker, _ := resilience.NewBreaker(resilience.BreakerPolicy{Clock: clock, ...})
clock.Advance(policy.OpenFor) // now half-open, hook already fired

It is safe for concurrent use.

func NewManualClock(start time.Time) *ManualClock

func (c *ManualClock) Advance(d time.Duration)

func (c *ManualClock) AfterFunc(d time.Duration, f func()) Timer

func (c *ManualClock) Now() time.Time

func (c *ManualClock) Set(t time.Time)

func (c *ManualClock) Sleep(ctx context.Context, d time.Duration) error

func (c *ManualClock) Sleeps() int

type Policy ¶

type Policy struct {
	MaxAttempts int
	Retryable   Classifier
	Backoff     Backoff
	Clock       Clock
}

Policy configures Retry.

MaxAttempts is the bound when positive. Zero means the caller's context is the bound, which is the right shape for work that has a deadline of its own: attempts = budget/interval is a derivation that binds first, ends the loop up to one interval short of the budget, and does so silently. A policy that bounds nothing — zero attempts and a context with no deadline — is ErrInvalidPolicy.

type PolicyError ¶

type PolicyError struct{ Cause error }

PolicyError identifies invalid retry or breaker configuration.

func (e *PolicyError) Error() string

func (e *PolicyError) Unwrap() []error

type RetryError ¶

type RetryError struct {
	Attempts int
	Last     error
}

RetryError is every give-up in this package: a retry whose attempts ran out, a retry or a condition whose context ended the loop, and a condition that never held within its bound. Attempts counts the calls that ran, and Last is the last failure or the context cause, absent when a condition simply never held.

It is one type rather than one per policy so that errors.Is(err, ErrGaveUp) is the whole of the question a caller has to ask. The detail stays reachable: Unwrap yields both ErrGaveUp and Last, so a caller that wants the deadline specifically still finds it.

func (e *RetryError) Error() string

func (e *RetryError) Unwrap() []error

type State ¶

type State int

State is a circuit breaker position.

const (
	// Closed admits every call and counts outcomes toward tripping.
	Closed State = iota
	// Open rejects every call with ErrOpen until OpenFor has elapsed.
	Open
	// HalfOpen admits up to MaxProbes concurrent calls to test recovery: any
	// failure reopens, SuccessesToClose successes close.
	HalfOpen
)

func (s State) String() string

type Timer ¶

type Timer interface {
	Stop() bool
}

Timer is a pending AfterFunc. Stop reports whether it prevented f from running.

type Trip ¶

type Trip func(Counts) bool

Trip decides from the current Counts whether a Closed breaker should open. It is consulted after every failure while Closed.

func ConsecutiveFailures(n int) Trip

func FailureRatio(ratio float64, minRequests int) Trip

type UntilPolicy ¶ added in v0.5.0

type UntilPolicy struct {
	// MaxAttempts is the bound when positive. Zero means the caller's context
	// is the bound, which for a condition is usually the natural one: polling
	// until a deadline does not have a meaningful attempt count, and deriving
	// one is how a poll ends up quietly short of the budget it was given.
	MaxAttempts int
	Backoff     Backoff
	Clock       Clock
}

UntilPolicy configures RetryUntil. It is a separate type from Policy because a condition has no retryable-error notion to classify: a probe reports whether the condition held, and an error from it is a reason to stop. Reusing Policy would have meant a Retryable field that RetryUntil silently ignored, which is a field set in good faith and obeyed by nobody.