Skip to content

forge.retry

Exponential backoff, jitter, and circuit breaker for resilient external service calls.

forge.retry

Retry and resilience module — exponential backoff, jitter, and circuit breaker.

Provides the @retry decorator and CircuitBreaker class for handling transient failures in external service calls. Supports configurable backoff strategies, exception filtering, and automatic logging.

Classes

CircuitBreaker

Async circuit breaker with three-state machine.

States: CLOSED (normal) → OPEN (failing) → HALF_OPEN (probing).

Source code in src/forge/retry/circuit.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
class CircuitBreaker:
    """
    Async circuit breaker with three-state machine.

    States: ``CLOSED`` (normal) → ``OPEN`` (failing) → ``HALF_OPEN`` (probing).
    """

    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_time: float = 60.0,
        *,
        name: str = "",
    ) -> None:
        if failure_threshold < 1:
            raise ValueError("failure_threshold must be >= 1")
        if recovery_time < 0:
            raise ValueError("recovery_time must be >= 0")

        self._failure_threshold = failure_threshold
        self._recovery_time = recovery_time
        self._name = name or hex(id(self))

        self._state = CircuitBreakerState.CLOSED
        self._failure_count = 0
        self._last_failure_time: float = 0.0

    @property
    def state(self) -> CircuitBreakerState:
        return self._state

    @property
    def name(self) -> str:
        return self._name

    async def call(self, coro: Any) -> Any:
        """Execute *coro* under circuit-breaker protection."""
        self._check_state()
        try:
            result = await coro
        except Exception:
            self._on_failure()
            raise
        else:
            self._on_success()
            return result

    async def __aenter__(self) -> Self:
        self._check_state()
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: object,
    ) -> bool:
        if exc_type is None:
            self._on_success()
        elif exc_type is not asyncio.CancelledError:
            self._on_failure()
        return False

    def __call__(self, fn: Any) -> Any:
        async def wrapper(*args: object, **kwargs: object) -> Any:
            return await self.call(fn(*args, **kwargs))

        wrapper.__name__ = fn.__name__
        wrapper.__qualname__ = fn.__qualname__
        wrapper.__module__ = fn.__module__
        wrapper.__doc__ = fn.__doc__
        return wrapper

    def reset(self) -> None:
        self._state = CircuitBreakerState.CLOSED
        self._failure_count = 0
        self._last_failure_time = 0.0
        _record_cb_state(self._name, self._state)

    def _check_state(self) -> None:
        if self._state is CircuitBreakerState.OPEN:
            if time.monotonic() - self._last_failure_time >= self._recovery_time:
                _logger.debug(
                    "Circuit breaker %r transitioning OPEN -> HALF_OPEN",
                    self._name,
                )
                self._state = CircuitBreakerState.HALF_OPEN
                _record_cb_state(self._name, self._state)
            else:
                raise CircuitBreakerOpenError(f"Circuit breaker {self._name!r} is OPEN")

    def _on_success(self) -> None:
        if self._state is CircuitBreakerState.HALF_OPEN:
            _logger.info("Circuit breaker %r recovered, closing", self._name)
        self._state = CircuitBreakerState.CLOSED
        self._failure_count = 0
        _record_cb_state(self._name, self._state)

    def _on_failure(self) -> None:
        self._failure_count += 1
        self._last_failure_time = time.monotonic()
        if (
            self._state is CircuitBreakerState.HALF_OPEN
            or self._failure_count >= self._failure_threshold
        ):
            _logger.warning(
                "Circuit breaker %r opening (failures=%d/%d)",
                self._name,
                self._failure_count,
                self._failure_threshold,
            )
            self._state = CircuitBreakerState.OPEN
            _record_cb_state(self._name, self._state)
Methods:
call async 
call(coro: Any) -> Any

Execute coro under circuit-breaker protection.

Source code in src/forge/retry/circuit.py 
83
84
85
86
87
88
89
90
91
92
93
async def call(self, coro: Any) -> Any:
    """Execute *coro* under circuit-breaker protection."""
    self._check_state()
    try:
        result = await coro
    except Exception:
        self._on_failure()
        raise
    else:
        self._on_success()
        return result

CircuitBreakerOpenError

Bases: Exception

Raised when a call is rejected because the circuit breaker is open.

Source code in src/forge/retry/circuit.py 
38
39
class CircuitBreakerOpenError(Exception):
    """Raised when a call is rejected because the circuit breaker is open."""

NonRetryableError

Bases: Exception

Raise this inside a retried call to abort retrying immediately.

Source code in src/forge/retry/module.py 
55
56
class NonRetryableError(Exception):
    """Raise this inside a retried call to abort retrying immediately."""

RetryError

Bases: Exception

Raised when all retry attempts have been exhausted.

Attributes

original_exception : The last exception raised by the wrapped callable. attempt_count : Total number of attempts made (including the initial call). total_delay : Total wall-clock time spent sleeping between retries.

Source code in src/forge/retry/module.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class RetryError(Exception):
    """
    Raised when all retry attempts have been exhausted.

    Attributes
    ----------
    original_exception :
        The last exception raised by the wrapped callable.
    attempt_count :
        Total number of attempts made (including the initial call).
    total_delay :
        Total wall-clock time spent sleeping between retries.
    """

    def __init__(
        self,
        original_exception: Exception,
        attempt_count: int,
        total_delay: float,
    ) -> None:
        self.original_exception = original_exception
        self.attempt_count = attempt_count
        self.total_delay = total_delay
        super().__init__(
            f"All {attempt_count} retry attempt(s) exhausted "
            f"(total delay={total_delay:.2f}s): {original_exception}"
        )

RetryModule

Bases: ForgeModule

Source code in src/forge/retry/module.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
class RetryModule(ForgeModule):
    name = "retry"
    dependencies: ClassVar[list[str]] = ["config"]

    def __init__(self) -> None:
        super().__init__()
        self._config_retry: RetryConfig | None = None

    async def setup(self, runtime: ForgeRuntime) -> None:
        from forge.config.module import ConfigModule

        config_module: ConfigModule = runtime.get(ConfigModule)  # type: ignore[assignment]
        self._config_retry = config_module.config.retry

    async def teardown(self) -> None:
        self._config_retry = None

    def health_check(self) -> HealthResult:
        return HealthResult.ok()

    def retry(
        self,
        fn: Callable[..., Any] | None = None,
        *,
        attempts: int | None = None,
        backoff: str | None = None,
        base_delay: float | None = None,
        max_delay: float | None = None,
        jitter: bool | None = None,
        timeout: float | None = None,
        retryable_exceptions: tuple[type[Exception], ...] | None = None,
    ) -> Any:
        """Return a :func:`retry` pre-configured with module defaults."""
        cfg = self._config_retry
        if cfg is None:
            raise RuntimeError("RetryModule not initialised")
        return retry(
            fn=fn,
            attempts=attempts if attempts is not None else cfg.default_attempts,
            backoff=backoff if backoff is not None else cfg.default_backoff,
            base_delay=base_delay if base_delay is not None else cfg.default_base_delay,
            max_delay=max_delay if max_delay is not None else cfg.default_max_delay,
            jitter=jitter if jitter is not None else cfg.default_jitter,
            timeout=timeout,
            retryable_exceptions=retryable_exceptions,
        )

    def circuit_breaker(
        self,
        failure_threshold: int | None = None,
        recovery_time: float | None = None,
        *,
        name: str = "",
    ) -> Any:
        """Return a :class:`~forge.retry.circuit.CircuitBreaker` with module defaults."""
        from forge.retry.circuit import CircuitBreaker

        cfg = self._config_retry
        if cfg is None:
            raise RuntimeError("RetryModule not initialised")
        cb_cfg = cfg.circuit_breaker
        return CircuitBreaker(
            failure_threshold=(
                failure_threshold if failure_threshold is not None else cb_cfg.failure_threshold
            ),
            recovery_time=(recovery_time if recovery_time is not None else cb_cfg.recovery_time),
            name=name,
        )
Methods:
circuit_breaker 
circuit_breaker(failure_threshold: int | None = None, recovery_time: float | None = None, *, name: str = '') -> Any

Return a :class:~forge.retry.circuit.CircuitBreaker with module defaults.

Source code in src/forge/retry/module.py 
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
def circuit_breaker(
    self,
    failure_threshold: int | None = None,
    recovery_time: float | None = None,
    *,
    name: str = "",
) -> Any:
    """Return a :class:`~forge.retry.circuit.CircuitBreaker` with module defaults."""
    from forge.retry.circuit import CircuitBreaker

    cfg = self._config_retry
    if cfg is None:
        raise RuntimeError("RetryModule not initialised")
    cb_cfg = cfg.circuit_breaker
    return CircuitBreaker(
        failure_threshold=(
            failure_threshold if failure_threshold is not None else cb_cfg.failure_threshold
        ),
        recovery_time=(recovery_time if recovery_time is not None else cb_cfg.recovery_time),
        name=name,
    )
retry 
retry(fn: Callable[..., Any] | None = None, *, attempts: int | None = None, backoff: str | None = None, base_delay: float | None = None, max_delay: float | None = None, jitter: bool | None = None, timeout: float | None = None, retryable_exceptions: tuple[type[Exception], ...] | None = None) -> Any

Return a :func:retry pre-configured with module defaults.

Source code in src/forge/retry/module.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
def retry(
    self,
    fn: Callable[..., Any] | None = None,
    *,
    attempts: int | None = None,
    backoff: str | None = None,
    base_delay: float | None = None,
    max_delay: float | None = None,
    jitter: bool | None = None,
    timeout: float | None = None,
    retryable_exceptions: tuple[type[Exception], ...] | None = None,
) -> Any:
    """Return a :func:`retry` pre-configured with module defaults."""
    cfg = self._config_retry
    if cfg is None:
        raise RuntimeError("RetryModule not initialised")
    return retry(
        fn=fn,
        attempts=attempts if attempts is not None else cfg.default_attempts,
        backoff=backoff if backoff is not None else cfg.default_backoff,
        base_delay=base_delay if base_delay is not None else cfg.default_base_delay,
        max_delay=max_delay if max_delay is not None else cfg.default_max_delay,
        jitter=jitter if jitter is not None else cfg.default_jitter,
        timeout=timeout,
        retryable_exceptions=retryable_exceptions,
    )

Functions:

constant 

constant(attempt: int, base_delay: float = 1.0, max_delay: float = 60.0) -> float

Constant backoff — always returns base_delay.

Source code in src/forge/retry/backoff.py 
32
33
34
def constant(attempt: int, base_delay: float = 1.0, max_delay: float = 60.0) -> float:  # noqa: ARG001
    """Constant backoff — always returns *base_delay*."""
    return base_delay

exponential 

exponential(attempt: int, base_delay: float = 1.0, max_delay: float = 60.0) -> float

Exponential backoff with full jitter.

Delay = U(0, min(base_delay * 2 ** (attempt - 1), max_delay))

Source code in src/forge/retry/backoff.py 
13
14
15
16
17
18
19
20
def exponential(attempt: int, base_delay: float = 1.0, max_delay: float = 60.0) -> float:
    """
    Exponential backoff with full jitter.

    Delay = ``U(0, min(base_delay * 2 ** (attempt - 1), max_delay))``
    """
    cap = min(base_delay * 2.0 ** (attempt - 1), max_delay)
    return random.uniform(0.0, cap)  # noqa: S311

get_backoff 

get_backoff(name: str) -> BackoffFn

Return the backoff function for the given strategy name.

Raises ValueError for unknown strategies.

Source code in src/forge/retry/backoff.py 
44
45
46
47
48
49
50
51
52
53
def get_backoff(name: str) -> BackoffFn:
    """
    Return the backoff function for the given strategy *name*.

    Raises ``ValueError`` for unknown strategies.
    """
    fn = _STRATEGIES.get(name)
    if fn is None:
        raise ValueError(f"Unknown backoff strategy {name!r}. Choose from {list(_STRATEGIES)}.")
    return fn

linear 

linear(attempt: int, base_delay: float = 1.0, max_delay: float = 60.0) -> float

Linear backoff.

Delay = min(base_delay * attempt, max_delay)

Source code in src/forge/retry/backoff.py 
23
24
25
26
27
28
29
def linear(attempt: int, base_delay: float = 1.0, max_delay: float = 60.0) -> float:
    """
    Linear backoff.

    Delay = ``min(base_delay * attempt, max_delay)``
    """
    return min(base_delay * attempt, max_delay)

retry 

retry(fn: Callable[..., Any] | None = None, *, attempts: int = 3, backoff: str = 'exponential', base_delay: float = 1.0, max_delay: float = 60.0, jitter: bool = True, timeout: float | None = None, retryable_exceptions: tuple[type[Exception], ...] | None = None) -> Any

Async retry — use as a decorator or async context manager.

Decorator form::

@retry(attempts=3)
async def fetch() -> bytes:
    ...

Context manager form::

async with retry(attempts=3):
    data = await fetch()
Source code in src/forge/retry/module.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def retry(
    fn: Callable[..., Any] | None = None,
    *,
    attempts: int = 3,
    backoff: str = "exponential",
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: bool = True,
    timeout: float | None = None,
    retryable_exceptions: tuple[type[Exception], ...] | None = None,
) -> Any:
    """
    Async retry — use as a decorator or async context manager.

    Decorator form::

        @retry(attempts=3)
        async def fetch() -> bytes:
            ...

    Context manager form::

        async with retry(attempts=3):
            data = await fetch()
    """
    impl = _RetryImpl(
        attempts=attempts,
        backoff=backoff,
        base_delay=base_delay,
        max_delay=max_delay,
        jitter=jitter,
        timeout=timeout,
        retryable_exceptions=retryable_exceptions,
    )
    if fn is not None:
        return impl(fn)
    return impl