tiphys.utils

13@dataclass
14class BackoffConfig:
15    """Configuration for exponential backoff."""
16
17    initial_delay: float = 1.0
18    max_delay: float = 60.0
19    multiplier: float = 2.0
20    jitter: float = 0.5  # +/- 50% randomization

23class ExponentialBackoff:
24    """
25    Exponential backoff with jitter for retry logic.
26
27    Usage:
28        backoff = ExponentialBackoff()
29        while not connected:
30            try:
31                await connect()
32                backoff.reset()
33            except ConnectionError:
34                await backoff.wait()
35    """
36
37    def __init__(self, config: BackoffConfig | None = None) -> None:
38        """
39        Initialize exponential backoff.
40
41        Args:
42            config: Backoff configuration. Uses defaults if not provided.
43        """
44        self.config = config or BackoffConfig()
45        self._attempt = 0
46
47    @property
48    def attempt(self) -> int:
49        """Current attempt number (0-indexed)."""
50        return self._attempt
51
52    def reset(self) -> None:
53        """Reset backoff after successful operation."""
54        self._attempt = 0
55
56    def next_delay(self) -> float:
57        """
58        Calculate the next delay without waiting.
59
60        Returns:
61            The delay duration in seconds.
62        """
63        # Calculate base delay with exponential growth
64        base_delay = min(
65            self.config.initial_delay * (self.config.multiplier**self._attempt),
66            self.config.max_delay,
67        )
68
69        # Add jitter (+/- jitter%)
70        jitter_range = base_delay * self.config.jitter
71        delay = base_delay + random.uniform(-jitter_range, jitter_range)
72
73        # Ensure minimum delay of 100ms
74        return max(0.1, delay)
75
76    async def wait(self) -> float:
77        """
78        Wait for the appropriate backoff duration.
79
80        Increments the attempt counter after waiting.
81
82        Returns:
83            The actual delay used in seconds.
84        """
85        delay = self.next_delay()
86        self._attempt += 1
87        await asyncio.sleep(delay)
88        return delay
89
90    def __repr__(self) -> str:
91        return f"ExponentialBackoff(attempt={self._attempt}, config={self.config})"