orchestrator/retry.py

"""
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