Rust Tokio 超时与重试退避

功能说明

retry_with_deadline 最多尝试三次，每次用 tokio::time::timeout 包裹 flaky_fetch，预算从 160ms 递增到 240ms。模拟服务前两次在 80ms 后返回 503，第三次成功。每次失败后按 10ms、20ms 线性退避，全部失败则返回 FetchError::Exhausted，携带尝试次数与最后一次原因。AtomicUsize 记录真实调用次数，便于验证没有多调或漏调。

关键代码

timeout 返回 Result<Result<String, String>, Elapsed>，代码用两层匹配区分“调用失败”和“超时”。超时被视为可重试错误，而不是立即返回，这样短暂阻塞不会直接让整批任务失败。预算使用 u64::from(attempt) 计算，避免整数类型不匹配。退避只在还有下一次尝试时执行。最终错误是枚举而非字符串，调用方可以稳定匹配。

使用注意

该实现没有总截止时间，最坏情况下总耗时是各次预算与退避之和；生产代码应再包一层总 timeout 或传入 deadline。没有随机抖动，多个客户端可能同时重试造成惊群。重试只适用于幂等操作，写请求需要幂等键。异步取消是安全的，但被取消的 future 可能已经产生副作用。真实场景还应区分 4xx 等不可重试错误。

验证方式

运行 cargo run，预期输出 success after 3 calls: payload，说明前两次失败都被正确重试。把 max_attempts 改为 2，应输出 gave up after 2 attempts 并带上第二次的 503 信息。可将某次 flaky_fetch 延迟调到超过预算，验证超时分支同样进入重试，并统计总尝试次数。

代码示例

use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;

use tokio::time::{sleep, timeout};

#[derive(Debug)]
enum FetchError {
    Exhausted { attempts: u32, last: String },
}

async fn flaky_fetch(calls: &AtomicUsize) -> Result<String, String> {
    let call = calls.fetch_add(1, Ordering::SeqCst) + 1;
    match call {
        1 | 2 => {
            sleep(Duration::from_millis(80)).await;
            Err(format!("upstream returned 503 on call {call}"))
        }
        3 => Ok("payload".to_string()),
        other => Err(format!("unexpected call {other}")),
    }
}

async fn retry_with_deadline(
    calls: &AtomicUsize,
    max_attempts: u32,
) -> Result<String, FetchError> {
    let mut last = String::from("no attempt executed");

    for attempt in 1..=max_attempts {
        let budget = Duration::from_millis(120 + u64::from(attempt) * 40);
        match timeout(budget, flaky_fetch(calls)).await {
            Ok(Ok(payload)) => return Ok(payload),
            Ok(Err(message)) => last = message,
            Err(_) => last = format!("attempt {attempt} exceeded {budget:?}"),
        }

        if attempt < max_attempts {
            let backoff = Duration::from_millis(10 * u64::from(attempt));
            sleep(backoff).await;
        }
    }

    Err(FetchError::Exhausted {
        attempts: max_attempts,
        last,
    })
}

#[tokio::main]
async fn main() {
    let calls = AtomicUsize::new(0);
    match retry_with_deadline(&calls, 3).await {
        Ok(payload) => println!(
            "success after {} calls: {payload}",
            calls.load(Ordering::SeqCst)
        ),
        Err(FetchError::Exhausted { attempts, last }) => {
            println!("gave up after {attempts} attempts: {last}");
        }
    }
}