retrier

README

Retrier

Retrier is a Go library for building resilient, scalable task processing systems. It provides:

• A dynamic worker pool that scales up/down based on load.
• Pluggable backoff strategies (linear, exponential, with/without jitter) for retries.
• An orchestration manager that routes tasks to named workers, aggregates results, and buffers them when storage is unavailable.
• Graceful lifecycle controls (start, stop, suspend) with context propagation.
• Circuit Breaker with sliding window failure rate detection to protect workers from cascading failures.
• Event publishing – receive real-time notifications for every completed task execution.
• Enhanced status – query worker statuses including circuit breaker state.
• Task deadlines – tasks that exceed their deadline are automatically marked as failed and are not retried.
• Context support – WorkerFn now receives a context.Context, enabling distributed tracing, cancellation, and deadline propagation.

---

Features

• Automatic worker scaling – configurable min/max goroutines, idle timeouts for scale‑down.
• Retry policies – built‑in linear, exponential, and jitter‑based backoff; custom strategies can be registered.
• Multi‑worker manager – each task type can have its own worker pool, all coordinated by a Manager.
• Result streaming – consume execution outcomes via channels for real‑time monitoring.
• Storage buffering – if the persistent store (Store) fails, results are kept in memory and flushed periodically.
• Circuit Breaker – sliding window failure‑rate based protection. Automatically opens when failure ratio exceeds a threshold, and allows probing after a timeout.
• Event Publisher – implement EventPublisher to receive every task result as soon as it is processed (e.g., for notifications, metrics, or logging).
• Comprehensive status – GetWorkerStatuses() now returns FullWorkerState which includes the circuit breaker state for each worker.
• Task Deadlines – each task can have a Deadline; if the deadline passes before execution, the task is marked as failed with a critical error and never retried.
• Context support – WorkerFn receives a context.Context, enabling distributed tracing, cancellation, and deadline propagation.
• Thread‑safe – all exported methods are safe for concurrent use.

---

Installation

go get github.com/devian2011/retrier

---

Core Concepts

1. Task

Represents a unit of work. It includes:

• ID (UUID) – unique identifier.
• Payload (byte slice) – the actual data to process.
• Worker – the named worker that should process this task.
• Status – current lifecycle state (pending, success, failure, etc.).
• Retries / MaxRetries – attempt counter and limit.
• BackOffCode and BackOffParams – retry scheduling configuration.
• Deadline (optional) – absolute time by which the task must complete. If the deadline is in the past when the task is picked up, it is immediately marked as failed with a critical state and will not be retried.
• Timestamps (CreatedAt, LastRun, NextRun) – for scheduling and monitoring.

2. TaskExecutionResult

Records the outcome of a single execution attempt. Contains:

• ID (UUID) – unique result record identifier.
• TaskID – reference to the parent Task.
• Status – success or failure of this specific attempt.
• RunAt – timestamp of the execution.
• Result ([]byte) – the output or error message from the worker.
• IsCritical – if true, the task is permanently failed and won't be retried.
• ExecutionTime – duration of the worker function call.

3. Worker

A pool of goroutines that executes a user‑defined WorkerFn.
It scales up when all workers are busy (up to maxWorkers) and scales down idle workers (down to minWorkers) after idleTimeout.
Tasks are submitted via Submit(), results are emitted on a channel obtained via GetOutChan().

Context support: The WorkerFn now receives a context.Context as its first argument. This allows you to:

• Propagate trace IDs for distributed tracing (e.g., OpenTelemetry).
• Handle cancellations or deadlines that were set on the task.
• Pass request-scoped values (e.g., user ID, authentication tokens).

4. Manager

Orchestrates multiple workers registered by name. It:

• Routes tasks to the correct worker (by task.Worker).
• Aggregates results from all workers into a central channel.
• Calls Store.SaveTask() to persist each result.
• Buffers results in memory if storage fails, and periodically retries flushing.
• Integrates with Circuit Breakers per worker to prevent overloading failing services.
• Publishes every completed result via an optional EventPublisher.
• Automatically checks task deadlines before submitting to the worker; expired tasks are marked as failed and are never retried.

5. BackOffStrategy

A registry of functions that compute the next execution time (time.Time) based on a BackOffParams object (which Task implements).
Built‑in strategies:

• linear – fixed delay (duration)
• jitter-linear – delay + random jitter (default 0.4)
• exponential – baseDelay * multiplier^(retries-1), capped at maxDelay
• jitter-exponential – same as exponential with additional jitter (default 0.2)

6. Circuit Breaker

A sliding‑window circuit breaker that protects a worker from repeated failures. It tracks success/failure counts over a configurable time window and opens the circuit when the failure rate exceeds a given threshold. While open, requests are rejected (skipped) for that worker. After a timeout, the breaker transitions to HalfOpen to allow a single probe request; if it succeeds, the circuit closes; otherwise it re‑opens.

The Manager uses a Breaker interface, so you can plug in any implementation. The default implementation uses a sliding window with configurable window size, failure threshold, minimum request count, and open‑state timeout.

7. EventPublisher

An optional interface that allows you to receive real‑time notifications for every completed task execution.

type EventPublisher interface {
    Publish(event WorkerExecutionResult)
}

If you pass a non‑nil publisher to NewManager, the Manager will call Publish after persisting the result (but before removing it from the “in‑flight” set). This can be used for:

• Streaming results to a message queue (e.g., Kafka, RabbitMQ).
• Sending webhooks or notifications.
• Feeding metrics and monitoring systems.
• Triggering downstream workflows.

8. Task Deadlines

You can set a Deadline on a Task – an absolute time by which the task must be processed. When the manager periodically fetches tasks, it compares the current time with the deadline. If the deadline has already passed (i.e., task.Deadline.Before(time.Now())), the task is never submitted to the worker. Instead, it is marked as StatusFailure with a critical error and will not be retried (even if MaxRetries is not exhausted).

This is useful for time‑sensitive operations such as user requests that have a timeout, or batch jobs that must finish by a certain time.

The deadline is stored in the Task struct and is persisted along with other fields via the Store interface.

type Task struct {
    // ...
    Deadline time.Time `json:"deadline"`
}

Example: if you create a task with a deadline 30 seconds from now, and the system is heavily backlogged, the task will be automatically dropped (marked as failed) once that deadline passes, preventing wasted work and helping maintain SLAs.

9. Context Support

The WorkerFn signature now includes a context.Context as the first parameter:

type WorkerFn func(ctx context.Context, payload []byte) (string, *ExecutionError)

When a task is submitted, you can attach a context to it (e.g., through a separate field or via a wrapper). The worker will pass this context to your function, enabling:

• Distributed tracing – create spans inside your task using OpenTelemetry.
• Cancellation – cancel long‑running tasks if needed.
• Deadline propagation – if the context carries a deadline, it will be respected.

If no context is attached to the task, the worker uses its own background context.

---

Persistence – Store Interface

The Manager relies on a Store interface to persist tasks and results. To implement your own storage backend, you need to understand what fields are used for retry scheduling.

type Store interface {
    // GetTasks must return tasks that are eligible for execution:
    //   - Status is either "pending" or "suspended"
    //   - NextRun is in the past (or zero)
    GetTasks() ([]Task, error)

    // SaveTask persists a task and its execution result.
    // If result is nil, it means the task is being saved for the first time (submit).
    SaveTask(task *Task, result *TaskExecutionResult) error
}

What to store for retry functionality?

When you store a Task, you must preserve at least the following fields so that the Manager can correctly schedule retries:

• ID – to identify the task.
• Worker – which worker pool to route to.
• Status – pending, suspended, success, or failure.
• Retries – current attempt count.
• MaxRetries – maximum allowed attempts.
• BackOffCode and BackOffParams – used to compute NextRun.
• CreatedAt, LastRun, NextRun – scheduling timestamps.
• Payload – the actual data to be processed.
• Deadline – if set, the time by which the task must be completed.

Additionally, each TaskExecutionResult should be stored to keep a history of attempts. The SaveTask method receives both the updated task and the result of the latest attempt. The manager updates the task’s Retries, LastRun, NextRun, and Status before calling SaveTask. Therefore, your store should simply persist the given Task and TaskExecutionResult as is.

Important: The manager expects GetTasks() to return only tasks that are ready to be executed (i.e., Status is pending or suspended and NextRun is in the past). This is crucial for correct retry timing.

In-Memory Store (MemStore)

For convenience, the package provides an in‑memory implementation of the Store interface called MemStore. It is designed for testing and development, not for production. MemStore stores only tasks that are not finished (i.e., status is pending or suspended). Completed tasks (success or failure) are ignored, making it suitable for testing retry workflows without a real database.

It is thread‑safe and keeps a copy of the tasks in a slice. Each call to GetTasks() returns a fresh copy to avoid external modifications.

Example usage:

store := retrier.NewMemStore()
task := &retrier.Task{
    ID:     retrier.GetID(),
    Worker: "test-worker",
    Status: retrier.StatusPending,
}
_ = store.SaveTask(task, nil)

tasks, _ := store.GetTasks()
// tasks contains the pending task

When a task finishes (status becomes success or failure), it will be automatically skipped on subsequent SaveTask calls, so it won't be returned by GetTasks().

This store is especially useful in unit tests where you want to simulate the persistence layer without external dependencies.

---

Quick Start

Creating a standalone worker

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/devian2011/retrier"
)

func main() {
	// Define your processing function (now with context).
	workerFn := func(ctx context.Context, payload []byte) (string, *retrier.ExecutionError) {
		// You can use ctx for tracing, cancellation, deadlines, etc.
		// Do some work...
		result := fmt.Sprintf("processed: %s", string(payload))
		return result, nil
	}

	ctx := context.Background()
	w, err := retrier.NewWorker(ctx, workerFn)
	if err != nil {
		panic(err)
    }
	w.SetMinAndMaxWorkers(2, 10)
	w.SetIdleTimeout(5 * time.Second)
	w.Start()
	defer w.Stop()

	task := retrier.Task{
		ID:      retrier.GetID(),
		Worker:  "default",
		Payload: []byte("hello world"),
	}

	if err := w.Submit(task); err != nil {
		log.Fatal(err)
	}

	// Read the result.
	select {
	case res := <-w.GetOutChan():
		fmt.Printf("Task %s finished with status %s\n", res.Task.ID, res.Result.Status)
	case <-time.After(2 * time.Second):
		log.Fatal("timeout")
	}
}

Using the Manager with multiple workers, circuit breakers, and event publisher

package main

import (
	"context"
	"log"
	"time"

	"github.com/devian2011/retrier"
)

// Implement your own Store, Logger, and EventPublisher.
type myStore struct{}
func (s myStore) GetTasks() ([]retrier.Task, error) { /* ... */ }
func (s myStore) SaveTask(task *retrier.Task, result *retrier.TaskExecutionResult) error { /* ... */ }

type myLogger struct{}
func (l myLogger) Infof(format string, args ...interface{}) { /* ... */ }
func (l myLogger) Errorf(format string, args ...interface{}) { /* ... */ }

type myEventPublisher struct{}
func (p myEventPublisher) Publish(event retrier.WorkerExecutionResult) {
    // e.g., send to Kafka, emit metrics, etc.
    log.Printf("Task %s finished with status %s", event.Task.ID, event.Result.Status)
}

func main() {
	ctx := context.Background()
	store := myStore{}
	logger := myLogger{}
	publisher := myEventPublisher{}

	manager := retrier.NewManager(
        ctx,
        store,
        logger,
        retrier.NewBackOffStrategy(),
        1000,               // max buffer size
        5*time.Second,      // fetch interval
        publisher,          // can be nil if not needed
    )

	// Create a circuit breaker for workerA
	breakerA := retrier.NewCircuitBreaker(
		60*time.Second,  // window size
		0.5,             // failure threshold (50%)
		10,              // minimum requests in window
		30*time.Second,  // open-state timeout
	)

	workerA, errA := retrier.NewWorker(ctx, processA) // processA must be func(context.Context, []byte)...
	if errA != nil {
		panic(errA)
    }
	workerA.SetMinAndMaxWorkers(1, 3)
	manager.RegisterWorker("workerA", workerA, breakerA)

	// WorkerB without circuit breaker (pass nil)
	workerB, errB := retrier.NewWorker(ctx, processB)
	if errB != nil {
		panic(errB)
	}
	workerB.SetMinAndMaxWorkers(2, 5)
	manager.RegisterWorker("workerB", workerB, nil)

	manager.Start()
	defer manager.Stop()

	task := retrier.Task{
		ID:      retrier.GetID(),
		Worker:  "workerA",
		Payload: []byte("data"),
	}
	if err := manager.Submit(task); err != nil {
		log.Fatal(err)
	}
}

---

Circuit Breaker Configuration

The default circuit breaker implementation (CircuitBreaker) is configured with four parameters:

• windowSize – duration of the sliding window over which success/failure counts are tracked.
• failureThreshold – ratio (0.0 to 1.0) of failures to total requests that triggers the open state.
• minRequests – minimum number of requests in the window before the threshold is evaluated.
• timeout – duration the circuit stays open before transitioning to half‑open.

Example: NewCircuitBreaker(60*time.Second, 0.5, 10, 30*time.Second)

When a worker is registered with a breaker, the Manager will automatically check Allow() before submitting a task. If the breaker is open, the task is skipped (and will be retried later). After each submission attempt, RecordSuccess() or RecordFailure() is called to update the breaker’s statistics.

If you pass nil as the breaker, the manager will skip circuit‑breaker checks entirely.

---

Enhanced Status – FullWorkerState

The Manager now provides a richer status snapshot via GetWorkerStatuses(), which returns a map of worker names to FullWorkerState:

type FullWorkerState struct {
    Status        WorkerStatus        `json:"status"`
    ActiveTasks   int32               `json:"active_tasks"`
    ActiveWorkers int32               `json:"active_workers"`
    CBState       CircuitBreakerState `json:"cb_state"`
}

This includes both the worker’s own status (running, stopped, etc.) and the current state of its circuit breaker (closed, open, half-open). This helps you monitor system health and detect issues with specific workers.

---

Configuring Retries with Backoff

Set the BackOffCode and BackOffParams on a Task. For example, an exponential backoff with jitter:

task := retrier.Task{
	ID:           retrier.GetID(),
	Worker:       "myWorker",
	Payload:      []byte("important task"),
	Retries:      0,
	MaxRetries:   5,
	BackOffCode:  retrier.JitterExponentialBackoff,
	BackOffParams: map[retrier.BackOffParam]interface{}{
		retrier.BaseDelayKey:  1 * time.Second,
		retrier.MultiplierKey: 2.0,
		retrier.MaxDelayKey:   5 * time.Minute,
		retrier.JitterKey:     0.3,
	},
}

When a task fails with a UsualState error, the Manager (or your custom logic) can call BackOffStrategy.Get(task) to compute the next NextRun time and reschedule it.

Using Deadlines

You can add a deadline to a task to enforce a maximum processing window. If the task cannot be picked up before the deadline, it will be automatically marked as failed with a critical error.

task := retrier.Task{
	ID:           retrier.GetID(),
	Worker:       "myWorker",
	Payload:      []byte("time-sensitive job"),
	MaxRetries:   3,
	Deadline:     time.Now().Add(30 * time.Second), // must start within 30 seconds
}

If the task remains queued past this deadline, the manager will not submit it to the worker and will set its status to StatusFailure with a critical error, so it will never be retried.

Using Context

Your WorkerFn receives a context. You can attach values or use it for tracing:

workerFn := func(ctx context.Context, payload []byte) (string, *retrier.ExecutionError) {
    // Extract trace ID from context
    traceID := ctx.Value("trace_id")
    // Start a span using OpenTelemetry
    ctx, span := tracer.Start(ctx, "my-task")
    defer span.End()
    // ... your business logic ...
    return "ok", nil
}

---

Error Handling

Your WorkerFn returns a (string, *ExecutionError). The ExecutionError carries:

• Err – the underlying error.
• State – either CriticalState (permanent failure, no retry) or UsualState (transient, should retry).

Example:

workerFn := func(ctx context.Context, payload []byte) (string, *retrier.ExecutionError) {
	if string(payload) == "invalid" {
		return "", &retrier.ExecutionError{
			Err:   errors.New("validation failed"),
			State: retrier.CriticalState,
		}
	}
	// ... simulate network error
	return "", &retrier.ExecutionError{
		Err:   errors.New("timeout"),
		State: retrier.UsualState,
	}
}

---

Lifecycle Methods

Method	Description
Worker.Start()	Spawns the minimum number of workers and sets status to Running.
Worker.Suspend()	Changes status to Suspended; new submissions are rejected, but active tasks run to completion.
Worker.Stop()	Cancels the internal context, waits for all goroutines to finish, closes channels, status becomes Stopped.
Manager.Start()	Starts all registered workers, launches collector and splitter pipelines.
Manager.Stop()	Stops all workers, cancels manager context, waits for pipelines.

---

Storage Buffering in Manager

The Manager uses a Store interface as described above. If SaveTask returns an error, the result is appended to an in‑memory buffer (up to maxBufferSize). A background goroutine (diskBufferSwap) attempts to flush the buffer every minute. You can customise the buffer size and flush interval (currently fixed at 1 minute; you can change it by modifying the source if needed).

---

Custom Backoff Strategies

Register your own strategy with the global BackOffStrategy:

strategy := retrier.NewBackOffStrategy()
strategy.Register("my-custom", func(params retrier.BackOffParams) (time.Time, error) {
	// Compute next time based on params.GetRetries(), params.GetBackOffParams()...
	return time.Now().Add(10 * time.Second), nil
})

Then set task.BackOffCode = "my-custom".

---

Full Example with Retry Loop

ctx := context.Background()

workerFn := func(ctx context.Context, payload []byte) (string, *retrier.ExecutionError) {
	// Simulate transient failure.
	return "", &retrier.ExecutionError{
		Err:   errors.New("service unavailable"),
		State: retrier.UsualState,
	}
}

w, err := retrier.NewWorker(ctx, workerFn)
if err != nil {
	panic(err)
}
w.SetMinAndMaxWorkers(1, 2)
w.Start()
defer w.Stop()

task := retrier.Task{
	ID:           retrier.GetID(),
	Worker:       "default",
	Payload:      []byte("retry me"),
	Retries:      0,
	MaxRetries:   3,
	BackOffCode:  retrier.ExponentialBackOff,
	BackOffParams: map[retrier.BackOffParam]interface{}{
		retrier.BaseDelayKey:  500 * time.Millisecond,
		retrier.MultiplierKey: 2.0,
		retrier.MaxDelayKey:   10 * time.Second,
	},
}

w.Submit(task)

// In a real system, the manager would handle rescheduling.
// Here we simulate a simple retry loop.
for i := 0; i <= task.MaxRetries; i++ {
	res := <-w.GetOutChan()
	if res.Result.Status == retrier.StatusSuccess {
		fmt.Println("Task succeeded")
		break
	}
	// Compute next run using the backoff strategy.
	strategy := retrier.NewBackOffStrategy()
	nextTime, _ := strategy.Get(&task)
	fmt.Printf("Retry attempt %d scheduled at %s\n", i+1, nextTime)
	// In practice, you would update task.Retries and resubmit.
}

---

Testing

The package includes a complete test suite. Run:

go test -v ./...

---

This README describes the entire retrier package. For API details, refer to the Go doc comments in the source files.

Documentation

Overview

Package retrier provides a configurable Task retry and backoff mechanism. It includes a worker pool with dynamic scaling, Task submission, and pluggable backoff strategies for scheduling retries.

Package retrier provides a dynamic, auto‑scaling worker pool for concurrent task processing. It supports graceful shutdown, suspension, and configurable limits with idle timeout.

Index

• Constants
• type BackOff
• type BackOffFn
• type BackOffParam
• type BackOffParams
• type BackOffStrategy
  • func NewBackOffStrategy() *BackOffStrategy
  • func (s *BackOffStrategy) Get(backOff BackOffParams) (time.Time, error)
  • func (s *BackOffStrategy) Register(code string, fn BackOffFn)
• type Breaker
• type CircuitBreaker
  • func NewSlidingWindowCircuitBreaker(windowSize time.Duration, failureThreshold float64, minRequests int, ...) *CircuitBreaker
  • func (cb *CircuitBreaker) Allow() bool
  • func (cb *CircuitBreaker) RecordFailure()
  • func (cb *CircuitBreaker) RecordSuccess()
  • func (cb *CircuitBreaker) Reset()
  • func (cb *CircuitBreaker) State() CircuitBreakerState
• type CircuitBreakerState
• type ErrorState
• type EventPublisher
• type ExecutionError
• type FullWorkerState
• type Logger
• type Manager
  • func NewManager(ctx context.Context, store Store, logger Logger, backOff BackOff, ...) *Manager
  • func (m *Manager) GetWorkerStatuses() map[string]FullWorkerState
  • func (m *Manager) RegisterWorker(name string, w ManagerWorker, b Breaker) error
  • func (m *Manager) Start()
  • func (m *Manager) Stop()
  • func (m *Manager) Submit(task *Task) error
  • func (m *Manager) UnregisterWorker(name string)
• type ManagerWorker
• type MemStore
  • func NewMemStore() *MemStore
  • func (ms *MemStore) GetTasks() ([]Task, error)
  • func (ms *MemStore) SaveTask(t *Task, _ *TaskExecutionResult) error
• type Store
• type Task
  • func (t *Task) GetBackOffCode() string
  • func (t *Task) GetBackOffParams() map[BackOffParam]interface{}
  • func (t *Task) GetRetries() int
  • func (t *Task) IsFinished() bool
• type TaskExecutionResult
• type TaskStatus
• type Worker
  • func NewWorker(ctx context.Context, cfg *WorkerConfig, fn WorkerFn) (*Worker, error)
  • func (w *Worker) GetOutChan() chan WorkerExecutionResult
  • func (w *Worker) GetStatus() WorkerState
  • func (w *Worker) Start()
  • func (w *Worker) Stop()
  • func (w *Worker) Submit(task *Task) error
  • func (w *Worker) Suspend()
  • func (w *Worker) UpdateConfig(cfg *WorkerConfig)
• type WorkerConfig
  • func NewWorkerCfg(min, max int32, idleTimeout time.Duration) (*WorkerConfig, error)
• type WorkerExecutionResult
• type WorkerFn
• type WorkerState
• type WorkerStatus

Constants

const (
	LinearBackOff            string = "linear"
	JitterLinearBackOff      string = "jitter-linear"
	ExponentialBackOff       string = "exponential"
	JitterExponentialBackoff string = "jitter-exponential"
)

Predefined backoff strategy codes. Register these with the BackOffStrategy.

Types

type BackOff

type BackOff interface {
	Get(backOff BackOffParams) (time.Time, error)
}

BackOff defines the interface for computing the next execution time based on backoff parameters.

type BackOffFn

type BackOffFn func(params BackOffParams) (time.Time, error)

BackOffFn is a function that computes the next execution time based on the provided BackOffParams. It returns the scheduled time or an error.

type BackOffParam

type BackOffParam string

BackOffParam defines the allowed parameter keys for backoff strategy configuration.

const (
	// DurationKey specifies the fixed duration for linear backoff.
	DurationKey BackOffParam = "duration"
	// JitterKey specifies the jitter factor (as a fraction) to add randomness.
	JitterKey BackOffParam = "jitter"
	// MultiplierKey specifies the multiplier for exponential backoff.
	MultiplierKey BackOffParam = "multiplier"
	// MaxDelayKey specifies the upper bound for the calculated delay.
	MaxDelayKey BackOffParam = "maxDelay"
	// BaseDelayKey specifies the initial delay for exponential backoff.
	BaseDelayKey BackOffParam = "baseDelay"
)

type BackOffParams

type BackOffParams interface {
	// GetBackOffCode returns the identifier of the desired backoff strategy.
	GetBackOffCode() string
	// GetBackOffParams returns the map of parameters for the strategy.
	GetBackOffParams() map[BackOffParam]interface{}
	// GetRetries returns the number of attempts already made.
	GetRetries() int
}

BackOffParams is an interface that any Task or configuration must implement to be used with the backoff strategy. It provides the necessary metadata.

type BackOffStrategy

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

BackOffStrategy manages a registry of named backoff strategies and provides a thread-safe way to compute the next execution time.

func NewBackOffStrategy

func NewBackOffStrategy() *BackOffStrategy

NewBackOffStrategy creates a new BackOffStrategy pre‑registered with the four standard strategies: linear, jitter‑linear, exponential, and jitter‑exponential.

func (*BackOffStrategy) Get

func (s *BackOffStrategy) Get(backOff BackOffParams) (time.Time, error)

Get computes the next execution time using the strategy identified by the BackOffParams. It returns an error if the strategy is not registered.

func (*BackOffStrategy) Register

func (s *BackOffStrategy) Register(code string, fn BackOffFn)

Register adds or overwrites a backoff strategy under the given code. This method is safe for concurrent use.

type Breaker

type Breaker interface {
	Allow() bool
	RecordSuccess()
	RecordFailure()
	State() CircuitBreakerState
	Reset()
}

Breaker defines the public interface for a circuit breaker.

type CircuitBreaker

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

CircuitBreaker implements a circuit breaker using a sliding time window. It tracks success/failure counts over a configurable window duration and opens the circuit when the failure rate exceeds a given threshold.

func NewSlidingWindowCircuitBreaker

func NewSlidingWindowCircuitBreaker(
	windowSize time.Duration,
	failureThreshold float64,
	minRequests int,
	timeout time.Duration,
) *CircuitBreaker

NewSlidingWindowCircuitBreaker creates a new circuit breaker with sliding window. Parameters:

windowSize: duration of the sliding window
failureThreshold: allowed failure ratio (e.g., 0.5 means 50% failures)
minRequests: minimum number of requests required before evaluating threshold
timeout: duration to wait in open state before attempting half-open

func (*CircuitBreaker) Allow

func (cb *CircuitBreaker) Allow() bool

Allow checks if a request is permitted.

func (*CircuitBreaker) RecordFailure

func (cb *CircuitBreaker) RecordFailure()

RecordFailure records a failure execution.

func (*CircuitBreaker) RecordSuccess

func (cb *CircuitBreaker) RecordSuccess()

RecordSuccess records a successful execution.

func (*CircuitBreaker) Reset

func (cb *CircuitBreaker) Reset()

Reset manually resets the circuit to closed state and clears windows.

func (*CircuitBreaker) State

func (cb *CircuitBreaker) State() CircuitBreakerState

State returns the current state (thread-safe).

type CircuitBreakerState

type CircuitBreakerState string

CircuitBreakerState represents the state of a circuit breaker.

const (
	// StateClosed CB state
	StateClosed CircuitBreakerState = "closed"
	// StateOpen CB state
	StateOpen CircuitBreakerState = "open"
	// StateHalfOpen CB state
	StateHalfOpen CircuitBreakerState = "half-open"
)

type ErrorState

type ErrorState string

ErrorState classifies errors for retry or abort decisions.

const (
	// CriticalState marks errors that are unrecoverable (e.g., validation failures).
	// Tasks with critical errors should not be retried.
	CriticalState ErrorState = "critical"
	// UsualState marks transient errors (e.g., network timeouts) that may be retried.
	UsualState ErrorState = "usual"
)

type EventPublisher

type EventPublisher interface {
	Publish(event WorkerExecutionResult)
}

EventPublisher if we need to get Task and Task results immediately we can add published for send tasks

type ExecutionError

type ExecutionError struct {
	Err   error
	State ErrorState
}

ExecutionError wraps an error with a state that indicates whether the error is critical.

type FullWorkerState

type FullWorkerState struct {
	Status        WorkerStatus        `json:"status"`
	ActiveTasks   int32               `json:"active_tasks"`
	ActiveWorkers int32               `json:"active_workers"`
	CBState       CircuitBreakerState `json:"cb_state"`
}

FullWorkerState worker state with data from Circuit Breaker

type Logger

type Logger interface {
	Infof(format string, args ...interface{})
	Errorf(format string, args ...interface{})
}

Logger specifies the logging capability required by the retry manager.

type Manager

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

Manager orchestrates worker pools, routes execution results to storage, and buffers data locally in-memory during storage outages.

func NewManager

func NewManager(
	ctx context.Context,
	store Store,
	logger Logger,
	backOff BackOff,
	maxBufferSize int,
	fetchTaskTimeout time.Duration,
	fetchTaskTimeoutMax time.Duration,
	eventPublisher EventPublisher,
) *Manager

NewManager initializes a new Manager with the required dependencies and configurations.

func (*Manager) GetWorkerStatuses

func (m *Manager) GetWorkerStatuses() map[string]FullWorkerState

GetWorkerStatuses returns a snapshot of the current state of all registered workers.

func (*Manager) RegisterWorker

func (m *Manager) RegisterWorker(name string, w ManagerWorker, b Breaker) error

RegisterWorker adds a new worker with an optional circuit breaker.

func (*Manager) Start

func (m *Manager) Start()

Start boots the manager, launches all registered workers, and spins up pipeline routines.

func (*Manager) Stop

func (m *Manager) Stop()

Stop gracefully shuts down the manager, ensures all results are persisted, and waits for all goroutines to finish.

func (*Manager) Submit

func (m *Manager) Submit(task *Task) error

Submit saves a Task to the store without executing it immediately.

func (*Manager) UnregisterWorker

func (m *Manager) UnregisterWorker(name string)

UnregisterWorker removes a worker and its associated circuit breaker.

type ManagerWorker

type ManagerWorker interface {
	Start()
	Stop()
	Submit(t *Task) error
	GetOutChan() chan WorkerExecutionResult
	GetStatus() WorkerState
}

ManagerWorker defines the contract for a component capable of processing tasks and streaming execution results asynchronously.

type MemStore

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

MemStore is an in-memory implementation of the Store interface. It retains only tasks that are not yet finished (pending or suspended), allowing them to be retried. Completed tasks (success or permanent failure) are skipped during SaveTask.

This implementation is intended for testing and demonstration purposes only. For production use, consider a persistent store with proper indexing and transaction support.

func NewMemStore

func NewMemStore() *MemStore

NewMemStore creates a new empty in-memory store.

func (*MemStore) GetTasks

func (ms *MemStore) GetTasks() ([]Task, error)

GetTasks returns all tasks currently stored in memory. It returns a copy of the internal slice to avoid external modifications.

func (*MemStore) SaveTask

func (ms *MemStore) SaveTask(t *Task, _ *TaskExecutionResult) error

SaveTask stores a Task if it is not yet finished (status is not "success" or "failure"). If a Task with the same ID already exists, it is updated in place. This ensures that retry counts, status, and NextRun are kept current.

type Store

type Store interface {
	GetTasks() ([]Task, error)
	SaveTask(task *Task, result *TaskExecutionResult) error
}

Store abstracts persistence layer operations for logging and auditing Task outcomes.

type Task

type Task struct {
	// ID uniquely identifies this Task across the entire system.
	ID uuid.UUID `json:"id"`
	// Ctx context for tracing
	Ctx context.Context `json:"-"`
	// Payload holds the strongly-typed input arguments required for Task execution.
	Payload []byte `json:"payload"`
	// ManagerWorker specifies the designated runner type or queue name for this Task.
	Worker string `json:"worker"`
	// Status tracks the current lifecycle phase of the Task (e.g., pending, running, failed).
	Status TaskStatus `json:"status"`

	// Retries count of execution times
	Retries int `json:"retries"`
	// MaxRetries max tries count
	MaxRetries int `json:"max_retries"`
	// BackOffCode code of back off strategy
	BackOffCode string `json:"backoff_code"`
	// BackOffParams params for back off strategy
	BackOffParams map[BackOffParam]interface{} `json:"backoff_params"`

	// Deadline is an optional time limit for Task completion.
	// If the current time exceeds this deadline before the Task starts executing,
	// the Task will be marked as failed with a critical error and will not be retried.
	// Zero value (time.Time{}) indicates no deadline.
	Deadline time.Time `json:"deadline"`

	// CreatedAt records the exact timestamp when the Task was initially created.
	CreatedAt time.Time `json:"created_at"`
	// LastRun records the timestamp of the most recent execution attempt, if any.
	LastRun time.Time `json:"last_run"`
	// NextRun records the scheduled timestamp when the Task should be picked up next.
	NextRun time.Time `json:"next_run"`
}

Task represents a generic executable unit of work with retry tracking.

func (*Task) GetBackOffCode

func (t *Task) GetBackOffCode() string

GetBackOffCode returns the backoff strategy code.

func (*Task) GetBackOffParams

func (t *Task) GetBackOffParams() map[BackOffParam]interface{}

GetBackOffParams returns the parameters for the backoff strategy.

func (*Task) GetRetries

func (t *Task) GetRetries() int

GetRetries returns the current retry count.

func (*Task) IsFinished

func (t *Task) IsFinished() bool

IsFinished is Task finished and will not be retried

type TaskExecutionResult

type TaskExecutionResult struct {
	// ID uniquely identifies this specific execution outcome record.
	ID uuid.UUID
	// TaskID references the parent Task that generated this execution Result.
	TaskID uuid.UUID
	// Status indicates whether this specific run succeeded or encountered an error.
	Status TaskStatus
	// RunAt records the exact timestamp when this execution attempt was performed.
	RunAt time.Time
	// Result stores the raw payload returned by the workerImpl, such as response data or error details.
	Result []byte
	// IsCritical if this is a validation error, we have no any tries
	IsCritical bool
	// ExecutionTime worker func duration
	ExecutionTime time.Duration
}

TaskExecutionResult records the outcome and metadata of a single execution attempt of a Task.

type TaskStatus

type TaskStatus string

TaskStatus Task status

const (
	// StatusPending Task status
	StatusPending TaskStatus = "pending"
	// StatusSuccess Task status
	StatusSuccess TaskStatus = "success"
	// StatusFailure Task status
	StatusFailure TaskStatus = "failure"
)

type Worker

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

Worker manages a dynamic pool of goroutines that process tasks from an input queue. It scales up when all workers are busy (up to maxWorkers) and scales down when workers are idle for idleTimeout (down to minWorkers). It provides thread‑safe operations for starting, stopping, suspending, and submitting tasks.

func NewWorker

func NewWorker(ctx context.Context, cfg *WorkerConfig, fn WorkerFn) (*Worker, error)

NewWorker constructs a new worker pool with the given configuration and processing function. The pool starts in the Created state; call Start() to begin processing.

func (*Worker) GetOutChan

func (w *Worker) GetOutChan() chan WorkerExecutionResult

GetOutChan returns the output channel where completed task results are delivered. The channel is closed when the pool is stopped. If the pool is restarted after Stop, a new channel is created; callers should obtain the new channel again.

func (*Worker) GetStatus

func (w *Worker) GetStatus() WorkerState

GetStatus returns a snapshot of the pool's current state. It is safe to call concurrently.

func (*Worker) Start

func (w *Worker) Start()

Start transitions the pool into the Running state and spawns the minimum number of workers. If the pool is already Running, it does nothing. If it was Suspended, it cancels the previous context to terminate old workers and starts fresh. If it was Stopped, it re‑creates the internal channels (since they were closed by Stop), resets all counters, and starts a new generation of workers. After Start returns, the pool is ready to accept tasks again.

func (*Worker) Stop

func (w *Worker) Stop()

Stop permanently shuts down the pool. It cancels all workers, waits for them to finish, and closes the input and output channels. After Stop, the pool can be restarted by calling Start() (which will recreate the channels). It is safe to call multiple times.

func (*Worker) Submit

func (w *Worker) Submit(task *Task) error

Submit enqueues a task for processing. It attempts a non‑blocking send first; if no worker is idle, it scales up (if possible) and then blocks until the task is accepted or the pool is cancelled/stopped. Returns an error if the pool is not Running.

func (*Worker) Suspend

func (w *Worker) Suspend()

Suspend puts the pool into the Suspended state. It does not cancel existing tasks; it only prevents new submissions. Active tasks continue to completion. The pool can be resumed by calling Start() again.

func (*Worker) UpdateConfig

func (w *Worker) UpdateConfig(cfg *WorkerConfig)

UpdateConfig replaces the current configuration with a new one. The change takes effect immediately for subsequent scaling decisions and idle timeouts. It is safe to call concurrently.

type WorkerConfig

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

WorkerConfig holds the tunable parameters for the worker pool.

func NewWorkerCfg

func NewWorkerCfg(min, max int32, idleTimeout time.Duration) (*WorkerConfig, error)

NewWorkerCfg creates a validated WorkerConfig. Returns an error if min > max, any value is negative, or idleTimeout <= 0.

type WorkerExecutionResult

type WorkerExecutionResult struct {
	Task   *Task
	Result *TaskExecutionResult
}

WorkerExecutionResult pairs the original Task with its execution result.

type WorkerFn

type WorkerFn func(ctx context.Context, payload []byte) (string, *ExecutionError)

WorkerFn is the user‑defined function that processes a task payload. It returns a result string and an optional ExecutionError. If the function panics, it is recovered and treated as a critical error.

type WorkerState

type WorkerState struct {
	Status        WorkerStatus `json:"status"`
	ActiveTasks   int32        `json:"active_tasks"`   // number of tasks currently being processed
	ActiveWorkers int32        `json:"active_workers"` // number of running worker goroutines
}

WorkerState is a snapshot of the pool's current status.

type WorkerStatus

type WorkerStatus string

WorkerStatus represents the current lifecycle state of the worker pool.

const (
	// WorkerStatusCreated indicates the pool is initialized but not yet started.
	WorkerStatusCreated WorkerStatus = "created"
	// WorkerStatusRunning indicates the pool is actively accepting and executing tasks.
	WorkerStatusRunning WorkerStatus = "running"
	// WorkerStatusStopped indicates the pool is completely shut down and cannot be reused.
	WorkerStatusStopped WorkerStatus = "stopped"
	// WorkerStatusSuspended indicates the pool is temporarily paused; active tasks are drained,
	// and new submissions are rejected.
	WorkerStatusSuspended WorkerStatus = "suspended"
)