"""重试策略:指数退避 + ±25% 抖动,最多重试 1 次。 只对 RETRYABLE_CODES 中的错误码重试,避免对永久错误做无意义重试。 退避公式:base_delay * (2 ** attempt) * jitter - attempt=0: base_delay * 1 * jitter(首次重试前等待) - attempt=1: base_delay * 2 * jitter jitter 在 [0.75, 1.25] 区间均匀分布,避免雷同请求同时重试。 """ from __future__ import annotations import logging import random from woc_bridge.ui.errors import RETRYABLE_CODES, is_retryable logger = logging.getLogger("woc-bridge") class RetryPolicy: """重试策略。 Attributes: max_attempts: 最大尝试次数(含首次),默认 2(即最多重试 1 次) base_delay: 基础退避秒数,默认 1.0 max_delay: 最大退避秒数,默认 30.0 """ def __init__( self, max_attempts: int = 2, base_delay: float = 1.0, max_delay: float = 30.0, ) -> None: """初始化重试策略。 Args: max_attempts: 最大尝试次数(含首次),>=1 base_delay: 基础退避秒数 max_delay: 最大退避秒数(退避不会超过此值) """ if max_attempts < 1: raise ValueError("max_attempts must be >= 1") self.max_attempts = max_attempts self.base_delay = base_delay self.max_delay = max_delay def should_retry(self, error_code: str, attempt: int) -> bool: """是否应该重试。 条件:attempt < max_attempts - 1 AND error_code in RETRYABLE_CODES。 attempt 为重试序号(0 表示首次失败后考虑第一次重试); max_attempts 含首次尝试,故允许的重试次数为 max_attempts - 1。 Args: error_code: 错误码(如 SEND_TIMEOUT / WINDOW_NOT_FOUND) attempt: 当前已尝试次数(0 表示首次失败,考虑第一次重试) Returns: True 表示应重试 """ if attempt >= self.max_attempts - 1: return False return is_retryable(error_code) def get_delay(self, attempt: int) -> float: """计算退避秒数。 指数退避:base_delay * (2 ** attempt) 抖动:× [0.75, 1.25] 均匀分布 上限:max_delay Args: attempt: 当前已尝试次数(0 表示首次失败后的第一次重试前等待) Returns: 退避秒数 """ exponential = self.base_delay * (2 ** attempt) jitter = random.uniform(0.75, 1.25) delay = min(exponential * jitter, self.max_delay) return delay