Documentation
¶
Index ¶
- Variables
- func Do[T any](ctx context.Context, action func(context.Context) (T, error), opts ...Option) (T, error)
- func DoErr(ctx context.Context, action func(context.Context) error, opts ...Option) error
- func DoErrHedged(ctx context.Context, action func(context.Context) error, opts ...Option) error
- func DoErrState(ctx context.Context, action func(context.Context, RetryState) error, ...) error
- func DoErrStateHedged(ctx context.Context, action func(context.Context, RetryState) error, ...) error
- func DoHedged[T any](ctx context.Context, action func(context.Context) (T, error), opts ...Option) (T, error)
- func DoState[T any](ctx context.Context, action func(context.Context, RetryState) (T, error), ...) (T, error)
- func DoStateHedged[T any](ctx context.Context, action func(context.Context, RetryState) (T, error), ...) (T, error)
- func FatalError(err error) error
- func InjectDeadlineHeader(ctx context.Context, header Header, key string)
- func WithPriority(ctx context.Context, p Priority) context.Context
- func WithTestingBypass(ctx context.Context) context.Context
- type AdaptiveBucket
- type AdaptiveLimiter
- type Backoff
- type Bulkhead
- type Config
- func (c *Config) Do(ctx context.Context, action func(context.Context) (any, error)) (any, error)
- func (c *Config) DoErr(ctx context.Context, action func(context.Context) error) error
- func (c *Config) DoErrHedged(ctx context.Context, action func(context.Context) error) error
- func (c *Config) DoHedged(ctx context.Context, action func(context.Context) (any, error)) (any, error)
- type Header
- type HealthState
- type Instrumenter
- type Option
- func WithAdaptiveBucket(bucket *AdaptiveBucket) Option
- func WithAdaptiveLimiter() Option
- func WithAdaptiveLimiterInstance(al *AdaptiveLimiter) Option
- func WithBackoff(backoff Backoff) Option
- func WithBaseDelay(delay time.Duration) Option
- func WithBulkhead(capacity uint) Option
- func WithBulkheadInstance(bh *Bulkhead) Option
- func WithChaos(cfg chaos.Config) Option
- func WithCircuitBreaker(cb *circuit.Breaker) Option
- func WithFallback[T any](f func(context.Context, error) (T, error)) Option
- func WithFallbackErr(f func(context.Context, error) error) Option
- func WithHedgingDelay(delay time.Duration) Option
- func WithInstrumenter(instr Instrumenter) Option
- func WithMaxAttempts(attempts uint) Option
- func WithMaxDelay(delay time.Duration) Option
- func WithMinDeadlineThreshold(threshold time.Duration) Option
- func WithName(name string) Option
- func WithPanicRecovery() Option
- func WithPriorityBulkhead(capacity uint, thresholds map[Priority]float64) Option
- func WithPriorityBulkheadInstance(bh *PriorityBulkhead) Option
- func WithRateLimiter(limit float64, interval time.Duration) Option
- func WithRateLimiterInstance(rl *RateLimiter) Option
- func WithRetry(attempts uint) Option
- func WithRetryIf(target error) Option
- func WithRetryIfFunc(f func(error) bool) Option
- func WithTimeout(timeout time.Duration) Option
- type PanicError
- type Policy
- type Priority
- type PriorityBulkhead
- type RateLimiter
- type RetryAfterError
- type RetryState
- type Retryer
- type StateEvent
- type StateMachine
- type TransitionFunc
Constants ¶
This section is empty.
Variables ¶
var ErrBulkheadFull = errors.New("bulkhead capacity reached")
ErrBulkheadFull is returned when the bulkhead capacity is reached.
var ErrRateLimitExceeded = errors.New("rate limit exceeded")
ErrRateLimitExceeded is returned when the rate limit is exceeded.
var ErrShedLoad = errors.New("load shed: limit reached")
ErrShedLoad is returned when the adaptive concurrency or priority bulkhead limit is reached.
Functions ¶
func Do ¶
func Do[T any](ctx context.Context, action func(context.Context) (T, error), opts ...Option) (T, error)
Do executes an action with retry logic using the provided options. This generic function handles functions returning (T, error).
func DoErr ¶
DoErr executes an action with retry logic using the provided options. This function handles functions returning only error.
func DoErrHedged ¶
DoErrHedged executes an action using speculative retries (hedging).
func DoErrState ¶
func DoErrState(ctx context.Context, action func(context.Context, RetryState) error, opts ...Option) error
DoErrState executes a stateful action with retry logic using the provided options.
func DoErrStateHedged ¶
func DoErrStateHedged(ctx context.Context, action func(context.Context, RetryState) error, opts ...Option) error
DoErrStateHedged executes a stateful action with speculative retries (hedging).
func DoHedged ¶
func DoHedged[T any](ctx context.Context, action func(context.Context) (T, error), opts ...Option) (T, error)
DoHedged executes an action using speculative retries (hedging). It starts multiple attempts concurrently if previous ones take too long.
func DoState ¶
func DoState[T any](ctx context.Context, action func(context.Context, RetryState) (T, error), opts ...Option) (T, error)
DoState executes a stateful action with retry logic using the provided options. The RetryState is passed to the closure, allowing it to adapt to failure history.
func DoStateHedged ¶
func DoStateHedged[T any](ctx context.Context, action func(context.Context, RetryState) (T, error), opts ...Option) (T, error)
DoStateHedged executes a stateful action with speculative retries (hedging).
func FatalError ¶
FatalError wraps an error to indicate that the retry loop should terminate immediately.
func InjectDeadlineHeader ¶ added in v1.0.5
InjectDeadlineHeader calculates the remaining time in the context and injects it into the provided header. This is useful for distributed deadline propagation. If the key is "Grpc-Timeout", it uses the gRPC-specific timeout format. Otherwise, it injects the remaining milliseconds as a string.
func WithPriority ¶ added in v1.0.6
WithPriority attaches a Priority to the context.
Types ¶
type AdaptiveBucket ¶
type AdaptiveBucket struct {
// contains filtered or unexported fields
}
AdaptiveBucket implements a client-side token bucket rate limiter for retries. It protects downstream services from thundering herds by depleting tokens when retries occur and refilling them on successful responses. This allows a fleet of clients to quickly cut off traffic to a degraded system, providing macro-level protection.
func DefaultAdaptiveBucket ¶
func DefaultAdaptiveBucket() *AdaptiveBucket
DefaultAdaptiveBucket creates a new AdaptiveBucket with standard defaults (max capacity: 500, retry cost: 5, success refill: 1).
func NewAdaptiveBucket ¶
func NewAdaptiveBucket(maxCapacity, retryCost, successRefill float64) *AdaptiveBucket
NewAdaptiveBucket creates a new AdaptiveBucket with the specified configuration. A common AWS-style configuration is maxCapacity=500, retryCost=5, successRefill=1.
func (*AdaptiveBucket) AcquireRetryToken ¶
func (b *AdaptiveBucket) AcquireRetryToken() bool
AcquireRetryToken attempts to consume a token for a retry. Returns true if a token was successfully consumed, false if the bucket is empty.
func (*AdaptiveBucket) AddSuccessToken ¶
func (b *AdaptiveBucket) AddSuccessToken()
AddSuccessToken adds a fraction of a token back to the bucket upon a successful response.
type AdaptiveLimiter ¶ added in v1.0.4
type AdaptiveLimiter struct {
// contains filtered or unexported fields
}
AdaptiveLimiter implements a dynamic concurrency limiter using Additive Increase, Multiplicative Decrease (AIMD) based on Round Trip Time (RTT), applying Little's Law.
func NewAdaptiveLimiter ¶ added in v1.0.4
func NewAdaptiveLimiter() *AdaptiveLimiter
NewAdaptiveLimiter creates a new AdaptiveLimiter with default configurations.
func (*AdaptiveLimiter) Execute ¶ added in v1.0.4
func (al *AdaptiveLimiter) Execute(ctx context.Context, action func() error) error
Execute wraps an action with adaptive concurrency control.
func (*AdaptiveLimiter) GetMaxConcurrency ¶ added in v1.0.4
func (al *AdaptiveLimiter) GetMaxConcurrency() int32
GetMaxConcurrency returns the current maximum concurrency limit.
func (*AdaptiveLimiter) Health ¶ added in v1.0.6
func (al *AdaptiveLimiter) Health() <-chan StateEvent
Health returns a channel that emits state change events. The channel is created lazily on first access. The caller should use a select with default to handle non-blocking receive.
type Backoff ¶
type Backoff interface {
// Next calculates the duration to wait before the specified attempt.
// The attempt parameter is 0-indexed.
Next(attempt uint) time.Duration
}
Backoff defines the interface for temporal distribution of retries.
func NewFullJitter ¶
NewFullJitter returns a Backoff implementation using the AWS Full Jitter algorithm. The base duration dictates the initial delay, and the cap defines the absolute maximum.
type Bulkhead ¶ added in v1.0.3
type Bulkhead struct {
// contains filtered or unexported fields
}
Bulkhead limits the number of concurrent executions to prevent overloading a resource.
func NewBulkhead ¶ added in v1.0.3
NewBulkhead creates a new Bulkhead with the specified capacity.
type Config ¶
type Config struct {
Name string
MaxAttempts uint
BaseDelay time.Duration
MaxDelay time.Duration
HedgingDelay time.Duration
Backoff Backoff
Policy *retryPolicy
Instrumenter Instrumenter
CircuitBreaker *circuit.Breaker
Fallback any
AdaptiveBucket *AdaptiveBucket
RecoverPanics bool
Bulkhead *Bulkhead
PriorityBulkhead *PriorityBulkhead
Timeout time.Duration
RateLimiter *RateLimiter
AdaptiveLimiter *AdaptiveLimiter
Chaos *chaos.Injector
MinDeadlineThreshold time.Duration
// contains filtered or unexported fields
}
Config represents the configuration for the retry execution.
func DefaultConfig ¶
func DefaultConfig() *Config
DefaultConfig returns a reasonable production-grade configuration.
func (*Config) DoErrHedged ¶
DoErrHedged satisfies the Retryer interface.
type Header ¶ added in v1.0.5
type Header interface {
Set(key, value string)
}
Header defines an interface for injecting metadata into outgoing requests. This allows Resile to remain agnostic of the underlying transport (HTTP, gRPC, etc.).
type HealthState ¶ added in v1.0.6
type HealthState int
HealthState represents the health state of a resilience component.
const ( // HealthStateHealthy represents a healthy state where requests are allowed. HealthStateHealthy HealthState = iota // HealthStateDegraded represents a degraded but functional state. HealthStateDegraded // HealthStateUnhealthy represents an unhealthy state (e.g., circuit open). HealthStateUnhealthy )
type Instrumenter ¶
type Instrumenter interface {
// BeforeAttempt is called before each execution attempt.
// It can return a new context (e.g., to inject trace spans).
BeforeAttempt(ctx context.Context, state RetryState) context.Context
// AfterAttempt is called after each execution attempt.
AfterAttempt(ctx context.Context, state RetryState)
// OnBulkheadFull is called when the bulkhead capacity is reached.
OnBulkheadFull(ctx context.Context, state RetryState)
// OnRateLimitExceeded is called when the rate limit is exceeded.
OnRateLimitExceeded(ctx context.Context, state RetryState)
}
Instrumenter defines the lifecycle hooks for monitoring retry executions. It is a zero-dependency interface to allow custom implementations for logging, metrics, and tracing.
type Option ¶
type Option func(*Config)
Option defines a functional option for configuring a retry execution.
func WithAdaptiveBucket ¶
func WithAdaptiveBucket(bucket *AdaptiveBucket) Option
WithAdaptiveBucket sets a token bucket for adaptive retries. The bucket should be shared across multiple executions to protect downstream services globally.
func WithAdaptiveLimiter ¶ added in v1.0.4
func WithAdaptiveLimiter() Option
WithAdaptiveLimiter integrates an adaptive concurrency limiter into the execution.
func WithAdaptiveLimiterInstance ¶ added in v1.0.4
func WithAdaptiveLimiterInstance(al *AdaptiveLimiter) Option
WithAdaptiveLimiterInstance integrates a shared adaptive concurrency limiter into the execution.
func WithBackoff ¶
WithBackoff sets a custom backoff algorithm.
func WithBaseDelay ¶
WithBaseDelay sets the initial delay for the backoff algorithm.
func WithBulkhead ¶ added in v1.0.3
WithBulkhead integrates a bulkhead into the execution.
func WithBulkheadInstance ¶ added in v1.0.3
WithBulkheadInstance integrates a shared bulkhead into the execution.
func WithCircuitBreaker ¶
WithCircuitBreaker integrates a circuit breaker into the retry execution.
func WithFallback ¶
WithFallback sets a function to be called if all retries are exhausted or if the circuit breaker is open. T must match the return type of the retry action.
func WithFallbackErr ¶
WithFallbackErr sets a fallback function for operations that only return an error.
func WithHedgingDelay ¶
WithHedgingDelay sets the delay for speculative retries (hedging). If a response doesn't arrive within this delay, another attempt is started concurrently.
func WithInstrumenter ¶
func WithInstrumenter(instr Instrumenter) Option
WithInstrumenter sets a telemetry instrumenter.
func WithMaxAttempts ¶
WithMaxAttempts sets the maximum number of execution attempts.
func WithMaxDelay ¶
WithMaxDelay sets the maximum delay for the backoff algorithm.
func WithMinDeadlineThreshold ¶ added in v1.0.5
WithMinDeadlineThreshold sets the minimum remaining time required to start a new attempt. If the context's deadline is sooner than this threshold, the execution is aborted early.
func WithPanicRecovery ¶ added in v1.0.1
func WithPanicRecovery() Option
WithPanicRecovery enables recovering from panics during execution. If a panic occurs, it is converted into a PanicError and treated as a retryable error.
func WithPriorityBulkhead ¶ added in v1.0.6
WithPriorityBulkhead integrates a priority-aware bulkhead into the execution.
func WithPriorityBulkheadInstance ¶ added in v1.0.6
func WithPriorityBulkheadInstance(bh *PriorityBulkhead) Option
WithPriorityBulkheadInstance integrates a shared priority-aware bulkhead into the execution.
func WithRateLimiter ¶ added in v1.0.3
WithRateLimiter integrates a rate limiter into the execution.
func WithRateLimiterInstance ¶ added in v1.0.3
func WithRateLimiterInstance(rl *RateLimiter) Option
WithRateLimiterInstance integrates a shared rate limiter into the execution.
func WithRetry ¶ added in v1.0.3
WithRetry sets the maximum number of execution attempts and adds a retry policy to the pipeline.
func WithRetryIf ¶
WithRetryIf sets a specific error to trigger a retry.
func WithRetryIfFunc ¶
WithRetryIfFunc sets a custom function to determine if an error should be retried.
func WithTimeout ¶ added in v1.0.3
WithTimeout sets a timeout for the execution.
type PanicError ¶ added in v1.0.1
PanicError represents a recovered panic during execution.
func (*PanicError) Error ¶ added in v1.0.1
func (p *PanicError) Error() string
Error implements the error interface.
type Policy ¶ added in v1.0.3
type Policy struct {
// contains filtered or unexported fields
}
Policy represents a composed resilience strategy. It is thread-safe and reusable across multiple calls.
func NewPolicy ¶ added in v1.0.3
NewPolicy creates a new Policy with the given options. The order of options determines the execution order from outermost to innermost. Example: NewPolicy(WithFallback(f), WithRetry(3)) results in Fallback(Retry(Action)).
func (*Policy) Health ¶ added in v1.0.6
func (p *Policy) Health() <-chan circuit.StateEvent
Health returns a channel that emits state change events from the underlying resilience components. It returns nil if no health-reporting component is configured. The caller should use a select with default to handle non-blocking receive. Note: Only circuit breaker health events are exposed through Policy.Health(). For adaptive limiter events, access the limiter directly via AdaptiveLimiter.Health().
type Priority ¶ added in v1.0.6
type Priority int
Priority represents the importance of a request.
func GetPriority ¶ added in v1.0.6
GetPriority retrieves the Priority from the context. Returns PriorityStandard if no priority is found.
type PriorityBulkhead ¶ added in v1.0.6
type PriorityBulkhead struct {
// contains filtered or unexported fields
}
PriorityBulkhead limits concurrency based on priority levels. It preserves capacity for higher-priority traffic by shedding lower-priority traffic when the system approaches saturation.
func NewPriorityBulkhead ¶ added in v1.0.6
func NewPriorityBulkhead(capacity uint, thresholds map[Priority]float64) *PriorityBulkhead
NewPriorityBulkhead creates a new PriorityBulkhead with the specified capacity and thresholds. The thresholds map defines the maximum utilization (0.0 to 1.0) allowed for each priority. If a priority is not in the map, it defaults to 1.0 (no shedding until full capacity).
func (*PriorityBulkhead) Execute ¶ added in v1.0.6
func (b *PriorityBulkhead) Execute(ctx context.Context, action func() error) error
Execute wraps an action with priority-aware bulkhead concurrency control. It returns ErrShedLoad if the utilization threshold for the priority is exceeded. It returns ErrBulkheadFull if the absolute capacity is reached.
Performance Note: This implementation uses a "soft limit" design where the utilization check is decoupled from the actual semaphore acquisition. This provides high throughput and avoids global locks, which is ideal for load shedding. While this might allow slight over-utilization under extremely high contention, it maintains high performance for the majority of cases.
type RateLimiter ¶ added in v1.0.3
type RateLimiter struct {
// contains filtered or unexported fields
}
RateLimiter implements a time-based token bucket rate limiter.
func NewRateLimiter ¶ added in v1.0.3
func NewRateLimiter(limit float64, interval time.Duration) *RateLimiter
NewRateLimiter creates a new RateLimiter with the specified limit and interval. Example: NewRateLimiter(100, time.Second) allows 100 requests per second.
type RetryAfterError ¶
RetryAfterError is implemented by errors that specify how long to wait before retrying. This is commonly used with HTTP 429 (Too Many Requests) or 503 (Service Unavailable) to respect Retry-After headers. It also supports pushback signals to cancel all retries.
type RetryState ¶
type RetryState struct {
// Name is the optional identifier for the operation being retried.
Name string
// Attempt is the current 0-indexed retry iteration.
Attempt uint
// MaxAttempts is the maximum number of attempts allowed.
MaxAttempts uint
// LastError is the error encountered in the previous attempt.
LastError error
// TotalDuration is the cumulative time spent across all attempts and sleeps.
TotalDuration time.Duration
// NextDelay is the duration to be slept before the next attempt.
NextDelay time.Duration
// contains filtered or unexported fields
}
RetryState encapsulates the current state of a retry execution.
type Retryer ¶
type Retryer interface {
// Do executes a function that returns a value and an error.
Do(ctx context.Context, action func(context.Context) (any, error)) (any, error)
// DoHedged executes a function using speculative retries (hedging).
DoHedged(ctx context.Context, action func(context.Context) (any, error)) (any, error)
// DoErr executes a function that returns only an error.
DoErr(ctx context.Context, action func(context.Context) error) error
// DoErrHedged executes a function using speculative retries (hedging).
DoErrHedged(ctx context.Context, action func(context.Context) error) error
}
Retryer defines the interface for executing actions with resilience.
type StateEvent ¶ added in v1.0.6
type StateEvent struct {
Component string
State HealthState
Timestamp time.Time
Message string
}
StateEvent represents a state change event emitted by resilience components.
type StateMachine ¶ added in v1.0.2
StateMachine represents a resilient state machine inspired by Erlang's gen_statem. Every state transition is protected by the configured resilience policies.
func NewStateMachine ¶ added in v1.0.2
func NewStateMachine[S any, D any, E any](initialState S, initialData D, transition TransitionFunc[S, D, E], opts ...Option) *StateMachine[S, D, E]
NewStateMachine creates a new resilient state machine with the provided initial state, data, and transition function.
func (*StateMachine[S, D, E]) GetData ¶ added in v1.0.2
func (sm *StateMachine[S, D, E]) GetData() D
GetData returns the current data of the state machine.
func (*StateMachine[S, D, E]) GetState ¶ added in v1.0.2
func (sm *StateMachine[S, D, E]) GetState() S
GetState returns the current state of the state machine.
func (*StateMachine[S, D, E]) Handle ¶ added in v1.0.2
func (sm *StateMachine[S, D, E]) Handle(ctx context.Context, event E) error
Handle processes an event and performs a state transition. The transition is executed within a resilience envelope (Retries, Circuit Breakers, etc.). If the transition succeeds, the state machine updates its internal state and data.
type TransitionFunc ¶ added in v1.0.2
type TransitionFunc[S any, D any, E any] func(ctx context.Context, state S, data D, event E, rs RetryState) (S, D, error)
TransitionFunc defines the signature for a state transition function. It takes the current state, data, and event, and returns the next state and data. It also receives the current RetryState to allow transitions to adapt to failure history.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
examples
|
|
|
adaptiveconcurrency
command
|
|
|
adaptiveretry
command
|
|
|
backpressure
command
|
|
|
basic
command
|
|
|
bulkhead
command
|
|
|
chaos
command
|
|
|
circuitbreaker
command
|
|
|
deadline
command
|
|
|
fallback
command
|
|
|
hedging
command
|
|
|
http
command
|
|
|
http_resume_stream
command
|
|
|
panicrecovery
command
|
|
|
prioritybulkhead
command
|
|
|
pushback
command
|
|
|
ratelimiter
command
|
|
|
sql
command
|
|
|
stateful
command
|
|
|
statemachine
command
|
|
|
telemetry
|
|
