Coverage for bzfs_main/util/retry.py: 100%
570 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 18:44 +0000
1# Copyright 2024 Wolfgang Hoschek AT mac DOT com
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14#
15"""Customizable generic retry framework; defaults to jittered exponential backoff with cap unless specified otherwise.
17Purpose:
18--------
19- Provide a reusable retry helper for transient failures using customizable policy and callbacks.
20- Centralize backoff, jitter, logging and metrics behavior while keeping call sites compact.
21- Prevent accidental retries: the loop retries only when the developer explicitly raises a ``RetryableError``, which reduces
22 the risk of retrying non-idempotent operations.
23- Provide a thread-safe, fast implementation; avoid shared RNG contention.
24- Provide both sync and async API, both with the same semantics, except the async API awaits ``fn``, awaitable
25 ``before_attempt`` / ``after_attempt`` / ``on_retryable_error`` / ``on_exhaustion`` / ``giveup`` results, and non-blocking
26 sleep.
27- Avoid unnecessary complexity and add zero dependencies beyond the Python standard library. Everything you need is in this
28 single Python file.
30Usage:
31------
32- Wrap work in a callable ``fn(retry: Retry)`` and therein raise ``RetryableError`` for failures that should be retried.
33- Construct a policy via ``RetryPolicy(...)`` that specifies how ``RetryableError`` shall be retried.
34- Invoke ``call_with_retries(fn=fn, policy=policy, log=logger)`` or call_with_retries_async with a standard logging.Logger.
35- On success, the result of calling ``fn`` is returned.
36- By default on exhaustion, call_with_retries* either re-raises the last underlying ``RetryableError.__cause__``, or raises
37 ``RetryError`` (wrapping the last ``RetryableError``), like so:
38 - if ``RetryPolicy.reraise`` is True and the last ``RetryableError.__cause__`` is not None, re-raise the last
39 ``RetryableError.__cause__`` with its original traceback.
40 - Otherwise, raise ``RetryError`` (wrapping the last ``RetryableError``, preserving its ``__cause__`` chain).
41 - The default is ``RetryPolicy.reraise=True``.
43Advanced Configuration:
44-----------------------
45- Tune ``RetryPolicy`` parameters to control maximum retries, sleep bounds, elapsed-time budget, logging, etc.
46- Use ``RetryPolicy.config: RetryConfig`` to control logging settings.
47- Set ``log=None`` to disable logging, or customize ``info_loglevel`` / ``warning_loglevel`` for structured logs.
48- Supply a ``giveup(AttemptOutcome)`` callback to stop retrying based on domain-specific logic (for example, decisions based
49 on time budget/quota or the previous N most recent AttemptOutcome objects (via AttemptOutcome.retry.previous_outcomes).
50- Use the ``any_giveup()`` / ``all_giveup()`` helper to consult more than one callback handler in ``giveup(AttemptOutcome)``.
51- Supply an ``on_exhaustion(AttemptOutcome)`` callback to customize behavior when giving up; it may raise an error or return
52 a fallback value.
54Observability:
55--------------
56- Supply an ``after_attempt(AttemptOutcome)`` callback to collect per-attempt metrics such as success flag,
57 exhausted/terminated state, attempt number, total elapsed duration (in nanoseconds), sleep duration (in nanoseconds), etc.
58- ``AttemptOutcome.result`` is either the successful result or the most recent ``RetryableError``, enabling integration with
59 metrics and tracing systems without coupling the retry loop to any specific backend.
60- Supply an ``after_attempt(AttemptOutcome)`` callback to customize logging 100%, if necessary.
61- Use the ``multi_after_attempt()`` helper to invoke more than one callback handler in ``after_attempt(AttemptOutcome)``.
63Expert Configuration:
64---------------------
65- Supply a ``backoff(BackoffContext)`` callback to plug in a custom backoff algorithm (e.g., decorrelated-jitter or
66 retry-after HTTP 429). The default is full-jitter exponential backoff with cap (aka industry standard).
67- Supply a ``before_attempt(Retry)`` callback for optional rate limiting or other forms of internal backpressure.
68- Supply a ``on_retryable_error(AttemptOutcome)`` callback, e.g. to count failures (RetryableError) caught by the retry loop.
69- Set ``RetryPolicy.max_previous_outcomes > 0`` to pass the N most recent AttemptOutcome objects to callbacks (default is 0).
70- If ``RetryPolicy.max_previous_outcomes > 0``, you can use ``RetryableError(..., attachment=...)`` to carry domain-specific
71 state from a failed attempt to the next attempt via ``retry.previous_outcomes``. This pattern helps if attempt N+1 is a
72 function of attempt N or all prior attempts (e.g., switching endpoints or resuming from an offset).
73- Use ``[Async]RetryTemplate`` as a bag-of-knobs configuration template for functions that shall be retried in similar ways.
74- Or package up all knobs plus a ``fn(retry: Retry)`` function into a self-contained auto-retrying higher level function by
75 constructing a ``[Async]RetryTemplate`` object (which is a ``Callable`` function itself).
76- To keep calling code retry-transparent, set ``RetryPolicy.reraise=True`` (the default) *and* raise retryable failures as
77 ``raise RetryableError(...) from exc``. Client code now won't notice whether call_with_retries* is used or not.
78- To make exhaustion observable to calling code, set ``RetryPolicy.reraise=False``: by default call_with_retries* now always
79 raises ``RetryError`` (wrapping the last ``RetryableError``) on exhaustion, so callers now catch ``RetryError`` and can
80 inspect the last underlying exception via ``err.outcome``, ``err.__cause__``, and even ``err.__cause__.__cause__`` when
81 present.
82- Set ``RetryPolicy.timing`` to customize reading the current monotonic time, sleeping and optional async termination.
83- The callback API is powerful enough to easily plug in advanced retry algorithms such as:
84 - Google SRE Client-Side Adaptive Throttling - https://sre.google/sre-book/handling-overload/
85 - gRPC retry throttling - https://grpc.io/docs/guides/retry/
86 - AWS SDK adaptive retry mode - https://docs.aws.amazon.com/sdkref/latest/guide/feature-retry-behavior.html
87 - Circuit breakers - https://martinfowler.com/bliki/CircuitBreaker.html (e.g. via `pybreaker` third-party library)
88 - Rate limiting with Fixed Window, Moving Window, and Sliding Windows (e.g. via `limits` third-party library)
89 - Various example backoff strategies such as decorrelated-jitter or retry-after HTTP 429 "Too Many Requests" responses,
90 etc can be found in test_retry_examples.py.
92Example Usage:
93--------------
94 import logging
95 from bzfs_main.util.retry import Retry, RetryPolicy, RetryableError, call_with_retries
97 def unreliable_operation(retry: Retry) -> str:
98 try:
99 if retry.count < 3:
100 raise ValueError("temporary failure connecting to foo.example.com")
101 return "ok"
102 except ValueError as exc:
103 # Preserve the underlying cause for correct error propagation and logging
104 raise RetryableError(display_msg="connect") from exc
106 retry_policy = RetryPolicy(
107 max_retries=10,
108 min_sleep_secs=0,
109 initial_max_sleep_secs=0.125,
110 max_sleep_secs=10,
111 max_elapsed_secs=60,
112 )
113 log = logging.getLogger(__name__)
114 result: str = call_with_retries(fn=unreliable_operation, policy=retry_policy, log=log)
115 print(result)
117 # Sample log output:
118 # INFO:Retrying connect [1/10] in 8.79ms ...
119 # INFO:Retrying connect [2/10] in 90.1ms ...
120 # INFO:Retrying connect [3/10] in 372ms ...
121 # ok
123Background:
124-----------
125For background on exponential backoff and jitter, see for example
126https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter
127"""
129from __future__ import (
130 annotations,
131)
132import argparse
133import dataclasses
134import functools
135import inspect
136import logging
137import random
138import sys
139import threading
140import time
141import traceback
142from collections.abc import (
143 Awaitable,
144 Iterable,
145 Mapping,
146 Sequence,
147)
148from dataclasses import (
149 dataclass,
150)
151from typing import (
152 TYPE_CHECKING,
153 Any,
154 Callable,
155 Final,
156 Generic,
157 NamedTuple,
158 NoReturn,
159 TypeVar,
160 Union,
161 cast,
162 final,
163)
165if TYPE_CHECKING:
166 import asyncio
168if sys.version_info >= (3, 11):
169 from builtins import (
170 BaseExceptionGroup,
171 )
173from bzfs_main.util.utils import (
174 human_readable_duration,
175)
177# constants:
178INFINITY_MAX_RETRIES: Final[int] = 2**90 - 1 # a number that's essentially infinity for all practical retry purposes
181#############################################################################
182def full_jitter_backoff_strategy(context: BackoffContext) -> tuple[int, int]:
183 """Default implementation of ``backoff`` callback for call_with_retries(); computes delay time before next retry attempt,
184 after failure.
186 Full-jitter picks a random sleep_nanos duration from the range [min_sleep_nanos, curr_max_sleep_nanos] and applies
187 exponential backoff with cap to the next attempt; thread-safe. Typically, min_sleep_nanos is 0 and exponential_base is 2.
188 Example curr_max_sleep_nanos sequence: 125ms --> 250ms --> 500ms --> 1s --> 2s --> 4s --> 8s --> 10s --> 10s...
189 Full-jitter provides optimal balance between reducing server load and minimizing retry latency.
190 """
191 policy: RetryPolicy = context.retry.policy
192 curr_max_sleep_nanos: int = context.curr_max_sleep_nanos
193 if policy.min_sleep_nanos == curr_max_sleep_nanos:
194 sleep_nanos = curr_max_sleep_nanos # perf
195 else:
196 sleep_nanos = context.rng.randint(policy.min_sleep_nanos, curr_max_sleep_nanos) # nanos to delay until next attempt
197 curr_max_sleep_nanos = round(curr_max_sleep_nanos * policy.exponential_base) # exponential backoff
198 curr_max_sleep_nanos = min(curr_max_sleep_nanos, policy.max_sleep_nanos) # ... with cap for next attempt
199 return sleep_nanos, curr_max_sleep_nanos
202def no_giveup(outcome: AttemptOutcome) -> object | None:
203 """Default implementation of ``giveup`` callback for call_with_retries(); never gives up; returning anything other than
204 ``None`` indicates to give up retrying; thread-safe."""
205 return None # don't give up retrying
208def giveup_if_backoff_exceeds_deadline(outcome: AttemptOutcome) -> object | None:
209 """Implementation of ``giveup`` callback for call_with_retries() that gives up if the computed backoff sleep exceeds the
210 ``RetryPolicy.max_elapsed_secs`` deadline; thread-safe."""
211 return (
212 None
213 if outcome.elapsed_nanos + outcome.sleep_nanos < outcome.retry.policy.max_elapsed_nanos
214 else "backoff exceeds deadline"
215 )
218def before_attempt_noop(retry: Retry) -> int:
219 """Default implementation of ``before_attempt`` callback for call_with_retries(); does nothing; thread-safe."""
220 return 0
223def after_attempt_log_failure(outcome: AttemptOutcome) -> None:
224 """Default implementation of ``after_attempt`` callback for call_with_retries(); performs simple logging of retry attempt
225 failures; thread-safe."""
226 retry: Retry = outcome.retry
227 policy: RetryPolicy = retry.policy
228 config: RetryConfig = policy.config
229 if outcome.is_success or retry.log is None or not config.enable_logging:
230 return
231 log: logging.Logger = retry.log
232 assert isinstance(outcome.result, RetryableError)
233 retryable_error: RetryableError = outcome.result
234 if not outcome.is_exhausted:
235 if log.isEnabledFor(config.info_loglevel): # Retrying X in Y ms ...
236 m1: str = config.format_msg(config.display_msg, retryable_error)
237 m2: str = config.format_pair(retry.count + 1, policy.max_retries)
238 m3: str = config.format_duration(outcome.sleep_nanos)
239 log.log(config.info_loglevel, "%s", f"{m1}{m2} in {m3}{config.dots}", extra=config.extra)
240 else:
241 if policy.max_retries > 0 and log.isEnabledFor(config.warning_loglevel) and not outcome.is_terminated:
242 reason: str = "" if outcome.giveup_reason is None else f"{outcome.giveup_reason}; "
243 format_duration: Callable[[int], str] = config.format_duration # lambda: nanos
244 log.log(
245 config.warning_loglevel,
246 "%s",
247 f"{config.format_msg(config.display_msg, retryable_error)}"
248 f"exhausted; giving up because {reason}the last "
249 f"{config.format_pair(retry.count, policy.max_retries)} retries across "
250 f"{config.format_pair(format_duration(outcome.elapsed_nanos), format_duration(policy.max_elapsed_nanos))} "
251 "failed",
252 exc_info=retryable_error if config.exc_info else None,
253 stack_info=config.stack_info,
254 extra=config.extra,
255 )
258def noop(outcome: AttemptOutcome) -> None:
259 """Default implementation of ``on_retryable_error`` callback for call_with_retries(); does nothing; thread-safe."""
262def on_exhaustion_raise(outcome: AttemptOutcome) -> NoReturn:
263 """Default implementation of ``on_exhaustion`` callback for call_with_retries(); always raises; thread-safe."""
264 assert outcome.is_exhausted
265 assert isinstance(outcome.result, RetryableError)
266 retryable_error: RetryableError = outcome.result
267 policy: RetryPolicy = outcome.retry.policy
268 cause: BaseException | None = retryable_error.__cause__
269 if policy.reraise and cause is not None:
270 raise cause.with_traceback(cause.__traceback__)
271 else:
272 raise RetryError(outcome=outcome) from retryable_error
275def _update_previous_outcomes(
276 previous_outcomes: tuple[AttemptOutcome, ...], outcome: AttemptOutcome, policy: RetryPolicy, retry: Retry
277) -> tuple[AttemptOutcome, ...]:
278 """Computes value of previous_outcomes for next retry iteration."""
279 n: int = policy.max_previous_outcomes
280 if n > 0: # outcome will be passed to next attempt via Retry.previous_outcomes
281 if previous_outcomes: # detach to reduce memory footprint
282 outcome = outcome.copy(retry=retry.copy(previous_outcomes=()))
283 if len(previous_outcomes) >= n:
284 evicted: AttemptOutcome = previous_outcomes[0]
285 assert isinstance(evicted.result, RetryableError)
286 _clear_exception_frames(evicted.result) # eviction: drop references to local variables to bound memory footprint
287 previous_outcomes = previous_outcomes[1:]
288 previous_outcomes = previous_outcomes + (outcome,)
289 return previous_outcomes
292def _clear_exception_frames(error: BaseException) -> None:
293 """Clear traceback references to local variables throughout the exception graph of causes, contexts, exception-groups."""
294 # on Python < 3.13 traceback.clear_frames() closes live generators and coroutines, which would be a bug.
295 if sys.version_info >= (3, 13):
296 pending: list[BaseException] = [error]
297 seen: set[int] = set()
298 while pending:
299 current: BaseException = pending.pop()
300 if id(current) in seen:
301 continue # graph cycle protection
302 seen.add(id(current))
303 for linked in (current.__cause__, current.__context__):
304 if linked is not None:
305 pending.append(linked)
306 if isinstance(current, BaseExceptionGroup):
307 pending.extend(current.exceptions)
308 traceback.clear_frames(current.__traceback__)
311#############################################################################
312_T = TypeVar("_T")
315def call_with_retries(
316 fn: Callable[[Retry], _T], # typically a lambda; wraps work and raises RetryableError for failures that shall be retried
317 policy: RetryPolicy, # specifies how ``RetryableError`` shall be retried
318 *,
319 backoff: BackoffStrategy = full_jitter_backoff_strategy, # computes delay time before next retry attempt, after failure
320 giveup: Callable[[AttemptOutcome], object | None] = no_giveup, # stop retrying based on domain-specific logic, e.g. time
321 before_attempt: Callable[[Retry], int] = before_attempt_noop, # e.g. wait due to rate limiting or internal backpressure
322 after_attempt: Callable[[AttemptOutcome], None] = after_attempt_log_failure, # e.g. record metrics and/or custom logging
323 on_retryable_error: Callable[[AttemptOutcome], None] = noop, # e.g. count failures (RetryableError) caught by retry loop
324 on_exhaustion: Callable[[AttemptOutcome], _T] = on_exhaustion_raise, # raise error or return fallback value
325 log: logging.Logger | None = None, # set this to ``None`` to disable logging
326) -> _T:
327 """Runs the function ``fn`` and returns its result; retries on failure as indicated by policy; thread-safe.
329 By default on exhaustion, call_with_retries() either re-raises the last underlying ``RetryableError.__cause__``, or raises
330 ``RetryError`` (wrapping the last ``RetryableError``), like so:
331 - if ``RetryPolicy.reraise`` is True and the last ``RetryableError.__cause__`` is not None, re-raise the last
332 ``RetryableError.__cause__`` with its original traceback.
333 - Otherwise, raise ``RetryError`` (wrapping the last ``RetryableError``, preserving its ``__cause__`` chain).
334 - The default is ``RetryPolicy.reraise=True``.
336 On the exhaustion path, ``on_exhaustion`` will be called exactly once (after the final after_attempt). The default
337 implementation raises as described above; custom ``on_exhaustion`` impls may return a fallback value instead of an error.
338 """
339 rng: random.Random | None = None
340 retry_count: int = 0
341 idle_nanos: int = 0
342 curr_max_sleep_nanos: int = policy.initial_max_sleep_nanos
343 previous_outcomes: tuple[AttemptOutcome, ...] = () # for safety pass *immutable* deque to callbacks
344 timing: RetryTiming = policy.timing
345 sleep: Callable[[int, Retry], None] = timing.sleep
346 monotonic_ns: Callable[[], int] = timing.monotonic_ns
347 call_start_nanos: Final[int] = monotonic_ns()
348 while True:
349 before_attempt_nanos: int = monotonic_ns() if retry_count != 0 else call_start_nanos
350 prev: tuple[AttemptOutcome, ...] = previous_outcomes
351 retry: Retry = Retry(
352 retry_count, call_start_nanos, before_attempt_nanos, before_attempt_nanos, idle_nanos, policy, log, prev
353 )
354 try:
355 if before_attempt is not before_attempt_noop:
356 before_attempt_sleep_nanos: int = before_attempt(retry)
357 assert before_attempt_sleep_nanos >= 0, before_attempt_sleep_nanos
358 if before_attempt_sleep_nanos > 0:
359 sleep(before_attempt_sleep_nanos, retry) # e.g. wait due to rate limiting or internal backpressure
360 attempt_start_nanos: int = monotonic_ns()
361 idle_nanos += attempt_start_nanos - before_attempt_nanos
362 retry = Retry(
363 retry_count, call_start_nanos, before_attempt_nanos, attempt_start_nanos, idle_nanos, policy, log, prev
364 )
365 timing.on_before_attempt(retry)
366 result: _T = fn(retry) # Call the target function and supply retry attempt number and other metadata
367 except RetryableError as retryable_error:
368 elapsed_nanos: int = monotonic_ns() - call_start_nanos
369 is_terminated: Callable[[Retry], bool] = timing.is_terminated
370 giveup_reason: object | None = None
371 sleep_nanos: int = 0
372 outcome = AttemptOutcome(retry, False, False, False, None, elapsed_nanos, sleep_nanos, retryable_error)
373 on_retryable_error(outcome) # e.g. count failures (RetryableError) caught by retry loop
374 if retry_count < policy.max_retries and elapsed_nanos < policy.max_elapsed_nanos:
375 if policy.max_sleep_nanos == 0 and backoff is full_jitter_backoff_strategy:
376 pass # perf: e.g. spin-before-block
377 elif retry_count == 0 and retryable_error.retry_immediately_once:
378 pass # retry once immediately without backoff
379 else: # jitter: default backoff strategy picks random sleep_nanos in [min_sleep_nanos, curr_max_sleep_nanos]
380 rng = _thread_local_rng() if rng is None else rng
381 sleep_nanos, curr_max_sleep_nanos = backoff( # compute delay before next retry attempt, after failure
382 BackoffContext(retry, curr_max_sleep_nanos, rng, elapsed_nanos, retryable_error)
383 )
384 assert sleep_nanos >= 0 and curr_max_sleep_nanos >= 0, (sleep_nanos, curr_max_sleep_nanos)
386 if sleep_nanos > 0:
387 outcome = AttemptOutcome(retry, False, False, False, None, elapsed_nanos, sleep_nanos, retryable_error)
388 if (not is_terminated(retry)) and (giveup_reason := giveup(outcome)) is None:
389 after_attempt(outcome)
390 sleep(sleep_nanos, retry)
391 idle_nanos += sleep_nanos
392 if not is_terminated(retry):
393 previous_outcomes = _update_previous_outcomes(previous_outcomes, outcome, policy, retry)
394 del outcome # help gc
395 retry_count += 1
396 continue # continue retry loop with next attempt
397 else:
398 sleep_nanos = 0
399 outcome = AttemptOutcome(
400 retry, False, True, is_terminated(retry), giveup_reason, elapsed_nanos, sleep_nanos, retryable_error
401 )
402 after_attempt(outcome)
403 return on_exhaustion(outcome) # raise error or return fallback value
404 else: # success
405 if after_attempt is not after_attempt_log_failure:
406 elapsed_nanos = monotonic_ns() - call_start_nanos
407 outcome = AttemptOutcome(retry, True, False, False, None, elapsed_nanos, 0, result)
408 after_attempt(outcome)
409 return result
412async def call_with_retries_async(
413 fn: Callable[[Retry], Awaitable[_T]], # wraps work and raises RetryableError for failures that shall be retried
414 policy: RetryPolicy, # specifies how ``RetryableError`` shall be retried
415 *,
416 backoff: BackoffStrategy = full_jitter_backoff_strategy, # computes delay time before next retry attempt, after failure
417 giveup: Callable[[AttemptOutcome], object | Awaitable] = no_giveup, # stop retrying based on custom logic, e.g. time
418 before_attempt: Callable[[Retry], int | Awaitable[int]] = before_attempt_noop, # e.g rate limiting/internal backpressure
419 after_attempt: Callable[[AttemptOutcome], Awaitable[None] | None] = after_attempt_log_failure, # e.g. metrics/logging
420 on_retryable_error: Callable[[AttemptOutcome], Awaitable[None] | None] = noop, # e.g. count RetryableError failures
421 on_exhaustion: Callable[[AttemptOutcome], _T | Awaitable[_T]] = on_exhaustion_raise, # raise error or return fallback
422 log: logging.Logger | None = None, # set this to ``None`` to disable logging
423) -> _T:
424 """Async version of call_with_retries() with the same semantics except it awaits ``fn``, awaitable ``before_attempt`` /
425 ``after_attempt`` / ``on_retryable_error`` / ``on_exhaustion`` / ``giveup`` results and non-blocking sleep. Note that
426 ``backoff``, and ``RetryTiming.is_terminated`` are not async as they are intentionally fast **synchronous** decisions;
427 they must not block."""
428 rng: random.Random | None = None
429 retry_count: int = 0
430 idle_nanos: int = 0
431 curr_max_sleep_nanos: int = policy.initial_max_sleep_nanos
432 previous_outcomes: tuple[AttemptOutcome, ...] = () # for safety pass *immutable* deque to callbacks
433 timing: RetryTiming = policy.timing
434 sleep: Callable[[int, Retry], Awaitable[None]] = timing.sleep_async
435 monotonic_ns: Callable[[], int] = timing.monotonic_ns
436 call_start_nanos: Final[int] = monotonic_ns()
437 while True:
438 before_attempt_nanos: int = monotonic_ns() if retry_count != 0 else call_start_nanos
439 prev: tuple[AttemptOutcome, ...] = previous_outcomes
440 retry: Retry = Retry(
441 retry_count, call_start_nanos, before_attempt_nanos, before_attempt_nanos, idle_nanos, policy, log, prev
442 )
443 try:
444 if before_attempt is not before_attempt_noop:
445 before_attempt_sleep_nanos: int = await _await_result(before_attempt(retry))
446 assert before_attempt_sleep_nanos >= 0, before_attempt_sleep_nanos
447 if before_attempt_sleep_nanos > 0:
448 await sleep(before_attempt_sleep_nanos, retry) # e.g. wait due to rate limiting or internal backpressure
449 attempt_start_nanos: int = monotonic_ns()
450 idle_nanos += attempt_start_nanos - before_attempt_nanos
451 retry = Retry(
452 retry_count, call_start_nanos, before_attempt_nanos, attempt_start_nanos, idle_nanos, policy, log, prev
453 )
454 timing.on_before_attempt(retry)
455 result: _T = await fn(retry) # Call the target function and supply retry attempt number and other metadata
456 except RetryableError as retryable_error:
457 elapsed_nanos: int = monotonic_ns() - call_start_nanos
458 is_terminated: Callable[[Retry], bool] = timing.is_terminated
459 giveup_reason: object | None = None
460 sleep_nanos: int = 0
461 outcome = AttemptOutcome(retry, False, False, False, None, elapsed_nanos, sleep_nanos, retryable_error)
462 await _await_result(on_retryable_error(outcome)) # e.g. count RetryableError failures
463 if retry_count < policy.max_retries and elapsed_nanos < policy.max_elapsed_nanos:
464 if policy.max_sleep_nanos == 0 and backoff is full_jitter_backoff_strategy:
465 pass # perf: e.g. spin-before-block
466 elif retry_count == 0 and retryable_error.retry_immediately_once:
467 pass # retry once immediately without backoff
468 else: # jitter: default backoff strategy picks random sleep_nanos in [min_sleep_nanos, curr_max_sleep_nanos]
469 rng = _thread_local_rng() if rng is None else rng
470 sleep_nanos, curr_max_sleep_nanos = backoff( # compute delay before next retry attempt, after failure
471 BackoffContext(retry, curr_max_sleep_nanos, rng, elapsed_nanos, retryable_error)
472 )
473 assert sleep_nanos >= 0 and curr_max_sleep_nanos >= 0, (sleep_nanos, curr_max_sleep_nanos)
475 if sleep_nanos > 0:
476 outcome = AttemptOutcome(retry, False, False, False, None, elapsed_nanos, sleep_nanos, retryable_error)
477 if (not is_terminated(retry)) and (giveup_reason := await _await_result(giveup(outcome))) is None:
478 await _await_result(after_attempt(outcome))
479 await sleep(sleep_nanos, retry)
480 idle_nanos += sleep_nanos
481 if not is_terminated(retry):
482 previous_outcomes = _update_previous_outcomes(previous_outcomes, outcome, policy, retry)
483 del outcome # help gc
484 retry_count += 1
485 continue # continue retry loop with next attempt
486 else:
487 sleep_nanos = 0
488 outcome = AttemptOutcome(
489 retry, False, True, is_terminated(retry), giveup_reason, elapsed_nanos, sleep_nanos, retryable_error
490 )
491 await _await_result(after_attempt(outcome))
492 return await _await_result(on_exhaustion(outcome)) # raise error or return fallback value
493 else: # success
494 if after_attempt is not after_attempt_log_failure:
495 elapsed_nanos = monotonic_ns() - call_start_nanos
496 outcome = AttemptOutcome(retry, True, False, False, None, elapsed_nanos, 0, result)
497 await _await_result(after_attempt(outcome))
498 return result
501async def _await_result(result: _T | Awaitable[_T]) -> _T:
502 """Returns a callback result, awaiting it first when it is awaitable."""
503 if inspect.isawaitable(result):
504 return await result
505 else:
506 return result
509def multi_after_attempt(handlers: Iterable[Callable[[AttemptOutcome], None]]) -> Callable[[AttemptOutcome], None]:
510 """Composes independent ``after_attempt`` handlers into one ``call_with_retries(after_attempt=...)`` callback that
511 invokes each handler in order; thread-safe."""
512 handlers = tuple(handlers)
513 if len(handlers) == 1:
514 return handlers[0] # perf
516 def _after_attempt(outcome: AttemptOutcome) -> None:
517 for handler in handlers:
518 handler(outcome)
520 return _after_attempt
523def multi_after_attempt_async(
524 handlers: Iterable[Callable[[AttemptOutcome], Awaitable[None] | None]],
525) -> Callable[[AttemptOutcome], Awaitable[None]]:
526 """Async version of ``multi_after_attempt()``."""
527 handlers = tuple(handlers)
529 async def _after_attempt_async(outcome: AttemptOutcome) -> None:
530 for handler in handlers:
531 await _await_result(handler(outcome))
533 return _after_attempt_async
536def any_giveup(handlers: Iterable[Callable[[AttemptOutcome], object | None]]) -> Callable[[AttemptOutcome], object | None]:
537 """Composes independent ``giveup`` handlers into one ``call_with_retries(giveup=...)`` callback that gives up retrying if
538 *any* handler gives up; that is if any handler returns a non-``None`` reason; thread-safe.
540 Handlers are evaluated in order and short-circuit: On giving up returns the first handler's reason for giving up.
541 """
542 handlers = tuple(handlers)
543 if len(handlers) == 1:
544 return handlers[0] # perf
546 def _giveup(outcome: AttemptOutcome) -> object | None:
547 for handler in handlers:
548 giveup_reason: object | None = handler(outcome)
549 if giveup_reason is not None:
550 return giveup_reason
551 return None # don't give up retrying
553 return _giveup
556def all_giveup(handlers: Iterable[Callable[[AttemptOutcome], object | None]]) -> Callable[[AttemptOutcome], object | None]:
557 """Composes independent ``giveup`` handlers into one ``call_with_retries(giveup=...)`` callback that gives up retrying if
558 *all* handlers give up; that is if all handlers return a non-``None`` reason; thread-safe.
560 Handlers are evaluated in order and short-circuit: stops at first ``None``; else returns the last non-``None`` reason.
561 """
562 handlers = tuple(handlers)
563 if len(handlers) == 1:
564 return handlers[0] # perf
566 def _giveup(outcome: AttemptOutcome) -> object | None:
567 giveup_reason: object | None = None
568 for handler in handlers:
569 giveup_reason = handler(outcome)
570 if giveup_reason is None:
571 return None # don't give up retrying
572 return giveup_reason
574 return _giveup
577#############################################################################
578class RetryableError(Exception):
579 """Indicates that the task that caused the underlying exception can be retried and might eventually succeed;
580 ``call_with_retries()`` will pass this exception to callbacks via ``AttemptOutcome.result``; can be subclassed."""
582 def __init__(
583 self,
584 *exc_args: object, # optional args passed into super().__init__()
585 display_msg: object = None, # for logging
586 retry_immediately_once: bool = False, # retry once immediately without backoff?
587 category: object = None, # optional classification e.g. "CONCURRENCY, "SERVER_ISSUE", "THROTTLING", "TRANSIENT", ...
588 attachment: object = None, # optional domain specific info passed to next attempt via Retry.previous_outcomes if
589 # RetryPolicy.max_previous_outcomes > 0. This helps when retrying is not just 'try again later', but
590 # 'try again differently based on what just happened'.
591 # Examples: switching network endpoints, adjusting per-attempt timeouts, capping retries by error-class, resuming
592 # with a token/offset, maintaining failure history for this invocation of call_with_retries().
593 # Example: 'cap retries to 3 for ECONNREFUSED but 12 for ETIMEDOUT' via attachment=collections.Counter
594 ) -> None:
595 super().__init__(*exc_args)
596 self.display_msg: object = display_msg
597 self.retry_immediately_once: bool = retry_immediately_once
598 self.category: object = category
599 self.attachment: object = attachment
601 def display_msg_str(self) -> str:
602 """Returns the display_msg as a str; for logging."""
603 return "" if self.display_msg is None else str(self.display_msg)
606#############################################################################
607@final
608class RetryError(Exception):
609 """Indicates that retries have been exhausted; the last RetryableError is in RetryError.__cause__."""
611 outcome: Final[AttemptOutcome]
612 """Metadata that describes why and how call_with_retries() gave up."""
614 def __init__(self, outcome: AttemptOutcome) -> None:
615 super().__init__(outcome)
616 self.outcome = outcome
619#############################################################################
620@final
621class Retry(NamedTuple):
622 """Attempt metadata provided to callback functions; includes the current retry attempt number; immutable."""
624 count: int # type: ignore[assignment]
625 """Attempt number; count=0 is the first attempt, count=1 is the second attempt aka first retry."""
627 call_start_time_nanos: int
628 """Value of time.monotonic_ns() at start of call_with_retries() invocation."""
630 before_attempt_start_time_nanos: int
631 """Value of time.monotonic_ns() at start of before_attempt() invocation."""
633 attempt_start_time_nanos: int
634 """Value of time.monotonic_ns() at start of fn() invocation."""
636 idle_nanos: int
637 """Sum of all before_attempt_sleep_nanos() plus AttemptOutcome.sleep_nanos across this call_with_retries() invocation, at
638 the start of fn() invocation."""
640 policy: RetryPolicy
641 """Policy that was passed into call_with_retries()."""
643 log: logging.Logger | None
644 """Logger that was passed into call_with_retries()."""
646 previous_outcomes: Sequence[AttemptOutcome]
647 """History/state of the N=max_previous_outcomes most recent outcomes for the current call_with_retries() invocation."""
649 def copy(self, **override_kwargs: Any) -> Retry:
650 """Creates a new object copying an existing one with the specified fields overridden for customization."""
651 return self._replace(**override_kwargs)
653 def before_attempt_sleep_nanos(self) -> int:
654 """Returns duration between the start of before_attempt() and the start of fn() attempt."""
655 return self.attempt_start_time_nanos - self.before_attempt_start_time_nanos
657 def __repr__(self) -> str:
658 return (
659 f"{type(self).__name__}(count={self.count!r}, call_start_time_nanos={self.call_start_time_nanos!r}, "
660 f"before_attempt_start_time_nanos={self.before_attempt_start_time_nanos!r}, "
661 f"attempt_start_time_nanos={self.attempt_start_time_nanos!r}, idle_nanos={self.idle_nanos!r})"
662 )
664 def __eq__(self, other: object) -> bool:
665 return self is other
667 def __ne__(self, other: object) -> bool:
668 return self is not other
670 def __hash__(self) -> int:
671 return object.__hash__(self)
674#############################################################################
675@final
676class AttemptOutcome(NamedTuple):
677 """Captures per-attempt state for ``after_attempt`` callbacks; immutable."""
679 retry: Retry
680 """Attempt metadata passed into fn(retry)."""
682 is_success: bool
683 """False if fn(retry) raised a RetryableError; True otherwise."""
685 is_exhausted: bool
686 """True if the loop is giving up retrying (possibly even due to is_terminated); False otherwise."""
688 is_terminated: bool
689 """True if the termination predicate has become true; False otherwise."""
691 giveup_reason: object | None
692 """Reason returned by giveup(); None means giveup() was not called or decided to not give up."""
694 elapsed_nanos: int
695 """Total duration between the start of call_with_retries() invocation and the end of this fn() attempt."""
697 sleep_nanos: int
698 """Duration of current sleep period."""
700 result: RetryableError | object
701 """Result of fn(retry); a RetryableError on retryable failure, or some other object on success."""
703 def attempt_elapsed_nanos(self) -> int:
704 """Returns duration between the start of this fn() attempt and the end of this fn() attempt."""
705 return self.elapsed_nanos + self.retry.call_start_time_nanos - self.retry.attempt_start_time_nanos
707 def copy(self, **override_kwargs: Any) -> AttemptOutcome:
708 """Creates a new outcome copying an existing one with the specified fields overridden for customization."""
709 return self._replace(**override_kwargs)
711 def __repr__(self) -> str:
712 return (
713 f"{type(self).__name__}("
714 f"retry={self.retry!r}, "
715 f"is_success={self.is_success!r}, "
716 f"is_exhausted={self.is_exhausted!r}, "
717 f"is_terminated={self.is_terminated!r}, "
718 f"giveup_reason={self.giveup_reason!r}, "
719 f"elapsed_nanos={self.elapsed_nanos!r}, "
720 f"sleep_nanos={self.sleep_nanos!r})"
721 )
723 def __eq__(self, other: object) -> bool:
724 return self is other
726 def __ne__(self, other: object) -> bool:
727 return self is not other
729 def __hash__(self) -> int:
730 return object.__hash__(self)
733#############################################################################
734@final
735class BackoffContext(NamedTuple):
736 """Captures per-backoff state for ``backoff`` callbacks."""
738 retry: Retry
739 """Attempt metadata passed into fn(retry)."""
741 curr_max_sleep_nanos: int
742 """Current maximum duration (in nanoseconds) to sleep before the next retry attempt;
743 Typically: ``RetryPolicy.initial_max_sleep_nanos <= curr_max_sleep_nanos <= RetryPolicy.max_sleep_nanos``."""
745 rng: random.Random
746 """Thread-local random number generator instance."""
748 elapsed_nanos: int
749 """Total duration between the start of call_with_retries() invocation and the end of this fn() attempt."""
751 retryable_error: RetryableError
752 """Result of failed fn(retry) attempt."""
754 def copy(self, **override_kwargs: Any) -> BackoffContext:
755 """Creates a new object copying an existing one with the specified fields overridden for customization."""
756 return self._replace(**override_kwargs)
758 def __repr__(self) -> str:
759 return (
760 f"{type(self).__name__}("
761 f"retry={self.retry!r}, "
762 f"curr_max_sleep_nanos={self.curr_max_sleep_nanos!r}, "
763 f"elapsed_nanos={self.elapsed_nanos!r})"
764 )
766 def __eq__(self, other: object) -> bool:
767 return self is other
769 def __ne__(self, other: object) -> bool:
770 return self is not other
772 def __hash__(self) -> int:
773 return object.__hash__(self)
776BackoffStrategy = Callable[[BackoffContext], tuple[int, int]] # typealias; returns sleep_nanos:int, curr_max_sleep_nanos:int
777"""Strategy that implements a backoff algorithm that reduces server load while minimizing retry latency; default is full
778jitter; various other example backoff strategies such as decorrelated-jitter or retry-after HTTP 429 "Too Many Requests"
779responses, etc can be found in test_retry_examples.py."""
782#############################################################################
783def _default_timing_is_terminated(retry: Retry) -> bool:
784 return False
787def _default_timing_sleep(sleep_nanos: int, retry: Retry) -> None:
788 time.sleep(sleep_nanos / 1_000_000_000)
791async def _default_timing_sleep_asyncio(sleep_nanos: int, retry: Retry) -> None:
792 import asyncio
794 await asyncio.sleep(sleep_nanos / 1_000_000_000)
797def _default_timing_on_before_attempt(retry: Retry) -> None:
798 if retry.policy.timing.is_terminated(retry):
799 raise RetryableError(display_msg="terminated before attempt") from RetryTerminationError()
802@final
803class RetryTerminationError(InterruptedError):
804 """Termination signal raised when retry loop exits before starting the next attempt."""
807@dataclass(frozen=True)
808@final
809class RetryTiming:
810 """Customizable callbacks for reading the current monotonic time, sleeping and optional async termination; immutable."""
812 monotonic_ns: Callable[[], int] = time.monotonic_ns
813 """Returns the system's current monotonic time in nanoseconds."""
815 is_terminated: Callable[[Retry], bool] = _default_timing_is_terminated
816 """Returns whether a predicate has become true; if so causes the retry loop to exit early between attempts; can be used
817 to indicate system shutdown or similar cancellation conditions; default is to always return ``False``; this function
818 should complete quickly without any blocking or sleeping."""
820 sleep: Callable[[int, Retry], None] = _default_timing_sleep
821 """Sleeps N nanoseconds between attempts; override to inject custom sleeping or for early wake-ups; thread-safe."""
823 sleep_async: Callable[[int, Retry], Awaitable[None]] = _default_timing_sleep_asyncio
824 """Sleeps N nanoseconds between attempts; override to inject custom sleeping or for early wake-ups; thread-safe."""
826 on_before_attempt: Callable[[Retry], None] = _default_timing_on_before_attempt
827 """Typically (but not necessarily) raises an error if ``is_terminated()`` is True; otherwise fn() will still run; this
828 function should complete quickly without any blocking or sleeping.
830 To disable this behavior: RetryTiming.make_from(...).copy(on_before_attempt=lambda retry: None).
831 """
833 def copy(self, **override_kwargs: Any) -> RetryTiming:
834 """Creates a new object copying an existing one with the specified fields overridden for customization; thread-
835 safe."""
836 return dataclasses.replace(self, **override_kwargs)
838 @staticmethod
839 def make_from(termination_event: threading.Event | None) -> RetryTiming:
840 """Convenience factory that creates a RetryTiming that performs async termination when termination_event is set."""
841 if termination_event is None:
842 return RetryTiming()
844 def _is_terminated(retry: Retry) -> bool:
845 return termination_event.is_set()
847 def _sleep(sleep_nanos: int, retry: Retry) -> None:
848 termination_event.wait(sleep_nanos / 1_000_000_000) # allow early wakeup on async termination
850 return RetryTiming(is_terminated=_is_terminated, sleep=_sleep)
852 @staticmethod
853 def make_from_asyncio(termination_event: asyncio.Event | None) -> RetryTiming:
854 """Convenience factory that creates a RetryTiming that performs async termination when termination_event is set;
855 Write an analog version of this function if you wish to replace asyncio with Trio or AnyIO or similar."""
856 if termination_event is None:
857 return RetryTiming()
859 def _is_terminated(retry: Retry) -> bool:
860 return termination_event.is_set()
862 async def _sleep_async(sleep_nanos: int, retry: Retry) -> None:
863 import asyncio
865 if sleep_nanos <= 0:
866 await asyncio.sleep(0) # perf: cooperative yield with less overhead than asyncio.wait_for(..., 0)
867 return
868 try:
869 await asyncio.wait_for(termination_event.wait(), timeout=sleep_nanos / 1_000_000_000)
870 except asyncio.TimeoutError:
871 pass # expected
873 return RetryTiming(is_terminated=_is_terminated, sleep_async=_sleep_async)
876#############################################################################
877def _format_msg(display_msg: str, retryable_error: RetryableError) -> str:
878 """Default implementation of ``format_msg`` callback for RetryConfig; creates simple log message; thread-safe."""
879 msg = display_msg + " " if display_msg else ""
880 errmsg: str = retryable_error.display_msg_str()
881 msg = msg + errmsg + " " if errmsg else msg
882 msg = msg if msg else "Retrying "
883 return msg
886def _format_pair(first: object, second: object) -> str:
887 """Default implementation of ``format_pair`` callback for RetryConfig; creates simple log message part; thread-safe."""
888 second = "∞" if INFINITY_MAX_RETRIES == second else second # noqa: SIM300
889 return f"[{first}/{second}]"
892@dataclass(frozen=True)
893@final
894class RetryConfig:
895 """Configures logging for call_with_retries(); all defaults work out of the box; immutable."""
897 display_msg: str = "Retrying" # message prefix for retry log messages
898 dots: str = " ..." # suffix appended to retry log messages
899 format_msg: Callable[[str, RetryableError], str] = _format_msg # lambda: display_msg, retryable_error
900 format_pair: Callable[[object, object], str] = _format_pair # lambda: first, second
901 format_duration: Callable[[int], str] = human_readable_duration # lambda: nanos
902 info_loglevel: int = logging.INFO # loglevel used when not giving up
903 warning_loglevel: int = logging.WARNING # loglevel used when giving up
904 enable_logging: bool = True # set to False to disable logging
905 exc_info: bool = False # passed into Logger.log()
906 stack_info: bool = False # passed into Logger.log()
907 extra: Mapping[str, object] | None = dataclasses.field(default=None, repr=False, compare=False) # passed to Logger.log()
908 context: object = dataclasses.field(default=None, repr=False, compare=False) # optional domain specific info
910 def copy(self, **override_kwargs: Any) -> RetryConfig:
911 """Creates a new config copying an existing one with the specified fields overridden for customization."""
912 return dataclasses.replace(self, **override_kwargs)
915#############################################################################
916@dataclass(frozen=True)
917@final
918class RetryPolicy:
919 """Configuration of maximum retries, sleep bounds, elapsed-time budget, logging, etc for call_with_retries(); immutable.
921 By default uses full jitter which works as follows: The maximum duration to sleep between attempts initially starts with
922 ``initial_max_sleep_secs`` and doubles on each retry, up to the final maximum of ``max_sleep_secs``.
923 Example: 125ms --> 250ms --> 500ms --> 1s --> 2s --> 4s --> 8s --> 10s --> 10s...
924 On each retry a random sleep duration in the range ``[min_sleep_secs, current max]`` is picked.
925 In a nutshell: ``0 <= min_sleep_secs <= initial_max_sleep_secs <= max_sleep_secs``. Typically, min_sleep_secs=0.
926 """
928 max_retries: int = INFINITY_MAX_RETRIES
929 """The maximum number of times ``fn`` will be invoked additionally after the first attempt invocation; must be >= 0."""
931 min_sleep_secs: float = 0
932 """The minimum duration to sleep between any two attempts."""
934 initial_max_sleep_secs: float = 0.125
935 """The initial maximum duration to sleep between any two attempts."""
937 max_sleep_secs: float = 10
938 """The final max duration to sleep between any two attempts; 0 <= min_sleep_secs <= initial_max_sleep_secs <=
939 max_sleep_secs."""
941 max_elapsed_secs: float = 47
942 """``fn`` will not be retried (or not retried anymore) once this much time has elapsed since the initial start of
943 call_with_retries(); set this to 365 * 86400 seconds or similar to effectively disable the time limit."""
945 exponential_base: float = 2
946 """Growth factor (aka multiplier) for backoff algorithm to calculate sleep duration; must be >= 1."""
948 max_elapsed_nanos: int = dataclasses.field(init=False, repr=False) # derived value
949 min_sleep_nanos: int = dataclasses.field(init=False, repr=False) # derived value
950 initial_max_sleep_nanos: int = dataclasses.field(init=False, repr=False) # derived value
951 max_sleep_nanos: int = dataclasses.field(init=False, repr=False) # derived value
953 reraise: bool = True
954 """On exhaustion, the default (``True``) is to re-raise the underlying exception when present."""
956 max_previous_outcomes: int = 0
957 """Pass the N=max_previous_outcomes most recent AttemptOutcome objects to callbacks via Retry.previous_outcomes."""
959 config: RetryConfig = dataclasses.field(default=RetryConfig(), repr=False, compare=False)
960 """Configures logging behavior."""
962 timing: RetryTiming = dataclasses.field(default=RetryTiming(), repr=False)
963 """Customizable callbacks for reading the current monotonic time, sleeping and optional async termination."""
965 context: object = dataclasses.field(default=None, repr=False, compare=False)
966 """Optional domain specific info."""
968 @classmethod
969 def from_namespace(cls, args: argparse.Namespace) -> RetryPolicy:
970 """Factory that reads the policy from argparse.ArgumentParser via args."""
971 return cls(
972 max_retries=getattr(args, "max_retries", INFINITY_MAX_RETRIES),
973 min_sleep_secs=getattr(args, "retry_min_sleep_secs", 0),
974 initial_max_sleep_secs=getattr(args, "retry_initial_max_sleep_secs", 0.125),
975 max_sleep_secs=getattr(args, "retry_max_sleep_secs", 10),
976 max_elapsed_secs=getattr(args, "retry_max_elapsed_secs", 47),
977 exponential_base=getattr(args, "retry_exponential_base", 2),
978 reraise=getattr(args, "retry_reraise", True),
979 max_previous_outcomes=getattr(args, "retry_max_previous_outcomes", 0),
980 config=getattr(args, "retry_config", RetryConfig()),
981 timing=getattr(args, "retry_timing", RetryTiming()),
982 context=getattr(args, "retry_context", None),
983 )
985 @classmethod
986 def no_retries(cls) -> RetryPolicy:
987 """Returns a policy that never retries."""
988 return cls(
989 max_retries=0,
990 min_sleep_secs=0,
991 initial_max_sleep_secs=0,
992 max_sleep_secs=0,
993 max_elapsed_secs=0,
994 )
996 def __post_init__(self) -> None: # validate and compute derived values
997 self._validate_min("max_retries", self.max_retries, 0)
998 self._validate_min("exponential_base", self.exponential_base, 1)
999 self._validate_min("min_sleep_secs", self.min_sleep_secs, 0)
1000 self._validate_min("initial_max_sleep_secs", self.initial_max_sleep_secs, 0)
1001 self._validate_min("max_sleep_secs", self.max_sleep_secs, 0)
1002 self._validate_min("max_elapsed_secs", self.max_elapsed_secs, 0)
1003 object.__setattr__(self, "max_elapsed_nanos", int(self.max_elapsed_secs * 1_000_000_000)) # derived value
1004 min_sleep_nanos: int = int(self.min_sleep_secs * 1_000_000_000)
1005 initial_max_sleep_nanos: int = int(self.initial_max_sleep_secs * 1_000_000_000)
1006 max_sleep_nanos: int = int(self.max_sleep_secs * 1_000_000_000)
1007 max_sleep_nanos = max(min_sleep_nanos, max_sleep_nanos)
1008 initial_max_sleep_nanos = min(max_sleep_nanos, max(min_sleep_nanos, initial_max_sleep_nanos))
1009 object.__setattr__(self, "min_sleep_nanos", min_sleep_nanos) # derived value
1010 object.__setattr__(self, "initial_max_sleep_nanos", initial_max_sleep_nanos) # derived value
1011 object.__setattr__(self, "max_sleep_nanos", max_sleep_nanos) # derived value
1012 self._validate_min("max_previous_outcomes", self.max_previous_outcomes, 0)
1013 assert 0 <= self.min_sleep_nanos <= self.initial_max_sleep_nanos <= self.max_sleep_nanos
1014 if not isinstance(self.reraise, bool):
1015 raise TypeError(f"{type(self).__name__}.reraise must be bool")
1017 def _validate_min(self, attr_name: str, value: float, minimum: float) -> None:
1018 if value < minimum:
1019 raise ValueError(f"Invalid {type(self).__name__}.{attr_name}: must be >= {minimum} but got {value}")
1021 def copy(self, **override_kwargs: Any) -> RetryPolicy:
1022 """Creates a new policy copying an existing one with the specified fields overridden for customization; thread-safe.
1024 Example usage: policy = retry_policy.copy(max_sleep_secs=2, max_elapsed_secs=10)
1025 """
1026 return dataclasses.replace(self, **override_kwargs)
1029#############################################################################
1030def _fn_not_implemented(_retry: Retry) -> NoReturn:
1031 """Default implementation of ``fn`` callback for RetryTemplate; always raises."""
1032 raise NotImplementedError("Provide fn when calling RetryTemplate")
1035NO_LOGGER: Final[logging.Logger] = logging.Logger("NULL") # noqa: LOG001 do not register dummy logger with Logger.manager
1036NO_LOGGER.addHandler(logging.NullHandler()) # prevents lastResort fallback
1037NO_LOGGER.disabled = True
1038NO_LOGGER.propagate = False
1039_R = TypeVar("_R")
1042@dataclass(frozen=True)
1043@final
1044class RetryTemplate(Generic[_T]):
1045 """Convenience class that aggregates all knobs for call_with_retries(); and is itself callable too; immutable."""
1047 fn: Callable[[Retry], _T] = _fn_not_implemented # set this to make the RetryTemplate object itself callable
1048 policy: RetryPolicy = RetryPolicy() # specifies how ``RetryableError`` shall be retried
1049 backoff: BackoffStrategy = full_jitter_backoff_strategy # computes delay time before next retry attempt, after failure
1050 giveup: Callable[[AttemptOutcome], object | None] = no_giveup # stop retrying based on domain-specific logic, e.g. time
1051 before_attempt: Callable[[Retry], int] = before_attempt_noop # e.g. wait due to rate limiting or internal backpressure
1052 after_attempt: Callable[[AttemptOutcome], None] = after_attempt_log_failure # e.g. record metrics and/or custom logging
1053 on_retryable_error: Callable[[AttemptOutcome], None] = noop # e.g. count failures (RetryableError) caught by retry loop
1054 on_exhaustion: Callable[[AttemptOutcome], _T] = on_exhaustion_raise # raise error or return fallback value
1055 log: logging.Logger | None = None # set this to ``None`` to disable logging
1057 def copy(self, **override_kwargs: Any) -> RetryTemplate[_T]:
1058 """Creates a new object copying an existing one with the specified fields overridden for customization; thread-safe.
1060 Example usage: retry_template.copy(policy=policy.copy(max_sleep_secs=2, max_elapsed_secs=10), log=None)
1061 """
1062 return dataclasses.replace(self, **override_kwargs)
1064 def __call__(self) -> _T:
1065 """Invokes ``self.fn`` via the call_with_retries() retry loop using the stored parameters; thread-safe.
1067 Example Usage: result: str = retry_template.copy(fn=...)()
1068 """
1069 return call_with_retries(
1070 fn=self.fn,
1071 policy=self.policy,
1072 backoff=self.backoff,
1073 giveup=self.giveup,
1074 before_attempt=self.before_attempt,
1075 after_attempt=self.after_attempt,
1076 on_retryable_error=self.on_retryable_error,
1077 on_exhaustion=self.on_exhaustion,
1078 log=self.log,
1079 )
1081 def call_with_retries(
1082 self,
1083 fn: Callable[[Retry], _R],
1084 policy: RetryPolicy | None = None,
1085 *,
1086 backoff: BackoffStrategy | None = None,
1087 giveup: Callable[[AttemptOutcome], object | None] | None = None,
1088 before_attempt: Callable[[Retry], int] | None = None,
1089 after_attempt: Callable[[AttemptOutcome], None] | None = None,
1090 on_retryable_error: Callable[[AttemptOutcome], None] | None = None,
1091 on_exhaustion: Callable[[AttemptOutcome], _R] | None = None,
1092 log: logging.Logger | None = None, # pass NO_LOGGER to override template logger and disable logging for this call
1093 ) -> _R:
1094 """Invokes ``fn`` via the call_with_retries() retry loop using the stored or overridden params; thread-safe.
1096 Example Usage: result: str = retry_template.call_with_retries(fn=...)
1097 """
1098 return call_with_retries(
1099 fn=fn,
1100 policy=self.policy if policy is None else policy,
1101 backoff=self.backoff if backoff is None else backoff,
1102 giveup=self.giveup if giveup is None else giveup,
1103 before_attempt=self.before_attempt if before_attempt is None else before_attempt,
1104 after_attempt=self.after_attempt if after_attempt is None else after_attempt,
1105 on_retryable_error=self.on_retryable_error if on_retryable_error is None else on_retryable_error,
1106 on_exhaustion=(
1107 cast(Callable[[AttemptOutcome], _R], self.on_exhaustion) if on_exhaustion is None else on_exhaustion
1108 ),
1109 log=None if log is NO_LOGGER else self.log if log is None else log,
1110 )
1112 def wraps(self, fn: Callable[..., _R]) -> Callable[..., _R]:
1113 """Returns a wrapper function that forwards all arguments to ``fn`` and retries it using this template; thread-safe.
1115 Example Usage:
1116 def fn(x: int) -> int:
1117 return x * 2
1118 func: Callable[[int], int] = retry_template.wraps(fn)
1119 y: int = func(5) # returns 10
1120 """
1122 @functools.wraps(fn)
1123 def wrapped(*args: Any, **kwargs: Any) -> _R:
1124 return self.call_with_retries(fn=lambda _retry: fn(*args, **kwargs))
1126 return wrapped
1129#############################################################################
1130@dataclass(frozen=True)
1131@final
1132class AsyncRetryTemplate(Generic[_T]):
1133 """Async version of ``RetryTemplate`` with the same semantics except it awaits ``fn``, awaitable ``before_attempt`` /
1134 ``after_attempt`` / ``on_retryable_error`` / ``on_exhaustion`` / ``giveup`` results and non-blocking sleep. Note that
1135 ``backoff``, and ``RetryTiming.is_terminated`` are not async as they are intentionally fast **synchronous** decisions;
1136 they must not block."""
1138 fn: Callable[[Retry], Awaitable[_T]] = _fn_not_implemented # set this to make the RetryTemplate object itself callable
1139 policy: RetryPolicy = RetryPolicy() # specifies how ``RetryableError`` shall be retried
1140 backoff: BackoffStrategy = full_jitter_backoff_strategy # computes delay time before next retry attempt, after failure
1141 giveup: Callable[[AttemptOutcome], object | Awaitable] = no_giveup # stop retrying based on custom logic, e.g. time
1142 before_attempt: Callable[[Retry], int | Awaitable[int]] = before_attempt_noop # e.g. rate limiting/internal backpressure
1143 after_attempt: Callable[[AttemptOutcome], Awaitable[None] | None] = after_attempt_log_failure # e.g. metrics/logging
1144 on_retryable_error: Callable[[AttemptOutcome], Awaitable[None] | None] = noop # e.g. count RetryableError failures
1145 on_exhaustion: Callable[[AttemptOutcome], _T | Awaitable[_T]] = on_exhaustion_raise # raise or fallback
1146 log: logging.Logger | None = None # set this to ``None`` to disable logging
1148 def copy(self, **override_kwargs: Any) -> AsyncRetryTemplate[_T]:
1149 """Creates a new object copying an existing one with the specified fields overridden for customization; thread-safe.
1151 Example usage: retry_template.copy(policy=policy.copy(max_sleep_secs=2, max_elapsed_secs=10), log=None)
1152 """
1153 return dataclasses.replace(self, **override_kwargs)
1155 async def __call__(self) -> _T:
1156 """Invokes ``self.fn`` via the call_with_retries_async() retry loop using the stored parameters; thread-safe.
1158 Example Usage: result: str = await retry_template.copy(fn=...)()
1159 """
1160 return await call_with_retries_async(
1161 fn=self.fn,
1162 policy=self.policy,
1163 backoff=self.backoff,
1164 giveup=self.giveup,
1165 before_attempt=self.before_attempt,
1166 after_attempt=self.after_attempt,
1167 on_retryable_error=self.on_retryable_error,
1168 on_exhaustion=self.on_exhaustion,
1169 log=self.log,
1170 )
1172 async def call_with_retries(
1173 self,
1174 fn: Callable[[Retry], Awaitable[_R]],
1175 policy: RetryPolicy | None = None,
1176 *,
1177 backoff: BackoffStrategy | None = None,
1178 giveup: Callable[[AttemptOutcome], object | Awaitable] | None = None,
1179 before_attempt: Callable[[Retry], int | Awaitable[int]] | None = None,
1180 after_attempt: Callable[[AttemptOutcome], Awaitable[None] | None] | None = None,
1181 on_retryable_error: Callable[[AttemptOutcome], Awaitable[None] | None] | None = None,
1182 on_exhaustion: Callable[[AttemptOutcome], _R | Awaitable[_R]] | None = None,
1183 log: logging.Logger | None = None, # pass NO_LOGGER to override template logger and disable logging for this call
1184 ) -> _R:
1185 """Invokes ``fn`` via the call_with_retries_async() retry loop using the stored or overridden params; thread-safe.
1187 Example Usage: result: str = await retry_template.call_with_retries(fn=...)
1188 """
1189 return await call_with_retries_async(
1190 fn=fn,
1191 policy=self.policy if policy is None else policy,
1192 backoff=self.backoff if backoff is None else backoff,
1193 giveup=self.giveup if giveup is None else giveup,
1194 before_attempt=self.before_attempt if before_attempt is None else before_attempt,
1195 after_attempt=self.after_attempt if after_attempt is None else after_attempt,
1196 on_retryable_error=self.on_retryable_error if on_retryable_error is None else on_retryable_error,
1197 on_exhaustion=(
1198 cast(Callable[[AttemptOutcome], Union[_R, Awaitable[_R]]], self.on_exhaustion)
1199 if on_exhaustion is None
1200 else on_exhaustion
1201 ),
1202 log=None if log is NO_LOGGER else self.log if log is None else log,
1203 )
1205 def wraps(self, fn: Callable[..., Awaitable[_R]]) -> Callable[..., Awaitable[_R]]:
1206 """Returns a wrapper function that forwards all arguments to ``fn`` and retries it using this template; thread-safe.
1208 Example Usage:
1209 async def fn(x: int) -> int:
1210 return x * 2
1211 func: Callable[[int], Awaitable[int]] = retry_template.wraps(fn)
1212 y: int = await func(5) # returns 10
1213 """
1215 @functools.wraps(fn)
1216 async def wrapped_async(*args: Any, **kwargs: Any) -> _R:
1217 return await self.call_with_retries(fn=lambda _retry: fn(*args, **kwargs))
1219 return wrapped_async
1222#############################################################################
1223def raise_retryable_error_from(
1224 exc: BaseException,
1225 *,
1226 display_msg: object = None,
1227 retry_immediately_once: bool = False,
1228 category: object = None,
1229 attachment: object = None,
1230) -> NoReturn:
1231 """Convenience function that raises a generic RetryableError that wraps the given underlying exception."""
1232 raise RetryableError(
1233 display_msg=type(exc).__name__ if display_msg is None else display_msg,
1234 retry_immediately_once=retry_immediately_once,
1235 category=category,
1236 attachment=attachment,
1237 ) from exc
1240ExceptionPredicate = Union[bool, Callable[[BaseException], bool]] # Type alias
1243def call_with_exception_handlers(
1244 fn: Callable[[], _T], # typically a lambda
1245 *,
1246 continue_scanning_if_no_predicate_matches: bool = False,
1247 handlers: Mapping[type[BaseException], Sequence[tuple[ExceptionPredicate, Callable[[BaseException], _T]]]],
1248) -> _T:
1249 """Convenience function that calls ``fn`` and returns its result; on exception runs the first matching handler in a per-
1250 exception handler chain; composes independent handlers via predicates into one function, in Event-Predicate-Action style.
1252 Lookup uses the exception type's Method Resolution Order (most-specific class in the exception class hierarchy wins). For
1253 the first class that exists as a key in ``handlers``, its chain is scanned in order. Each chain element is
1254 ``(predicate, handler)`` where ``predicate`` is either ``True`` (always matches), ``False`` (disabled), or
1255 ``predicate(exc) -> bool``. The first matching handler is called with the exception and its return value is returned. If
1256 no predicate matches then, by default, the original exception is re-raised and no less-specific handler chains are
1257 consulted. Set ``continue_scanning_if_no_predicate_matches=True`` to continue scanning exception base classes instead.
1259 Typically (but not necessarily) the handler raises a ``RetryableError``, via ``raise_retryable_error_from`` or similar.
1260 Or it may raise another exception type (which will not be retried), or even return a fallback value instead of raising.
1262 Example: turn transient ssh/zfs command failures into RetryableError for call_with_retries(), including feature flags:
1264 def run_remote(retry: Retry) -> str:
1265 p = subprocess.run(["ssh", "foo.example.com", "zfs", "list", "-H"], text=True, capture_output=True, check=True)
1266 return p.stdout
1268 def fn(retry: Retry) -> str:
1269 return call_with_exception_handlers(
1270 fn=lambda: run_remote(retry),
1271 handlers={
1272 TimeoutError: [(True, raise_retryable_error_from)],
1273 ConnectionResetError: [(True, lambda exc: raise_retryable_error_from(exc, display_msg="ssh reset"))],
1274 subprocess.CalledProcessError: [
1275 (lambda exc: exc.returncode == 255, lambda exc: raise_retryable_error_from(exc, display_msg="ssh error")),
1276 (lambda exc: "cannot receive" in (exc.stderr or ""), lambda exc: raise_retryable_error_from(exc, display_msg="zfs recv")),
1277 ],
1278 OSError: [
1279 (lambda exc: getattr(exc, "errno", None) in {errno.ETIMEDOUT, errno.EHOSTUNREACH},
1280 lambda exc: raise_retryable_error_from(exc, display_msg=f"network: {exc}")),
1281 (False, lambda exc: raise_retryable_error_from(exc, display_msg="disabled handler example")),
1282 ],
1283 },
1284 )
1286 stdout: str = call_with_retries(fn=fn, policy=RetryPolicy(max_retries=3))
1288 Example: return a fallback value (no retry loop required):
1290 def read_optional_file(path: str) -> str:
1291 return call_with_exception_handlers(
1292 fn=lambda: open(path, encoding="utf-8").read(),
1293 handlers={FileNotFoundError: [(True, lambda _exc: "")]},
1294 )
1295 """
1296 try:
1297 return fn()
1298 except BaseException as exc:
1299 for cls in type(exc).__mro__:
1300 handler_chain = handlers.get(cls)
1301 if handler_chain is not None:
1302 for predicate, handler in handler_chain:
1303 if predicate is True or (predicate is not False and predicate(exc)):
1304 return handler(exc)
1305 if not continue_scanning_if_no_predicate_matches:
1306 raise
1307 raise
1310#############################################################################
1311@final
1312class _ThreadLocalRNG(threading.local):
1313 """Caches a per-thread random number generator."""
1315 def __init__(self) -> None:
1316 self.rng: random.Random | None = None
1319_THREAD_LOCAL_RNG: Final[_ThreadLocalRNG] = _ThreadLocalRNG()
1322def _thread_local_rng() -> random.Random:
1323 """Returns a per-thread RNG for backoff jitter; for perf avoids locking and initializing a new random.Random() at high
1324 frequency."""
1325 threadlocal: _ThreadLocalRNG = _THREAD_LOCAL_RNG
1326 rng: random.Random | None = threadlocal.rng
1327 if rng is None:
1328 rng = random.Random() # noqa: S311 jitter isn't security sensitive, and random.SystemRandom.randint() is slow
1329 threadlocal.rng = rng
1330 return rng