API: Retries

This page documents the available recovery retry strategies. Retry strategies determine when the circuit breaker should attempt to recover by moving from the OPEN state to the HALF_OPEN state.

For a high-level guide on choosing a retry strategy, see the Retries Component Guide.

---

fluxgate.retries.Always

Bases: RetryBase

Retry strategy that always allows transition to HALF_OPEN.

Circuit immediately attempts to recover on every call.

Examples:

>>> retry = Always()  # Always try to recover

__call__

__call__(_changed_at: float, _reopens: int) -> bool

Source code in fluxgate/retries.py

def __call__(self, _changed_at: float, _reopens: int) -> bool:
    return True

fluxgate.retries.Never

Bases: RetryBase

Retry strategy that never allows transition to HALF_OPEN.

Circuit stays OPEN indefinitely until manually reset.

Examples:

>>> retry = Never()  # Require manual intervention

__call__

__call__(_changed_at: float, _reopens: int) -> bool

Source code in fluxgate/retries.py

def __call__(self, _changed_at: float, _reopens: int) -> bool:
    return False

fluxgate.retries.Cooldown

Cooldown(duration: float, jitter_ratio: float = 0.0)

Bases: RetryBase

Retry strategy with fixed cooldown period.

Allows transition to HALF_OPEN after a fixed duration has elapsed. Optional jitter can be added to avoid thundering herd.

Examples:

>>> # Wait 60 seconds before retry
>>> retry = Cooldown(duration=60.0)
>>>
>>> # Wait 60 seconds with ±10% jitter
>>> retry = Cooldown(duration=60.0, jitter_ratio=0.1)

Parameters:

Name	Type	Description	Default
duration	float	Cooldown duration in seconds	required
jitter_ratio	float	Random jitter ratio (0.0 to 1.0, default 0.0)	0.0

Source code in fluxgate/retries.py

def __init__(self, duration: float, jitter_ratio: float = 0.0) -> None:
    if duration <= 0:
        raise ValueError("Duration must be greater than zero")
    if not (0.0 <= jitter_ratio < 1.0):
        raise ValueError("Jitter ratio must be between 0.0 and 1.0")
    self._duration = duration
    self._jitter_ratio = jitter_ratio

__call__

__call__(changed_at: float, _reopens: int) -> bool

Source code in fluxgate/retries.py

def __call__(self, changed_at: float, _reopens: int) -> bool:
    if self._jitter_ratio > 0:
        jitter = uniform(-self._jitter_ratio, self._jitter_ratio)
        actual_duration = self._duration * (1 + jitter)
    else:
        actual_duration = self._duration
    elapsed = time.time() - changed_at
    return elapsed >= actual_duration

fluxgate.retries.Backoff

Backoff(
    initial: float,
    multiplier: float = 2.0,
    max_duration: float = 300.0,
    jitter_ratio: float = 0.0,
)

Bases: RetryBase

Retry strategy with exponential backoff.

Wait time increases exponentially with each reopen: initial * (multiplier ^ reopens). Useful for giving the service more time to recover after repeated failures.

Examples:

>>> # Start with 10s, double each time, max 300s
>>> retry = Backoff(initial=10.0, multiplier=2.0, max_duration=300.0)
>>> # Reopens: 0->10s, 1->20s, 2->40s, 3->80s, 4->160s, 5+->300s
>>>
>>> # With jitter to avoid thundering herd
>>> retry = Backoff(initial=10.0, multiplier=2.0, jitter_ratio=0.1)

Parameters:

Name	Type	Description	Default
initial	float	Initial wait duration in seconds	required
multiplier	float	Backoff multiplier (must be > 1.0)	2.0
max_duration	float	Maximum wait duration in seconds	300.0
jitter_ratio	float	Random jitter ratio (0.0 to 1.0, default 0.0)	0.0

Source code in fluxgate/retries.py

def __init__(
    self,
    initial: float,
    multiplier: float = 2.0,
    max_duration: float = 300.0,
    jitter_ratio: float = 0.0,
) -> None:
    if initial <= 0:
        raise ValueError("Initial duration must be greater than zero")
    if multiplier <= 1.0:
        raise ValueError("Multiplier must be greater than 1.0")
    if max_duration < initial:
        raise ValueError("Max duration must be >= initial duration")
    if not (0.0 <= jitter_ratio < 1.0):
        raise ValueError("Jitter ratio must be between 0.0 and 1.0")

    self._initial = initial
    self._multiplier = multiplier
    self._max = max_duration
    self._jitter_ratio = jitter_ratio

__call__

__call__(changed_at: float, reopens: int) -> bool

Source code in fluxgate/retries.py

def __call__(self, changed_at: float, reopens: int) -> bool:
    # Calculate wait duration based on reopens count
    duration = min(self._initial * (self._multiplier**reopens), self._max)

    if self._jitter_ratio > 0:
        jitter = uniform(-self._jitter_ratio, self._jitter_ratio)
        actual_duration = duration * (1 + jitter)
    else:
        actual_duration = duration

    elapsed = time.time() - changed_at
    return elapsed >= actual_duration