Coverage for src/taipanstack/resilience/adaptive/adaptive_retry.py: 100%

1""" 

2Adaptive Retry — learns optimal backoff from runtime outcomes. 

3 

4Tracks recent retry outcomes in a rolling window and computes 

5the best delay for each attempt level, favouring delays that 

6historically led to successful retries. 

7""" 

8 

9from __future__ import annotations 

10 

11import logging 

12import statistics 

13import threading 

14from collections import defaultdict, deque 

15from dataclasses import dataclass 

16 

17from taipanstack.resilience.retry import RetryConfig 

18 

19logger = logging.getLogger("taipanstack.resilience.adaptive.retry") 

20 

21 

22@dataclass(frozen=True) 

23class RetryMetrics: 

24 """Snapshot of adaptive retry metrics. 

25 

26 Attributes: 

27 success_rate: Overall success rate (0.0 - 1.0). 

28 avg_delay: Average delay across all successful retries. 

29 p95_delay: 95th percentile delay. 

30 total_outcomes: Total tracked outcomes. 

31 

32 """ 

33 

34 success_rate: float 

35 avg_delay: float 

36 p95_delay: float 

37 total_outcomes: int 

38 

39 

40@dataclass(frozen=True) 

41class _Outcome: 

42 """Record of a single retry outcome.""" 

43 

44 attempt: int 

45 success: bool 

46 elapsed: float 

47 

48 

49class AdaptiveRetry: 

50 """Retry strategy that learns optimal delays from outcomes. 

51 

52 Maintains per-attempt-level statistics and returns the delay 

53 that historically led to successful retries at that attempt 

54 level. 

55 

56 Args: 

57 min_delay: Minimum delay in seconds. 

58 max_delay: Maximum delay in seconds. 

59 window_size: Number of recent outcomes to track. 

60 max_attempts: Default max attempts for ``to_retry_config()``. 

61 

62 Example: 

63 >>> ar = AdaptiveRetry(min_delay=0.1, max_delay=30.0) 

64 >>> ar.record_outcome(attempt=1, success=True, elapsed=0.5) 

65 >>> delay = ar.get_delay(attempt=1) 

66 

67 """ 

68 

69 def __init__( 

70 self, 

71 *, 

72 min_delay: float = 0.1, 

73 max_delay: float = 60.0, 

74 window_size: int = 50, 

75 max_attempts: int = 3, 

76 ) -> None: 

77 """Initialize the adaptive retry. 

78 

79 Args: 

80 min_delay: Minimum delay. 

81 max_delay: Maximum delay. 

82 window_size: Rolling window size. 

83 max_attempts: Default max attempts. 

84 

85 """ 

86 self._min_delay = min_delay 

87 self._max_delay = max_delay 

88 self._max_attempts = max_attempts 

89 self._lock = threading.Lock() 

90 

91 # Per-attempt deque of outcomes 

92 self._outcomes: deque[_Outcome] = deque(maxlen=window_size) 

93 # Per-attempt successful delays 

94 self._success_delays: dict[int, deque[float]] = defaultdict( 

95 lambda: deque(maxlen=window_size), 

96 ) 

97 

98 def record_outcome( 

99 self, 

100 attempt: int, 

101 success: bool, 

102 elapsed: float, 

103 ) -> None: 

104 """Record a retry outcome. 

105 

106 Args: 

107 attempt: Attempt number (1-indexed). 

108 success: Whether the attempt succeeded. 

109 elapsed: Time elapsed before this attempt was made. 

110 

111 """ 

112 outcome = _Outcome(attempt=attempt, success=success, elapsed=elapsed) 

113 with self._lock: 

114 self._outcomes.append(outcome) 

115 if success: 

116 self._success_delays[attempt].append(elapsed) 

117 

118 def get_delay(self, attempt: int) -> float: 

119 """Get the learned optimal delay for this attempt level. 

120 

121 If there is historical data for this attempt level, returns 

122 the median of successful delays. Otherwise uses exponential 

123 backoff with the configured bounds. 

124 

125 Args: 

126 attempt: Attempt number (1-indexed). 

127 

128 Returns: 

129 Delay in seconds. 

130 

131 """ 

132 with self._lock: 

133 delays = list(self._success_delays.get(attempt, [])) 

134 

135 if delays: 

136 # Use median of successful delays as the optimal delay 

137 learned = statistics.median(delays) 

138 return max(self._min_delay, min(learned, self._max_delay)) 

139 

140 # Fallback: exponential backoff 

141 fallback_delay = self._min_delay * (2.0 ** (attempt - 1)) 

142 return max(self._min_delay, min(fallback_delay, self._max_delay)) 

143 

144 def to_retry_config(self) -> RetryConfig: 

145 """Export current state as a standard ``RetryConfig``. 

146 

147 Uses the learned initial delay (attempt=1) if available. 

148 

149 Returns: 

150 A ``RetryConfig`` snapshot. 

151 

152 """ 

153 initial = self.get_delay(1) 

154 return RetryConfig( 

155 max_attempts=self._max_attempts, 

156 initial_delay=initial, 

157 max_delay=self._max_delay, 

158 jitter=False, 

159 ) 

160 

161 def _calculate_delay_metrics(self, all_delays: list[float]) -> tuple[float, float]: 

162 """Calculate average and p95 delays from a list of delays.""" 

163 if not all_delays: 

164 return 0.0, 0.0 

165 

166 avg_delay = statistics.mean(all_delays) 

167 sorted_delays = sorted(all_delays) 

168 idx = int(len(sorted_delays) * 0.95) 

169 p95_delay = sorted_delays[min(idx, len(sorted_delays) - 1)] 

170 return avg_delay, p95_delay 

171 

172 @property 

173 def metrics(self) -> RetryMetrics: 

174 """Snapshot of current adaptive retry metrics.""" 

175 with self._lock: 

176 total = len(self._outcomes) 

177 successes = sum(1 for o in self._outcomes if o.success) 

178 all_delays = [o.elapsed for o in self._outcomes] 

179 

180 success_rate = successes / total if total > 0 else 1.0 

181 avg_delay, p95_delay = self._calculate_delay_metrics(all_delays) 

182 

183 return RetryMetrics( 

184 success_rate=success_rate, 

185 avg_delay=avg_delay, 

186 p95_delay=p95_delay, 

187 total_outcomes=total, 

188 )