resiliency

Documentation

Index

• Constants
• func Join(errs ...error) error
• func MakePanicError(panicVal any, log logr.Logger) error
• func Permanent(err error) error
• func Retry(ctx context.Context, b backoff.BackOff, operation func() error) error
• func RetryExponential(ctx context.Context, operation func() error) error
• func RetryExponentialWithTimeout(ctx context.Context, timeout time.Duration, operation func() error) error
• func RetryGet[T any](ctx context.Context, b backoff.BackOff, factory func() (T, error)) (T, error)
• func RetryGetExponential[T any](ctx context.Context, factory func() (T, error)) (T, error)
• func RunWithTimeout(op func(), timeout time.Duration) bool
• type AsyncWorker
  • func NewAsyncWorker[InputT, OutputT any](lifetimeCtx context.Context, getResult func(context.Context, InputT) OutputT, ...) *AsyncWorker[InputT, OutputT]
  • func (aw *AsyncWorker[InputT, OutputT]) Enqueue(input InputT) error
  • func (aw *AsyncWorker[InputT, OutputT]) Results() <-chan OutputT
• type DebounceLast
  • func NewDebounceLast[T any, R any, RF Runner[T, R]](runner RF, delay, maxDelay time.Duration) *DebounceLast[T, R, RF]
  • func (dl *DebounceLast[T, R, RF]) Run(ctx context.Context, arg T) (R, error)
• type DebounceLastAction
  • func NewDebounceLastAction[T any](action func(T), delay, maxDelay time.Duration) *DebounceLastAction[T]
  • func (dl *DebounceLastAction[T]) Run(ctx context.Context, arg T)
• type ResultWithError
• type Runner
• type WorkQueue
  • func NewWorkQueue(lifetimeCtx context.Context, maxConcurrency uint8) *WorkQueue
  • func (wq *WorkQueue) Enqueue(work WorkQueueItem) error
• type WorkQueueItem

Constants

const DefaultConcurrency uint8 = 0

Functions

func Join

func Join(errs ...error) error

Joins multiple errors into a single error, accounting for permanent errors.

func MakePanicError

func MakePanicError(panicVal any, log logr.Logger) error

Logs a panic value and associated call stack and returns it as an error.

func Permanent

func Permanent(err error) error

Creates a permanent error that stops the retry loop.

func Retry

func Retry(ctx context.Context, b backoff.BackOff, operation func() error) error

Try calling operation function with exponential back-off until it succeeds, or a permanent error occurs, or the passed context is cancelled.

func RetryExponential

func RetryExponential(ctx context.Context, operation func() error) error

Try calling operation function with exponential back-off until either: - the operation succeeds, or - a permanent error occurs, or - passed context is cancelled.

func RetryExponentialWithTimeout

func RetryExponentialWithTimeout(ctx context.Context, timeout time.Duration, operation func() error) error

Try calling operation function with exponential back-off until either: - the operation succeeds, or - a permanent error occurs, or - passed context is cancelled, or - timeout is reached.

func RetryGet

func RetryGet[T any](ctx context.Context, b backoff.BackOff, factory func() (T, error)) (T, error)

Try calling factory function with given backoff policy until a value is successfully created, or a permanent error occurs, or the passed context is cancelled.

func RetryGetExponential

func RetryGetExponential[T any](ctx context.Context, factory func() (T, error)) (T, error)

Try calling factory function with exponential back-off until either: - a value is successfully created, or - a permanent error occurs, or - passed context is cancelled.

func RunWithTimeout

func RunWithTimeout(op func(), timeout time.Duration) bool

Runs a function and returns when the function returns or when the specified timeout is reached. Returns true if the function returned before the timeout, and false if the timeout was reached. Note: this should not be used in a tight loop as each invocation creates a goroutine and a timer.

Types

type AsyncWorker

type AsyncWorker[InputT, OutputT any] struct {
	// contains filtered or unexported fields
}

Async worker is a worker that runs asynchronously, processing items from a work queue and returning results through a channel, with limited concurrency.

func NewAsyncWorker

func NewAsyncWorker[InputT, OutputT any](
	lifetimeCtx context.Context,
	getResult func(context.Context, InputT) OutputT,
	maxConcurrency uint8,
) *AsyncWorker[InputT, OutputT]

func (*AsyncWorker[InputT, OutputT]) Enqueue

func (aw *AsyncWorker[InputT, OutputT]) Enqueue(input InputT) error

func (*AsyncWorker[InputT, OutputT]) Results

func (aw *AsyncWorker[InputT, OutputT]) Results() <-chan OutputT

type DebounceLast

type DebounceLast[T any, R any, RF Runner[T, R]] struct {
	// contains filtered or unexported fields
}

DebounceLast calls the runner function after the specified delay, but only if no new calls have arrived in the meantime. If new calls arrive, the runner will be delayed further, but no more than maxDelay. After the runner function completes, the callers of Run() will all receive the same result (and error, if any).

func NewDebounceLast

func NewDebounceLast[T any, R any, RF Runner[T, R]](runner RF, delay, maxDelay time.Duration) *DebounceLast[T, R, RF]

func (*DebounceLast[T, R, RF]) Run

func (dl *DebounceLast[T, R, RF]) Run(ctx context.Context, arg T) (R, error)

type DebounceLastAction

type DebounceLastAction[T any] struct {
	// contains filtered or unexported fields
}

A variant of DebounceLast that calls an "action" (function with no return value). Because there is no return value, the Run method does not wait for the action to complete (the action is executed in a separate goroutine, fully asynchronously). The action will be called after the specified delay, but only if no new calls arrive in the meantime. If new calls arrive, the action will be delayed further, but no more than maxDelay.

func NewDebounceLastAction

func NewDebounceLastAction[T any](action func(T), delay, maxDelay time.Duration) *DebounceLastAction[T]

func (*DebounceLastAction[T]) Run

func (dl *DebounceLastAction[T]) Run(ctx context.Context, arg T)

type ResultWithError

type ResultWithError[R any] struct {
	V   R
	Err error
}

type Runner

type Runner[T any, R any] interface {
	~func(T) (R, error)
}

type WorkQueue

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

WorkQueue runs work concurrently, but limits the number of concurrent executions.

func NewWorkQueue

func NewWorkQueue(lifetimeCtx context.Context, maxConcurrency uint8) *WorkQueue

func (*WorkQueue) Enqueue

func (wq *WorkQueue) Enqueue(work WorkQueueItem) error

Queues a work item to be executed. If the lifetime context is done, returns an error. The enqueue operation involves an (unbounded) channel write, so it may block, but for a very short time. On the other hand, the channel write acts as a memory barrier, so any writes to the work queue item parameters will be completed by the time the work item is executed.

type WorkQueueItem

type WorkQueueItem = func(ctx context.Context)