269 lines
11 KiB
Rust
269 lines
11 KiB
Rust
//! complete() 指数退避重试 — 仅非流式同步路径，不动 stream()
|
||
//!
|
||
//! 背景: complete() 是 LLM 一次性请求，网络波动/瞬态 5xx/429 会让整轮对话静默失败。
|
||
//! 各调用方(title.rs/knowledge_inject.rs/project.rs/ai_node.rs)虽已有人工兜底降级，
|
||
//! 但重试能在不侵入上层的前提下吸收瞬态错误，对低频后台调用零风险。
|
||
//!
|
||
//! 策略:
|
||
//! - 最多 3 次尝试(初次 + 2 次重试)，指数退避 1s → 2s → 4s + ±20% jitter
|
||
//! - 总挂钟预算 30s(含 sleep + 各次请求耗时)，超预算直接放弃重试返回最后错误
|
||
//! - 仅对可重试错误重试: timeout / connect / 5xx / 429; 不重试 4xx(非429) / 参数错 / 鉴权错
|
||
//! - 不引入 reqwest-retry / tower: complete 路径简单，手写 loop 零依赖、可调试、不与 SSE/双层 Semaphore 耦合
|
||
//!
|
||
//! 流式 stream() 路径不进入此模块(流式长生成不能整体 timeout，重试会丢增量)。
|
||
//!
|
||
//! 重试期间不释放 Semaphore permit(已在调用方持有)，对并发池有挤占 —— 但 complete 调用低频可接受。
|
||
|
||
use std::future::Future;
|
||
use std::time::{Duration, SystemTime};
|
||
|
||
use tracing::warn;
|
||
|
||
/// 最多尝试次数(含初次)。B-260616-07: 3 次 = 初次 + 2 次重试。
|
||
///
|
||
/// 配置化 TODO: 未来接入 per-provider config(`AiProviderRecord.config` JSON)或全局开关
|
||
/// (`useSetting('df-ai-max-retries')`)时改为读取配置。当前低频后台调用，常量足够。
|
||
pub const MAX_COMPLETE_ATTEMPTS: u32 = 3;
|
||
|
||
/// 退避基准(秒): 第 n 次重试前 sleep `BASE_BACKOFF_SECS * 2^(n-1)`。
|
||
/// 即 1s → 2s → 4s。
|
||
const BASE_BACKOFF_SECS: u64 = 1;
|
||
|
||
/// 整体重试挂钟预算(含 sleep 与各次请求耗时)。超过即放弃，返回最后错误。
|
||
/// 与单请求 60s timeout 配合: 最坏 3 次 × 60s = 180s，30s 预算会在此之前主动止损。
|
||
const MAX_TOTAL_BUDGET: Duration = Duration::from_secs(30);
|
||
|
||
/// jitter 上限(相对 base 的 ±比例)。避免重试风暴对齐。
|
||
const JITTER_RATIO: f64 = 0.2;
|
||
|
||
/// 一次尝试的分类结果 —— 在 anyhow 不透明化前决定是否值得重试。
|
||
///
|
||
/// CR-30-1: 暴露给 agentic.rs 流前重试复用(Init 失败分类 Fatal/Retryable 决定是否重试)。
|
||
#[derive(Debug)]
|
||
pub enum AttemptOutcome<T> {
|
||
/// 成功，携带结果。
|
||
Ok(T),
|
||
/// 可重试错误(timeout / connect / 5xx / 429)。携带人类可读错误描述。
|
||
Retryable(String),
|
||
/// 不可重试错误(4xx 非 429 / 鉴权 / 参数错 / 解析错)。携带描述，直接放弃。
|
||
Fatal(String),
|
||
}
|
||
|
||
/// 判定 reqwest::Error 是否可重试(仅 timeout / connect 类)。
|
||
///
|
||
/// body 解析错(reqwest::Error::is_decode)与请求构造错(is_builder)属 Fatal，不重试。
|
||
pub fn is_reqwest_error_retryable(e: &reqwest::Error) -> bool {
|
||
e.is_timeout() || e.is_connect() || (e.is_request() && !e.is_builder())
|
||
}
|
||
|
||
/// 判定 HTTP 状态码是否可重试: 5xx 与 429 重试，其余(含 4xx)不重试。
|
||
///
|
||
/// CR-30-1: 暴露给 agentic.rs 流前重试复用(4xx Fatal 立即放弃,5xx/429 重试)。
|
||
pub fn is_status_retryable(status: u16) -> bool {
|
||
status == 429 || (500..600).contains(&status)
|
||
}
|
||
|
||
/// 指数退避 + jitter: 返回第 `attempt`(1-based)次重试前应 sleep 的时长。
|
||
/// attempt=1 → ~1s, attempt=2 → ~2s, attempt=3 → ~4s，各 ±20% jitter。
|
||
///
|
||
/// jitter 用 SystemTime 纳秒取模生成(无依赖)，避免多客户端同步重试风暴。
|
||
/// 以毫秒粒度计算后向下取整(避免秒级截断把 0.9s 砍成 0)。
|
||
///
|
||
/// CR-30-1: 暴露 pub 供 src-tauri/agentic.rs 流前重试复用(对齐决策 F-260616-07 a1
|
||
/// "复用 retry.rs backoff_delay 退避 1s→2s→4s+jitter"),避免重写退避逻辑。
|
||
pub fn backoff_delay(attempt: u32) -> Duration {
|
||
let base_ms = BASE_BACKOFF_SECS.saturating_mul(1u64 << (attempt - 1)) * 1000;
|
||
// 纳秒 → [0, 2000) 区间，再映射到 [-1.0, +1.0) 比例
|
||
let nanos = SystemTime::now()
|
||
.duration_since(SystemTime::UNIX_EPOCH)
|
||
.map(|d| d.subsec_nanos() as u64)
|
||
.unwrap_or(0);
|
||
let jitter_ratio = (nanos % 2000) as f64 / 1000.0 - 1.0; // [-1.0, 1.0)
|
||
let factor = 1.0 + jitter_ratio * JITTER_RATIO; // [0.8, 1.2]
|
||
let ms = (base_ms as f64 * factor).max(0.0) as u64;
|
||
Duration::from_millis(ms)
|
||
}
|
||
|
||
/// 对一个返回 `AttemptOutcome` 的异步操作做指数退避重试。
|
||
///
|
||
/// `attempt_fn` 每次被调用执行一次尝试并返回分类结果。
|
||
/// 达到 `MAX_COMPLETE_ATTEMPTS` 或超出 `MAX_TOTAL_BUDGET` 或遇 Fatal 即终止。
|
||
/// 终止时若仍失败，返回最后一次的 Retryable/Fatal 描述作为 `anyhow::Error`。
|
||
///
|
||
/// 调用方约定: `attempt_fn` 内部应保留完整请求重发能力(每次重建 RequestBuilder)，
|
||
/// 因为 reqwest::RequestBuilder 一次 `.send()` 后不可复用。
|
||
///
|
||
/// CR-30-1: 仍 pub(crate)(complete() 内部用),流前重试 stream_recv/agentic 不走此函数
|
||
/// (流式 request 不可整体 retry_with_backoff 包裹,需在 agentic loop 内手写循环复用
|
||
/// backoff_delay + is_status_retryable)。
|
||
pub(crate) async fn retry_with_backoff<T, F, Fut>(
|
||
label: &str,
|
||
attempt_fn: F,
|
||
) -> anyhow::Result<T>
|
||
where
|
||
F: Fn(u32) -> Fut,
|
||
Fut: Future<Output = AttemptOutcome<T>>,
|
||
{
|
||
let deadline = tokio::time::Instant::now() + MAX_TOTAL_BUDGET;
|
||
let mut last_err = String::from("未尝试");
|
||
|
||
for attempt in 1..=MAX_COMPLETE_ATTEMPTS {
|
||
match attempt_fn(attempt).await {
|
||
AttemptOutcome::Ok(v) => return Ok(v),
|
||
AttemptOutcome::Fatal(msg) => {
|
||
// 不可重试: 立即放弃，不浪费预算
|
||
return Err(anyhow::anyhow!("{}", msg));
|
||
}
|
||
AttemptOutcome::Retryable(msg) => {
|
||
last_err = msg;
|
||
let is_last = attempt == MAX_COMPLETE_ATTEMPTS;
|
||
if is_last {
|
||
warn!(
|
||
label = label,
|
||
attempt = attempt,
|
||
"重试已达上限({}次)，放弃", MAX_COMPLETE_ATTEMPTS
|
||
);
|
||
break;
|
||
}
|
||
let now = tokio::time::Instant::now();
|
||
if now >= deadline {
|
||
warn!(
|
||
label = label,
|
||
attempt = attempt,
|
||
"重试预算({:?})已耗尽，放弃", MAX_TOTAL_BUDGET
|
||
);
|
||
break;
|
||
}
|
||
let delay = backoff_delay(attempt).min(deadline.saturating_duration_since(now));
|
||
warn!(
|
||
label = label,
|
||
attempt = attempt,
|
||
delay_ms = delay.as_millis() as u64,
|
||
err = %last_err,
|
||
"可重试错误，退避后重试"
|
||
);
|
||
tokio::time::sleep(delay).await;
|
||
}
|
||
}
|
||
}
|
||
|
||
Err(anyhow::anyhow!("{} 重试耗尽: {}", label, last_err))
|
||
}
|
||
|
||
// ============================================================
|
||
// 单测(纯逻辑，不发 HTTP)
|
||
// ============================================================
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use std::sync::atomic::{AtomicU32, Ordering};
|
||
use std::sync::Arc;
|
||
|
||
#[test]
|
||
fn backoff_is_monotonic_within_jitter() {
|
||
// base 序列: 1, 2, 4 秒; jitter ±20% 后仍在 [0.8,4.8] 范围
|
||
for attempt in 1..=3 {
|
||
let d = backoff_delay(attempt);
|
||
let base_ms = (BASE_BACKOFF_SECS * (1u64 << (attempt - 1))) * 1000;
|
||
let lo = (base_ms as f64 * (1.0 - JITTER_RATIO)) as u64;
|
||
let hi = (base_ms as f64 * (1.0 + JITTER_RATIO)) as u64;
|
||
assert!(
|
||
d.as_millis() as u64 >= lo && d.as_millis() as u64 <= hi,
|
||
"attempt={} delay={:?} 应在 [{},{}]ms",
|
||
attempt,
|
||
d,
|
||
lo,
|
||
hi
|
||
);
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn status_classification() {
|
||
assert!(is_status_retryable(429));
|
||
assert!(is_status_retryable(500));
|
||
assert!(is_status_retryable(503));
|
||
assert!(!is_status_retryable(400));
|
||
assert!(!is_status_retryable(401));
|
||
assert!(!is_status_retryable(403));
|
||
assert!(!is_status_retryable(404));
|
||
}
|
||
|
||
/// 初次成功 → 不重试
|
||
#[tokio::test]
|
||
async fn first_attempt_ok() {
|
||
let calls = Arc::new(AtomicU32::new(0));
|
||
let calls_clone = calls.clone();
|
||
let r = retry_with_backoff("test", move |_| {
|
||
let c = calls_clone.clone();
|
||
async move {
|
||
c.fetch_add(1, Ordering::SeqCst);
|
||
AttemptOutcome::Ok(42u32)
|
||
}
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(r, 42);
|
||
assert_eq!(calls.load(Ordering::SeqCst), 1);
|
||
}
|
||
|
||
/// 第二次成功 → 恰好 2 次调用
|
||
#[tokio::test]
|
||
async fn retry_once_then_ok() {
|
||
let calls = Arc::new(AtomicU32::new(0));
|
||
let calls_clone = calls.clone();
|
||
let r = retry_with_backoff("test", move |_| {
|
||
let c = calls_clone.clone();
|
||
async move {
|
||
let n = c.fetch_add(1, Ordering::SeqCst) + 1;
|
||
if n == 1 {
|
||
AttemptOutcome::Retryable("timeout".into())
|
||
} else {
|
||
AttemptOutcome::Ok(7u32)
|
||
}
|
||
}
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(r, 7);
|
||
assert_eq!(calls.load(Ordering::SeqCst), 2);
|
||
}
|
||
|
||
/// 全部 Retryable → 用尽 3 次，返回错误
|
||
#[tokio::test]
|
||
async fn all_retryable_exhausts() {
|
||
let calls = Arc::new(AtomicU32::new(0));
|
||
let calls_clone = calls.clone();
|
||
let r = retry_with_backoff("test", move |_| {
|
||
let c = calls_clone.clone();
|
||
async move {
|
||
c.fetch_add(1, Ordering::SeqCst);
|
||
AttemptOutcome::<u32>::Retryable("503".into())
|
||
}
|
||
})
|
||
.await;
|
||
assert!(r.is_err());
|
||
assert!(r.unwrap_err().to_string().contains("重试耗尽"));
|
||
assert_eq!(calls.load(Ordering::SeqCst), MAX_COMPLETE_ATTEMPTS);
|
||
}
|
||
|
||
/// Fatal 立即放弃，不重试
|
||
#[tokio::test]
|
||
async fn fatal_aborts_immediately() {
|
||
let calls = Arc::new(AtomicU32::new(0));
|
||
let calls_clone = calls.clone();
|
||
let r = retry_with_backoff("test", move |_| {
|
||
let c = calls_clone.clone();
|
||
async move {
|
||
c.fetch_add(1, Ordering::SeqCst);
|
||
AttemptOutcome::<u32>::Fatal("401".into())
|
||
}
|
||
})
|
||
.await;
|
||
assert!(r.is_err());
|
||
assert!(r.unwrap_err().to_string().contains("401"));
|
||
assert_eq!(calls.load(Ordering::SeqCst), 1, "Fatal 不应触发重试");
|
||
}
|
||
}
|