Kris
15fc62d478
- 新增六层UI自动化架构：从Backend到Capabilities的完整分层实现 - 添加WeChat 4.0分辨率适配Profile与图像模板资源 - 实现幂等缓存、熔断器、重试策略、链路追踪与监控指标 - 新增头像下载安全校验、发布朋友圈路径白名单防护 - 优化密钥缓存、DB校验逻辑与初始化流程 - 补充完整错误码体系与启动清场机制
84 lines
2.6 KiB
Python
84 lines
2.6 KiB
Python
"""重试策略：指数退避 + ±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
|