File size: 2,600 Bytes
aac350d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 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 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 | """
Retry policy with exponential backoff + jitter.
Used by the orchestrator to wrap provider invocations.
"""
from __future__ import annotations
import asyncio
import random
import time
from dataclasses import dataclass
from typing import Awaitable, Callable, Tuple, Type
from loguru import logger
@dataclass
class RetryPolicy:
"""Exponential backoff with full jitter."""
max_attempts: int = 3
initial_backoff_seconds: float = 0.5
max_backoff_seconds: float = 8.0
retriable_exceptions: Tuple[Type[BaseException], ...] = (
TimeoutError, ConnectionError, OSError,
)
def backoff(self, attempt: int) -> float:
"""Compute backoff for the given attempt (1-indexed)."""
delay = self.initial_backoff_seconds * (2 ** (attempt - 1))
delay = min(delay, self.max_backoff_seconds)
# full jitter
return random.uniform(0, delay)
def with_retry_sync(
fn: Callable,
policy: RetryPolicy,
label: str = "",
on_retry: Callable[[int, Exception], None] | None = None,
):
"""Synchronous retry wrapper."""
attempt = 0
last_exc: Exception | None = None
while attempt < policy.max_attempts:
try:
attempt += 1
return fn()
except policy.retriable_exceptions as e:
last_exc = e
if attempt >= policy.max_attempts:
break
delay = policy.backoff(attempt)
if on_retry:
on_retry(attempt, e)
logger.warning(
f"[retry] {label} attempt {attempt}/{policy.max_attempts} "
f"failed: {e}; sleeping {delay:.2f}s"
)
time.sleep(delay)
raise last_exc # type: ignore
async def with_retry_async(
fn: Callable[[], Awaitable],
policy: RetryPolicy,
label: str = "",
on_retry: Callable[[int, Exception], None] | None = None,
):
"""Asynchronous retry wrapper."""
attempt = 0
last_exc: Exception | None = None
while attempt < policy.max_attempts:
try:
attempt += 1
return await fn()
except policy.retriable_exceptions as e:
last_exc = e
if attempt >= policy.max_attempts:
break
delay = policy.backoff(attempt)
if on_retry:
on_retry(attempt, e)
logger.warning(
f"[retry] {label} attempt {attempt}/{policy.max_attempts} "
f"failed: {e}; sleeping {delay:.2f}s"
)
await asyncio.sleep(delay)
raise last_exc # type: ignore
|