Documentation
¶
Overview ¶
Package routine provides utilities for managing recurring tasks (routines) with circuit breaker protection.
The primary purpose of this package is to provide a robust and managed way to run recurring background tasks that might fail, especially those interacting with external services or resources. It combines two key concepts:
1. Circuit Breaker (`Routine`): This pattern prevents an application from repeatedly performing an operation that's likely to fail. When failures reach a threshold, the "circuit" opens, and calls are blocked for a set period (`resetTimeout`). After the timeout, it enters a "half-open" state, allowing one test call. If that succeeds, the circuit closes; otherwise, it re-opens. This provides:
- Fault Tolerance: Prevents cascading failures by isolating problems.
- Resource Protection: Stops wasting resources (CPU, network, memory, API quotas) on calls that are continuously failing.
- Automatic Recovery: Gives the failing service/resource time to recover before trying again automatically.
2. Managed Background Loops (`group`): This component manages multiple circuit breakers (`Routine` instances) identified by unique keys. It handles the lifecycle of running the associated task (`fn`) in a background goroutine, ensuring:
- Organization: Keeps track of different background tasks centrally.
- Deduplication: Ensures only one instance of a loop runs for a given key, even if `StartLoop` is called multiple times for the same key.
- Control: Allows periodic execution (`interval`), on-demand triggering (`ForceUpdate`), context-based cancellation, and manual state resets (`ResetRoutine`).
In essence, use `routine` to reliably run background jobs that need to be resilient to temporary failures without overwhelming either your application or the dependencies they rely on. See `group.StartLoop` for the primary entry point for running managed routines.
Job chains (Runner) ¶
Runner, Job, and Schedule build condition-gated, chainable operations on top of a Routine: Runner wraps one Routine so a job chain gets the same circuit-breaker protection as any other routine, adds a single-flight guard so a slow run is never overlapped by its own next trigger, and exposes three ways to fire that guarded execution — Run/Trigger directly, StartSchedule on an arbitrary Schedule (interface-compatible with robfig/cron/v3's cron.Schedule, so real cron syntax is a parser away without libroutine depending on one), or SubscribeMessenger to react to a github.com/contenox/contenox/libbus.Messenger subject.
Condition and Operation are plain funcs, so a Job can drive anything (including github.com/contenox/contenox/libprocess), and every trigger path runs through the same Routine — a chain that starts failing repeatedly opens its circuit and backs off instead of being retried every tick or event forever.
Index ¶
- Variables
- func GetGroup() *group
- type Condition
- type Job
- type LoopConfig
- type Operation
- type Routine
- func (rm *Routine) Allow() bool
- func (rm *Routine) Execute(ctx context.Context, fn func(ctx context.Context) error) error
- func (rm *Routine) ExecuteWithRetry(ctx context.Context, interval time.Duration, iterations int, ...) error
- func (rm *Routine) ForceClose()
- func (rm *Routine) ForceOpen()
- func (rm *Routine) GetResetTimeout() time.Duration
- func (rm *Routine) GetState() State
- func (rm *Routine) GetThreshold() int
- func (rm *Routine) Loop(ctx context.Context, interval time.Duration, triggerChan <-chan struct{}, ...)
- func (rm *Routine) MarkFailure()
- func (rm *Routine) MarkSuccess()
- type RunResult
- type Runner
- func (r *Runner) Run(ctx context.Context) (*RunResult, error)
- func (r *Runner) Running() bool
- func (r *Runner) StartSchedule(ctx context.Context, sched Schedule)
- func (r *Runner) SubscribeMessenger(ctx context.Context, bus libbus.Messenger, subject string) (libbus.Subscription, error)
- func (r *Runner) Trigger(ctx context.Context)
- type RunnerOption
- type Schedule
- type State
Constants ¶
This section is empty.
Variables ¶
var ErrAlreadyRunning = errors.New("libroutine: job is already running")
ErrAlreadyRunning is returned by Run when the Runner's job chain is already executing. It is distinct from ErrCircuitOpen: this guards against overlapping a single slow run with itself, independent of the underlying Routine's failure-count state.
var ErrCircuitOpen = errors.New("circuit breaker is open")
ErrCircuitOpen is returned by Execute when the circuit breaker is in the Open state and blocking calls. Callers can use errors.Is(err, ErrCircuitOpen) to check.
Functions ¶
Types ¶
type Condition ¶ added in v0.38.0
Condition gates whether a Job's Operation runs. A nil Condition on a Job always proceeds.
type Job ¶ added in v0.38.0
type Job struct {
// Name identifies this job in RunResult and error messages.
Name string
// Condition, if set, is evaluated before Operation. A false result (no
// error) skips Operation and Next without failing the run — the
// condition simply wasn't met, not an error.
Condition Condition
Operation Operation
// Next, if set, runs after Operation succeeds. It does not run if
// Condition returns false, or if Operation or Condition errors.
Next *Job
}
Job is one step in a chain: check Condition, run Operation, and — if both succeed — continue into Next. A Job is driven by a Runner, which adds the circuit-breaker protection and single-flight guard around it.
type LoopConfig ¶
type LoopConfig struct {
Key string // A unique string identifier for this routine. Used to prevent duplicates and manage state.
Threshold int // The number of consecutive failures of `fn` before the circuit breaker opens.
ResetTimeout time.Duration // The duration the circuit breaker stays open before transitioning to half-open.
Interval time.Duration // The time duration between executions of `fn` when the circuit is closed or half-open (and the attempt succeeds).
Operation func(ctx context.Context) error // The function to execute periodically. It receives the context and should return an error on failure
}
type Operation ¶ added in v0.38.0
Operation is the work a Job performs once its Condition allows it.
type Routine ¶
type Routine struct {
// contains filtered or unexported fields
}
Routine implements a circuit breaker pattern to protect functions from repeated failures. It tracks failures, opens the circuit when a threshold is reached, and attempts to reset automatically after a timeout (via the HalfOpen state). Routines are typically managed by the group but can be used standalone if needed.
func NewRoutine ¶
NewRoutine creates a new circuit breaker (`Routine`) instance.
Parameters:
- threshold: The number of consecutive failures required to move the state from Closed to Open. Must be greater than 0.
- resetTimeout: The period the circuit breaker will remain Open before transitioning to HalfOpen. Must be positive.
func (*Routine) Allow ¶
Allow checks if the circuit breaker permits an operation based on its current state. It's consulted by `Execute`. Direct use is uncommon but possible for manual checks.
Returns:
- true: If the state is Closed, or if the state is HalfOpen and no test is ongoing.
- false: If the state is Open and the reset timeout hasn't elapsed, or if the state is HalfOpen and a test operation is already in progress.
Note: This method may transition the state from Open to HalfOpen if the timeout has passed.
func (*Routine) Execute ¶
Execute runs the provided function if allowed by the circuit breaker. Use this method when you need to protect a single, on-demand operation with circuit breaking, without requiring automatic looping or retries. For automatic retries, see `ExecuteWithRetry`. For recurring background tasks, see `group.StartLoop` or `Routine.Loop`.
func (*Routine) ExecuteWithRetry ¶
func (rm *Routine) ExecuteWithRetry(ctx context.Context, interval time.Duration, iterations int, fn func(ctx context.Context) error) error
ExecuteWithRetry attempts to run the function `fn` using `Execute`, retrying on failure up to `iterations` times with a fixed `interval` between attempts. Retries stop early if `fn` succeeds or if the context `ctx` is cancelled. The circuit breaker logic applies to *each* attempt via `Execute`.
Parameters:
- ctx: Context for cancellation. Retries stop if ctx is cancelled.
- interval: Fixed duration to wait between retry attempts. Consider jitter/backoff for production.
- iterations: Maximum number of execution attempts (including the first).
- fn: The function to execute.
Returns:
- nil: If `fn` executes successfully within the allowed attempts.
- context.Canceled or context.DeadlineExceeded: If the context is cancelled/times out.
- error: The last error encountered (either from `fn` or `ErrCircuitOpen`) if all attempts fail.
func (*Routine) ForceClose ¶
func (rm *Routine) ForceClose()
ForceClose manually sets the circuit breaker state to Closed and resets the failure count and test flag. Use primarily for testing or manual operational intervention.
func (*Routine) ForceOpen ¶
func (rm *Routine) ForceOpen()
ForceOpen sets the circuit breaker to the Open state.
func (*Routine) GetResetTimeout ¶
GetResetTimeout returns the reset timeout duration configured for this circuit breaker.
func (*Routine) GetState ¶
GetState returns the current State (Closed, Open, HalfOpen) of the circuit breaker.
func (*Routine) GetThreshold ¶
GetThreshold returns the failure threshold configured for this circuit breaker.
func (*Routine) Loop ¶
func (rm *Routine) Loop(ctx context.Context, interval time.Duration, triggerChan <-chan struct{}, fn func(ctx context.Context) error, errHandling func(err error))
Loop continuously executes the function `fn` based on the circuit breaker state and a timer or trigger channel. This is the core execution logic used by `group.StartLoop`.
The loop runs `Execute(fn)`: 1. Immediately when the loop starts. 2. After the `interval` duration elapses. 3. Immediately when a signal is received on `triggerChan`. 4. Stops when the `ctx` context is cancelled.
Parameters:
- ctx: Context for cancellation. The loop terminates when ctx is done.
- interval: The duration between scheduled executions when the circuit allows.
- triggerChan: A channel used to force immediate execution attempts (e.g., via `group.ForceUpdate`). Reads are non-blocking.
- fn: The function to execute in each cycle.
- errHandling: A callback function invoked when `Execute(fn)` returns an error. Use this for logging or custom error handling specific to the loop runner. Note: `ErrCircuitOpen` will also be passed here when calls are blocked.
func (*Routine) MarkFailure ¶
func (rm *Routine) MarkFailure()
MarkFailure records a failed operation. If the state was Closed, it increments the failure count. If the count reaches the threshold, it transitions to Open. If the state was HalfOpen, it transitions back to Open. Called internally by `Execute` on failure.
func (*Routine) MarkSuccess ¶
func (rm *Routine) MarkSuccess()
MarkSuccess resets the circuit breaker after a successful call.
type RunResult ¶ added in v0.38.0
type RunResult struct {
Name string
Skipped bool // Condition evaluated to false; Operation did not run
Err error
Duration time.Duration
// Next is the chained job's result, set only when this job's Operation
// succeeded and Next was run.
Next *RunResult
}
RunResult reports the outcome of running a Job, including its chain.
type Runner ¶ added in v0.38.0
type Runner struct {
// contains filtered or unexported fields
}
Runner drives one Job's execution through a dedicated Routine, so a job chain gets the same circuit-breaker protection (see Routine) as any other managed operation in this package, plus a single-flight guard so a slow run is never overlapped by its own next trigger. Runner is safe for concurrent use.
func NewRunner ¶ added in v0.38.0
NewRunner returns a Runner for job, protected by a Routine constructed with threshold and resetTimeout (see NewRoutine). The job chain is not started; use Run, Trigger, StartSchedule, or SubscribeMessenger to drive it.
func (*Runner) Run ¶ added in v0.38.0
Run executes the job chain synchronously through the Runner's Routine and returns its result. It returns ErrAlreadyRunning without running anything if the chain is already executing, and returns ErrCircuitOpen (see Routine.Execute) without running anything if the circuit breaker is open — in both cases the result is nil. Callers that want either condition silently absorbed instead of reported should use Trigger.
func (*Runner) Running ¶ added in v0.38.0
Running reports whether the job chain is currently executing.
func (*Runner) StartSchedule ¶ added in v0.38.0
StartSchedule runs r.Trigger each time sched fires, until ctx is cancelled. A tick that lands while the previous run is still in flight, or while the Runner's circuit breaker is open, is dropped (see Trigger) rather than queued. StartSchedule returns immediately; the schedule loop runs in a background goroutine that exits when ctx is done.
func (*Runner) SubscribeMessenger ¶ added in v0.38.0
func (r *Runner) SubscribeMessenger(ctx context.Context, bus libbus.Messenger, subject string) (libbus.Subscription, error)
SubscribeMessenger triggers r (see Trigger) every time a message is published to subject on bus, so a Job chain can react to an external event — e.g. another component publishing "process.myproc.running" — through this codebase's existing pub/sub abstraction (libbus.Messenger) rather than a bespoke event mechanism.
The subscription is torn down automatically when ctx is done (see libbus.Messenger.Stream); the returned Subscription can also be used to unsubscribe earlier.
func (*Runner) Trigger ¶ added in v0.38.0
Trigger requests a run without blocking the caller: it launches Run in a background goroutine and silently drops the request if the chain is already running or the circuit is open, rather than queuing or erroring on it. This is what StartSchedule ticks and SubscribeMessenger deliveries use.
type RunnerOption ¶ added in v0.38.0
type RunnerOption func(*Runner)
RunnerOption configures a Runner at construction.
func WithResultHook ¶ added in v0.38.0
func WithResultHook(fn func(*RunResult)) RunnerOption
WithResultHook registers fn to be called with the RunResult of every completed run (from Run, Trigger, a Schedule tick, or a SubscribeMessenger delivery) that actually executed the job — i.e. not when Execute short-circuited with ErrCircuitOpen or ErrAlreadyRunning. fn is called synchronously after the run completes and must not block.
This is a lighter-weight alternative to WithTracker for callers that just want the typed RunResult tree; use WithTracker instead to integrate with this codebase's standard instrumentation seam (metrics, logging, tracing, audit trails).
func WithTracker ¶ added in v0.38.0
func WithTracker(tracker libtracker.ActivityTracker) RunnerOption
WithTracker wires an ActivityTracker to observe every Run: Start is called when a run begins (after the single-flight and circuit-breaker gates pass), reportErr is called if the job chain failed or the circuit was open, reportChange is called with the job's Name and its RunResult on success, and end always fires. Without WithTracker, a Runner uses libtracker.NoopTracker.
type Schedule ¶ added in v0.38.0
Schedule computes the next run time after t. It is interface-compatible with robfig/cron/v3's cron.Schedule, so a caller who wants real cron expression syntax can parse one with that package and pass the result here directly, without libroutine depending on a cron parser itself.
func Every ¶ added in v0.38.0
Every returns a Schedule that fires at a fixed interval, with no dependency beyond the standard library. Use it directly for simple polling-style jobs, or as a placeholder until a real cron expression is wired in via a Schedule implementation.
Note this is a different tool than group.StartLoop's fixed interval: that method drives a single func(ctx) error directly under a Routine with no job-chain, condition, or cron-shaped scheduling around it. Reach for StartLoop when a plain recurring function is all you need, and for a Runner with Every (or a real Schedule) when you also want Job's condition gating or chaining.
type State ¶
type State int
State represents the operational state of the Routine (circuit breaker). Circuit Breaker States:
(Success) < Succeeded > --+ / |
[CLOSED] --- Failure Threshold Reached ---> [OPEN]
^ | | | Reset Timeout Elapsed | Success (Test OK) V
[HALF-OPEN] <---- Failure (Test Failed) ---[ Test Call ]
| +-- Failure (Test Failed) --> (Reverts to OPEN)
const ( // Closed allows operations to execute and counts failures. Closed State = iota // Open prevents operations from executing immediately. After a timeout, // it transitions to HalfOpen. Open // HalfOpen allows a single operation attempt. If successful, transitions to Closed. // If it fails, transitions back to Open. HalfOpen )
Source Files