退避与重试预算

主线位置：不在两条主线上。它们是 Policy 的零件：预算在 Policy::retry 里决定「还能不能重试」，退避给出 Waiting 状态要等的那个 Sleep。重试状态机本身见 Policy 与重试状态机。

flowchart TB
  res[内层返回结果] --> pol[Policy::retry]
  pol --> ok{算成功吗}
  ok -->|成功| dep["budget.deposit()"]
  dep --> none1[返回 None]
  ok -->|失败| wd{"budget.withdraw()"}
  wd -->|false| none2[返回 None，不重试]
  wd -->|true| nb["backoff.next_backoff()"]
  nb --> some["返回 Some(Sleep)"]
  some --> waiting[进入 Waiting]

Retry 不会替你调用这两个模块。上图是 budget 模块文档示例的做法（成功时存、失败时取）再加上退避；存取的时机、退避的节奏，全由你的 Policy 决定。

ExponentialBackoff

两个 trait

pub trait MakeBackoff {
    /// The backoff type produced by this maker.
    type Backoff: Backoff;

    /// Constructs a new backoff type.
    fn make_backoff(&mut self) -> Self::Backoff;
}

/// A backoff trait where a single mutable reference represents a single
/// backoff session. Implementors must also implement [`Clone`] which will
/// reset the backoff back to the default state for the next session.
pub trait Backoff {
    /// The future associated with each backoff. This usually will be some sort
    /// of timer.
    type Future: Future<Output = ()>;

    /// Initiate the next backoff in the sequence.
    fn next_backoff(&mut self) -> Self::Future;
}

MakeBackoff 造出一次「退避会话」，Backoff 每调一次 next_backoff 给出下一段等待。Backoff::Future 的约束和 Policy::Future 一样是 Future<Output = ()>，所以 next_backoff() 的返回值可以直接作为 Policy::retry 的返回值。

这一套是 0.5.0 加的（#685），PR 说 ExponentialBackoff 的实现取自 linkerd2-proxy，是「the initial building blocks for a more fully batteries included retry policy」。到 0.5.3 为止，那个「开箱即用的策略」还没有出现。

参数校验

ExponentialBackoffMaker::new 返回 Result：

    pub fn new(
        min: time::Duration,
        max: time::Duration,
        jitter: f64,
        rng: R,
    ) -> Result<Self, InvalidBackoff> {
        if min > max {
            return Err(InvalidBackoff("maximum must not be less than minimum"));
        }
        if max == time::Duration::from_millis(0) {
            return Err(InvalidBackoff("maximum must be non-zero"));
        }
        if jitter < 0.0 {
            return Err(InvalidBackoff("jitter must not be negative"));
        }
        if jitter > 100.0 {
            return Err(InvalidBackoff("jitter must not be greater than 100"));
        }
        if jitter.is_nan() {
            return Err(InvalidBackoff("jitter must not be NaN"));
        }
        // ...

InvalidBackoff 的 Display 在原因前加 invalid backoff: 。正负无穷被前两条范围检查拦下，所以最后只需要单独查 NaN（NaN 和任何数比较都是 false，会穿过前面两条）。这条 NaN 检查是 CHANGELOG 里 0.5.3 唯一一条 retry 条目（#843）：之前写的是 !jitter.is_finite()，错误信息是 jitter must be finite，PR 的说法是「Infinities are already covered by the bounds checks on jitter, so checking for finiteness should be redundant」。

min 可以是 0，校验不拦，但 0 乘多少都是 0，抖动也按 0 算，于是每次退避都是 0（见下面实验的第 6 组）。

Default 是 min 50ms、max 为 u64::MAX 毫秒（相当于不封顶）、jitter 0.99，随机数用 HasherRng::default()。

每次等多久

base、jitter、next_backoff：

    fn base(&self) -> time::Duration {
        // ...
        self.min
            .checked_mul(2_u32.saturating_pow(self.iterations))
            .unwrap_or(self.max)
            .min(self.max)
    }

    /// Returns a random, uniform duration on `[0, base*self.jitter]` no greater
    /// than `self.max`.
    fn jitter(&mut self, base: time::Duration) -> time::Duration {
        if self.jitter == 0.0 {
            time::Duration::default()
        } else {
            let jitter_factor = self.rng.next_f64();
            // ...
            let rand_jitter = jitter_factor * self.jitter;
            let secs = (base.as_secs() as f64) * rand_jitter;
            let nanos = (base.subsec_nanos() as f64) * rand_jitter;
            let remaining = self.max - base;
            time::Duration::new(secs as u64, nanos as u32).min(remaining)
        }
    }

    fn next_backoff(&mut self) -> Self::Future {
        let base = self.base();
        let next = base + self.jitter(base);

        self.iterations += 1;

        tokio::time::sleep(next)
    }

第 n 次（从 0 数）的基数是 min × 2ⁿ，溢出就取 max，再和 max 取小。指数用 saturating_pow，乘法用 checked_mul，退避次数再多也不会 panic。
抖动是一个 0 到 1 之间（不含 1）的随机数 × jitter × 基数，再截到 max - base，所以基数加抖动不会超过 max。jitter 是相对基数的比例：0.99 表示最多再加 99%。
秒和纳秒两部分分开乘，纳秒部分用 as u32 转回整数。Rust 的浮点转整数是饱和的，jitter 大于约 4.29 时纳秒部分可能超过 u32::MAX，被截断（第 5 组）。
next_backoff 返回 tokio::time::Sleep，所以用到 ExponentialBackoff 就需要 tokio 的计时器，retry feature 因此依赖 tokio/time。

随机数来自 HasherRng：一个 RandomState 加一个计数器，每次 next_u64 把计数器哈希一遍，计数器加一。它 derive(Clone)，复制时计数器也一起复制。而 make_backoff 每次都是 self.rng.clone()，maker 自己的 rng 从来不用。

下面的程序在暂停的 tokio 时钟下运行，用 Sleep::deadline() - Instant::now() 读出每次退避的时长。Max 是一个总返回 u64::MAX 的 Rng，让 next_f64() 恒为略小于 1 的值，抖动取到上限：

tower = { version = "=0.5.3", features = ["retry"] }
tokio = { version = "=1.54.1", features = ["macros", "rt", "time", "test-util"] }

use std::time::Duration;
use tokio::time::Instant;
use tower::{
  retry::backoff::{Backoff, ExponentialBackoffMaker, MakeBackoff},
  util::rng::{HasherRng, Rng},
};

// Always returns the largest value, so next_f64() is just below 1.0
#[derive(Clone)]
struct Max;

impl Rng for Max {
  fn next_u64(&mut self) -> u64 {
    u64::MAX
  }
}

fn delays<B: Backoff<Future = tokio::time::Sleep>>(b: &mut B, n: usize) -> Vec<Duration> {
  (0..n).map(|_| b.next_backoff().deadline() - Instant::now()).collect()
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  let ms = Duration::from_millis;

  println!("1. jitter 0.0, min 100ms, max 1s");
  let mut maker = ExponentialBackoffMaker::new(ms(100), ms(1000), 0.0, HasherRng::default()).unwrap();
  println!("   {:?}", delays(&mut maker.make_backoff(), 6));

  println!("2. jitter 1.0, rng always ~1.0");
  let mut maker = ExponentialBackoffMaker::new(ms(100), ms(1000), 1.0, Max).unwrap();
  println!("   {:?}", delays(&mut maker.make_backoff(), 6));

  println!("3. two backoffs from one maker, HasherRng");
  let mut maker = ExponentialBackoffMaker::new(ms(100), ms(10_000), 1.0, HasherRng::default()).unwrap();
  let mut a = maker.make_backoff();
  let mut b = maker.make_backoff();
  println!("   a {:?}", delays(&mut a, 3));
  println!("   b {:?}", delays(&mut b, 3));

  println!("4. clone after two backoffs");
  let mut c = maker.make_backoff();
  delays(&mut c, 2);
  let mut d = c.clone();
  println!("   c {:?}", delays(&mut c, 1));
  println!("   d {:?}", delays(&mut d, 1));

  println!("5. jitter 10.0, base 1.9s, rng always ~1.0");
  let mut maker = ExponentialBackoffMaker::new(ms(1900), ms(60_000), 10.0, Max).unwrap();
  println!("   {:?}", delays(&mut maker.make_backoff(), 1));

  println!("6. min 0, max 1s, jitter 1.0");
  let mut maker = ExponentialBackoffMaker::new(ms(0), ms(1000), 1.0, HasherRng::default()).unwrap();
  println!("   {:?}", delays(&mut maker.make_backoff(), 4));

  println!("7. validation");
  let cases = [
    ("min > max", ms(2), ms(1), 0.5),
    ("max == 0", ms(0), ms(0), 0.5),
    ("jitter < 0", ms(1), ms(2), -0.1),
    ("jitter > 100", ms(1), ms(2), 100.5),
    ("jitter inf", ms(1), ms(2), f64::INFINITY),
    ("jitter NaN", ms(1), ms(2), f64::NAN),
  ];
  for (name, min, max, jitter) in cases {
    match ExponentialBackoffMaker::new(min, max, jitter, HasherRng::default()) {
      Ok(_) => println!("   {name:<18} Ok"),
      Err(e) => println!("   {name:<18} Err({e})"),
    }
  }
}

M2 Max、rustc 1.99.0、--release，其中一次的输出：

1. jitter 0.0, min 100ms, max 1s
   [100ms, 200ms, 400ms, 800ms, 1s, 1s]
2. jitter 1.0, rng always ~1.0
   [199.999999ms, 399.999999ms, 799.999999ms, 1s, 1s, 1s]
3. two backoffs from one maker, HasherRng
   a [147.230554ms, 397.855367ms, 770.772776ms]
   b [147.230554ms, 397.855367ms, 770.772776ms]
4. clone after two backoffs
   c [770.772776ms]
   d [770.772776ms]
5. jitter 10.0, base 1.9s, rng always ~1.0
   [15.194967295s]
6. min 0, max 1s, jitter 1.0
   [0ns, 0ns, 0ns, 0ns]
7. validation
   min > max          Err(invalid backoff: maximum must not be less than minimum)
   max == 0           Err(invalid backoff: maximum must be non-zero)
   jitter < 0         Err(invalid backoff: jitter must not be negative)
   jitter > 100       Err(invalid backoff: jitter must not be greater than 100)
   jitter inf         Err(invalid backoff: jitter must not be greater than 100)
   jitter NaN         Err(invalid backoff: jitter must not be NaN)

连跑 5 次，第 3、4 组的具体数值每次都不同（RandomState 每次运行的种子不同），但「a 和 b 完全相同」「c 和 d 完全相同」「c 等于 a 的第三个值」每次都成立；其余各组每次相同。

第 1 组：100、200、400、800，然后封顶在 1s。
第 2 组：抖动取到上限，每次几乎翻倍，但第 4 次起被 max - base 截住，总和正好 1s。
第 3 组：同一个 maker 造出的退避，抖动序列完全相同。 原因就是上面说的 self.rng.clone()：每个 ExponentialBackoff 都从 maker 的那个计数器起点开始。抖动的本意是让同时失败的请求错开重试时间（ExponentialBackoff 的文档说「in order to avoid retry spikes」），但同一个进程里从同一个 maker 出来的请求，第 n 次重试会等完全一样长。不同进程的 RandomState 种子不同，彼此之间仍然是错开的。
第 4 组：clone 不会重置。 Backoff 的文档说实现者的 Clone「will reset the backoff back to the default state」，但 ExponentialBackoff 是 derive(Clone)，iterations 和随机数计数器原样复制，d 接着 c 的进度走。要从头开始，得用 make_backoff 造新的。
第 5 组：基数 1.9s，抖动比例接近 10，按公式应该是 1.9s + 约 19s。实际秒部分 1 × 9.99… 截成 9，纳秒部分 9×10⁸ × 9.99… 约 9×10⁹，超过 u32::MAX，饱和成 4294967295ns，Duration::new 把它进位成 4.29…s。结果 1.9 + 9 + 4.294967295 = 15.194967295s。jitter 不超过 1 的常见配置碰不到这个问题。
第 7 组：正无穷报的是 greater than 100，不是 NaN 那条。

Budget：重试是放大器

只按「最多重试 N 次」来配置重试，budget 的模块文档指出了两个问题：N 是拍脑袋定的；更要命的是重试风暴。下游故障率一上升，客户端开始重试，重试又加重了下游的负载，失败更多，重试更多。文档的原话：「If each client is configured to retry up to 3 times, this can quadruple the number of requests being sent!」如果客户端的客户端也配了重试，倍数还要相乘。

预算的思路是让重试的总量跟着成功的总量走：成功的请求往预算里存一笔，每次重试从预算里取一笔，取不出来就不重试。下游大面积失败时没有新的存款，重试自然停下来。

Budget trait 只有两个方法：

pub trait Budget {
    /// Store a "deposit" in the budget, which will be used to permit future
    /// withdrawals.
    fn deposit(&self);

    /// Check whether there is enough "balance" in the budget to issue a new
    /// retry.
    ///
    /// If there is not enough, false is returned.
    fn withdraw(&self) -> bool;
}

两个方法都是 &self：预算要被所有请求共享，而策略每个请求一份副本，所以预算通常放在 Arc 里，副本之间共享同一个。什么时候 deposit、什么时候 withdraw，trait 不管。模块文档的示例是：retry 收到 Ok 就 deposit 并返回 None，收到 Err 就 withdraw，取到了才重试。

📝NOTE

冷知识：Budget 以前是一个结构体

0.5.0 之前，tower::retry::budget::Budget 是具体的结构体，withdraw 返回 Result<(), Overdrawn>，换不了别的算法。#703（0.5.0）把 Budget 改成 trait，原来的结构体改名叫 TpsBudget，withdraw 改成返回 bool，Overdrawn 类型删掉了。CHANGELOG 的说法：「Budget is now a trait. This allows end-users to implement their own budget and bucket implementations.」PR 描述里还提到一个按重试次数递增成本的 SimpleBucket，0.5.3 里并没有这个类型，TpsBudget 是唯一的实现。

TpsBudget：十个窗口的滑动计数

TpsBudget：

pub struct TpsBudget {
    generation: Mutex<Generation>,
    /// Initial budget allowed for every second.
    reserve: isize,
    /// Slots of a the TTL divided evenly.
    slots: Box<[AtomicIsize]>,
    /// The amount of time represented by each slot.
    window: Duration,
    /// The changers for the current slot to be committed
    /// after the slot expires.
    writer: AtomicIsize,
    /// Amount of tokens to deposit for each put().
    deposit_amount: isize,
    /// Amount of tokens to withdraw for each try_get().
    withdraw_amount: isize,
}

文档叫它「Transactions Per Minute config」，又说它「uses a token bucket」。从代码看，参数的单位是秒，结构是把 ttl 均分成 10 个槽的滑动窗口计数器：

writer：当前窗口里存取的净额，存款加、取款减。
slots：过去 10 个窗口各自的净额。
reserve：一笔固定的底数，永远算在余额里。
generation：当前窗口的下标和开始时间，只有它用 Mutex，计数都是原子操作。

new：把比例换成整数

new：

    pub fn new(ttl: Duration, min_per_sec: u32, retry_percent: f32) -> Self {
        // assertions taken from finagle
        assert!(ttl >= Duration::from_secs(1));
        assert!(ttl <= Duration::from_secs(60));
        assert!(retry_percent >= 0.0);
        assert!(retry_percent <= 1000.0);
        assert!(min_per_sec < i32::MAX as u32);

        let (deposit_amount, withdraw_amount) = if retry_percent == 0.0 {
            // If there is no percent, then you gain nothing from deposits.
            // Withdrawals can only be made against the reserve, over time.
            (0, 1)
        } else if retry_percent <= 1.0 {
            (1, (1.0 / retry_percent) as isize)
        } else {
            // Support for when retry_percent is between 1.0 and 1000.0,
            // meaning for every deposit D, D * retry_percent withdrawals
            // can be made.
            (1000, (1000.0 / retry_percent) as isize)
        };
        let reserve = (min_per_sec as isize)
            .saturating_mul(ttl.as_secs() as isize) // ttl is between 1 and 60 seconds
            .saturating_mul(withdraw_amount);

        // AtomicIsize isn't clone, so the slots need to be built in a loop...
        let windows = 10u32;
        // ...
            window: ttl / windows,
        // ...
    }

参数不合法直接 assert! panic，不像 ExponentialBackoffMaker::new 返回 Result。ttl 只能在 1 到 60 秒之间，注释说这些限制来自 Finagle。
比例换成「存一次加几、取一次减几」两个整数。retry_percent 为 0.2 时存 1 取 5，五次成功换一次重试。除法结果用 as isize 截断，所以 1 / retry_percent 不是整数时，实际能重试的比例比配置的高。
reserve = min_per_sec × ttl秒数 × withdraw_amount，正好够 min_per_sec × ttl 次取款。它照顾刚启动、还没攒下成功记录的客户端。
窗口数写死为 10，每个窗口 ttl / 10。

Default 是 TpsBudget::new(Duration::from_secs(10), 10, 0.2)：存 1 取 5，reserve = 10 × 10 × 5 = 500，窗口 1 秒。

📝NOTE

没有蠢问题

问： min_per_sec 是 10，是不是说每秒至少（或者最多）能重试 10 次？

答：都不是。reserve 不是每秒补充一次的令牌，而是一个常数，一直算在余额里；真正随时间变化的是取款记录，它在 writer 和 slots 里待满大约一个 ttl 后才被清掉。所以默认配置下，一个请求都没成功过的客户端可以一口气重试 100 次（min_per_sec × ttl），然后要等这些取款记录过期，大约 10 秒后才能再重试。下面实验的第 1、2 组就是这个情形。

存、取、过期

put 和 try_get 都先调 expire，再改 writer：

    fn put(&self, amt: isize) {
        self.expire();
        self.writer.fetch_add(amt, Ordering::SeqCst);
    }

    fn try_get(&self, amt: isize) -> bool {
        debug_assert!(amt >= 0);

        self.expire();

        let sum = self.sum();
        if sum >= amt {
            self.writer.fetch_add(-amt, Ordering::SeqCst);
            true
        } else {
            false
        }
    }

sum 是 writer + 所有槽 + reserve，都用饱和加法。try_get 先算余额、再扣，两步之间没有锁，并发的取款可能同时看到同一个余额。

expire 负责让窗口往前走：

    fn expire(&self) {
        let mut gen = self.generation.lock().expect("generation lock");

        let now = Instant::now();
        let diff = now.saturating_duration_since(gen.time);
        if diff < self.window {
            // not expired yet
            return;
        }

        let to_commit = self.writer.swap(0, Ordering::SeqCst);
        self.slots[gen.index].store(to_commit, Ordering::SeqCst);

        let mut diff = diff;
        let mut idx = (gen.index + 1) % self.slots.len();
        while diff > self.window {
            self.slots[idx].store(0, Ordering::SeqCst);
            diff -= self.window;
            idx = (idx + 1) % self.slots.len();
        }

        gen.index = idx;
        gen.time = now;
    }

距上次推进不到一个窗口，什么都不做。
否则把 writer 的净额写进当前下标的槽，writer 清零。槽是覆盖写入的，上一轮（约 10 个窗口前）留在这个槽里的数就此作废，这就是「过期」。
中间整段跳过的窗口，对应的槽清零。条件是严格的 diff > self.window。
新的下标指向的槽不清零，它的旧值（约 10 个窗口前的）要等下一次推进时才被覆盖，在那之前仍然算在余额里。
gen.time = now，窗口从这一刻重新算，不和整秒对齐。

now 来自 tokio::time::Instant，所以可以用 tokio 的暂停时钟精确地推时间：

tower = { version = "=0.5.3", features = ["retry"] }
tokio = { version = "=1.54.1", features = ["macros", "rt", "time", "test-util"] }

use std::time::Duration;
use tokio::time::{advance, Instant};
use tower::retry::budget::{Budget, TpsBudget};

fn drain(b: &TpsBudget) -> u32 {
  let mut n = 0;
  while b.withdraw() {
    n += 1;
  }
  n
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  println!("1. default");
  let b = TpsBudget::default();
  println!("   {b:?}");
  println!("   withdrawals with no deposits: {}", drain(&b));
  for _ in 0..100 {
    b.deposit();
  }
  println!("   after 100 deposits: {}", drain(&b));

  println!("2. when does the reserve come back? (one withdraw attempt every 500ms)");
  let start = Instant::now();
  loop {
    advance(Duration::from_millis(500)).await;
    if b.withdraw() {
      println!("   first success at +{:?}", Instant::now() - start);
      break;
    }
  }

  println!("3. one jump after exhausting");
  for jump in [10_000, 10_001] {
    let b = TpsBudget::default();
    drain(&b);
    advance(Duration::from_millis(jump)).await;
    println!("   after {jump}ms: {} withdrawals", drain(&b));
  }
}

M2 Max、rustc 1.99.0、--release，连跑 3 次输出相同：

1. default
   Budget { deposit: 1, withdraw: 5, balance: 500 }
   withdrawals with no deposits: 100
   after 100 deposits: 20
2. when does the reserve come back? (one withdraw attempt every 500ms)
   first success at +11s
3. one jump after exhausting
   after 10000ms: 0 withdrawals
   after 10001ms: 100 withdrawals

第 1 组：Debug 里的 balance 就是 sum()，起始 500 全是 reserve。没有存款时能取 100 次（500 / 5）。之后 100 次成功存进 100，正好再换 20 次重试，这就是 20% 的比例。
第 2 组：每 500ms 试一次，11 秒时才第一次成功，比 ttl 多一个窗口。时间是暂停的，推进全靠 withdraw 里的 expire：每隔一次调用推进一个窗口，第 10 秒时下标转回 0，0 号槽里还是 0 秒时那 −500，没被清掉（新下标的槽不清零）；到第 11 秒的推进才把它覆盖成 0。
第 3 组：一次跳 10000ms，循环里 diff > window 在 diff 减到正好 1 秒时停下，下标又落回 0 号槽，旧的 −500 还在，一次都取不出；多 1ms，0 号槽也被清零，reserve 全部回来。

所以一笔存款或取款的有效期大约是 ttl，视调用的时机，最多再多一个窗口。

❗IMPORTANT

动脑筋：retry_percent 设成 0.6，存 10 次之后能取几次？

下面用 ttl 1 秒、min_per_sec 0（没有 reserve），对几个比例各存 10 次，再取到取不出为止。0.6 那一行的取款次数是 6 吗？

tower = { version = "=0.5.3", features = ["retry"] }
tokio = { version = "=1.54.1", features = ["macros", "rt", "time", "test-util"] }

use std::time::Duration;
use tower::retry::budget::{Budget, TpsBudget};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  // ttl 1s, min_per_sec 0, then 10 deposits
  for pct in [0.2_f32, 0.3, 0.4, 0.6, 2.0] {
    let b = TpsBudget::new(Duration::from_secs(1), 0, pct);
    for _ in 0..10 {
      b.deposit();
    }
    let mut n = 0;
    while b.withdraw() {
      n += 1;
    }
    println!("{pct}: {b:?} -> {n} withdrawals");
  }
}

动脑筋答案