Rust in Everything
基于 tower 0.5.3

Policy 与重试状态机

主线位置：主要在请求主线上。Retry::poll_ready 原样转发给内层；重试全部发生在响应 future 里，而且 future 每次重试前要自己去问内层 poll_ready。就绪主线在这里被搬进了请求主线。

stateDiagram-v2
  [*] --> Called: 调用 call
  Called --> [*]: 不重试，返回结果
  Called --> Waiting: retry 返回 Some
  Waiting --> Retrying: 等待完成
  Retrying --> Retrying: 内层 Pending
  Retrying --> Called: 就绪后再 call
  Retrying --> [*]: poll_ready 报错

Retry 自己不知道什么算失败、要等多久、最多重试几次。这些都交给一个 Policy，Retry 只负责一件事：把「发请求 → 看结果 → 等一会 → 再发」这个循环做成一个 future。

Policy：两个方法

Policy（文档注释已删）：

pub trait Policy<Req, Res, E> {
    type Future: Future<Output = ()>;

    fn retry(&mut self, req: &mut Req, result: &mut Result<Res, E>) -> Option<Self::Future>;

    fn clone_request(&mut self, req: &Req) -> Option<Req>;
}
  • retry 在每次内层返回结果后被调用（前提是这次留了请求副本，见下一条），成功和失败都会调用，由策略自己判断哪种结果算「失败」。返回 None 表示不重试，这个结果直接交给调用方；返回 Some(fut) 表示要重试，Retry 先等 fut 完成再重发。fut 的输出是 ()，它只表示「等多久」：可以是一个 Sleep（退避），也可以是立即完成的 future::ready(())。
  • req 和 result 都是 &mut。改 req，下一次重试发出去的就是改过的请求（比如在请求里记一个重试次数）；改 result，可以把失败改成成功，或者在放弃时换成一个「重试用尽」的错误。
  • clone_request 在每次发请求之前被调用，留一份请求的副本给下次重试用。请求的所有权要交给内层的 call，发出去就拿不回来，所以只能事先复制。返回 None 表示这个请求复制不了（比如 body 是一个只能读一次的流），这次发出之后 retry 根本不会被调用，结果原样返回。

req 和 result 改成 &mut 是 0.5.0 的破坏性改动（#584），PR 给出的两个用途正是上面说的：在请求里标记「这是重试」，以及在重试用尽时改写最终的响应。

policy.rs 末尾有一行只在测试里编译的 _obj_safe，用 Box<dyn Policy<...>> 当参数类型，保证这个 trait 一直是 object safe 的。

📝NOTE
冷知识：retry 返回的 future 以前要「生出」下一个策略

0.5.0 之前，Policy 的方法都是 &self，retry 返回的 future 输出是 Self：要重试时，策略得造一个新的自己（比如 Attempts(self.0 - 1)），future 完成后 Retry 用它替换掉旧的。响应 future 的第二个状态也不叫 Waiting，叫 Checking，完成时把新策略写回 policy 字段。

#681（0.5.0）把方法改成 &mut self，future 输出改成 ()。PR 的理由：「The motivation for this change is to simplify the trait a bit. By the trait methods having mutable references it means that for each retry request session you can mutate the local policy. This is because the policy is cloned for each individual request that arrives into the retry middleware. In addition, this allows the Policy trait to be object safe.」旧的 trait 声明是 pub trait Policy<Req, Res, E>: Sized（future 要输出 Self），带着 Sized 就不能做成 dyn；这次改动去掉了 Sized，同时加上了上面那行 _obj_safe 检查。状态名也在这次改成了 Waiting。

Retry::call：先发请求，再复制自己

Retry 只有两个字段，#[derive(Clone)]：

    #[derive(Clone, Debug)]
    pub struct Retry<P, S> {
        policy: P,
        service: S,
    }

poll_ready 和 call：

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        // NOTE: the Future::poll impl for ResponseFuture assumes that Retry::poll_ready is
        // equivalent to Ready.service.poll_ready. If this ever changes, that code must be updated
        // as well.
        self.service.poll_ready(cx)
    }

    fn call(&mut self, request: Request) -> Self::Future {
        let cloned = self.policy.clone_request(&request);
        let future = self.service.call(request);

        ResponseFuture::new(cloned, self.clone(), future)
    }

call 做三件事，顺序有讲究：

  1. 先 clone_request，因为下一步 request 就被移走了。
  2. 用 self.service 发第一次请求。这个服务刚被调用方 poll_ready 过，第一次发送走的是正常的就绪主线。
  3. self.clone() 复制整个 Retry，策略和内层服务一起，放进响应 future。之后的重试都用这份副本，不再碰 self。

所以 Retry 的 Service 实现要求 S: Clone 和 P: Clone（第 72 行）。Retry 的文档（第 22 行起）说得直白：内层不能 Clone 就在下面套一层 Buffer；内层不能 Clone 时的编译错误见 Clone 不满足。

复制发生在 call 之后。像 ConcurrencyLimit 这样把就绪状态存在自己身上的服务，副本不带许可；这没关系，因为副本在重试前一定会被重新 poll_ready（见下一节）。

代价是每个请求都复制一次，不管最后有没有重试，也不管 clone_request 有没有返回 None。下面的程序让内层每次都失败，clone 时打印一行：

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

use std::{
  future::{ready, Ready},
  task::{Context, Poll},
};
use tower::{retry::{Policy, Retry}, Service, ServiceExt};

struct Inner;

impl Clone for Inner {
  fn clone(&self) -> Self {
    println!("  Inner::clone");
    Inner
  }
}

impl Service<u32> for Inner {
  type Response = u32;
  type Error = String;
  type Future = Ready<Result<u32, String>>;

  fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), String>> {
    Poll::Ready(Ok(()))
  }

  fn call(&mut self, req: u32) -> Self::Future {
    println!("  Inner::call({req})");
    ready(Err(format!("failed {req}")))
  }
}

#[derive(Clone)]
struct Once {
  can_clone: bool,
  used: bool,
}

impl Policy<u32, u32, String> for Once {
  type Future = Ready<()>;

  fn retry(&mut self, _: &mut u32, _: &mut Result<u32, String>) -> Option<Ready<()>> {
    println!("  Policy::retry used={}", self.used);
    if self.used {
      return None;
    }
    self.used = true;
    Some(ready(()))
  }

  fn clone_request(&mut self, req: &u32) -> Option<u32> {
    println!("  Policy::clone_request -> {}", if self.can_clone { "Some" } else { "None" });
    self.can_clone.then_some(*req)
  }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  for can_clone in [true, false] {
    println!("can_clone = {can_clone}");
    let mut svc = Retry::new(Once { can_clone, used: false }, Inner);
    let svc = svc.ready().await.unwrap();
    let fut = svc.call(1);
    println!("  -- call returned, awaiting");
    println!("  result: {:?}", fut.await);
  }
}

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

can_clone = true
  Policy::clone_request -> Some
  Inner::call(1)
  Inner::clone
  -- call returned, awaiting
  Policy::retry used=false
  Policy::clone_request -> Some
  Inner::call(1)
  Policy::retry used=true
  result: Err("failed 1")
can_clone = false
  Policy::clone_request -> None
  Inner::call(1)
  Inner::clone
  -- call returned, awaiting
  result: Err("failed 1")
  • 两轮都是 clone_request → Inner::call → Inner::clone，和源码的顺序一致；can_clone = false 时照样复制了一次。
  • 重试那次没有新的 Inner::clone：用的是 future 里那份副本。重试前又调了一次 clone_request，给可能的下一次重试留副本。
  • clone_request 返回 None 时，retry 一次都没被调用。

内层服务的 clone 越贵（比如一整摞中间件，每层都有 Arc），这笔固定开销越明显。

策略：每个请求一份副本

self.clone() 里也包括策略，所以每个请求（连同它的所有重试，文档叫 request session）用的是自己那份策略。在 retry 里改字段（比如剩余次数减一）只影响这一个请求；要跨请求共享状态，文档建议放进 Arc，下一页的 TpsBudget 就是这么用的。

但第一次 clone_request 是在 self.policy 上调的，发生在 self.clone() 之前。下面的策略在两个方法里各数一个数，对同一个 Retry 连发 3 个请求，每个请求重试一次：

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

use std::{
  convert::Infallible,
  future::{ready, Ready},
};
use tower::{retry::{Policy, Retry}, service_fn, Service, ServiceExt};

#[derive(Clone)]
struct Counting {
  clones: u32,  // bumped in clone_request
  retries: u32, // bumped in retry
}

impl Policy<u32, u32, Infallible> for Counting {
  type Future = Ready<()>;

  fn retry(&mut self, req: &mut u32, _: &mut Result<u32, Infallible>) -> Option<Ready<()>> {
    self.retries += 1;
    println!("  req {req}: retry         clones={} retries={}", self.clones, self.retries);
    (self.retries < 2).then(|| ready(()))
  }

  fn clone_request(&mut self, req: &u32) -> Option<u32> {
    self.clones += 1;
    println!("  req {req}: clone_request clones={}", self.clones);
    Some(*req)
  }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let inner = service_fn(|x: u32| async move { Ok::<_, Infallible>(x) });
  let mut svc = Retry::new(Counting { clones: 0, retries: 0 }, inner);
  for req in 1..=3 {
    svc.ready().await.unwrap().call(req).await.unwrap();
  }
}

（这个策略连成功也重试一次，只是为了让 retry 被调用两次。）

❗IMPORTANT
动脑筋：第三个请求第一次调用 clone_request 时，打印的 clones 是几？

retries 每个请求都从 0 开始，这一点上面已经说了。clones 呢？是 1（和 retries 一样每个请求重新开始），3，还是 5（每个请求调两次，前两个请求一共调了 4 次）？完整输出在文末。

ResponseFuture：三个状态

ResponseFuture 和它的 State：

    pub struct ResponseFuture<P, S, Request>
    where
        P: Policy<Request, S::Response, S::Error>,
        S: Service<Request>,
    {
        request: Option<Request>,
        #[pin]
        retry: Retry<P, S>,
        #[pin]
        state: State<S::Future, P::Future>,
    }

    enum State<F, P> {
        // Polling the future from [`Service::call`]
        Called {
            #[pin]
            future: F
        },
        // Polling the future from [`Policy::retry`]
        Waiting {
            #[pin]
            waiting: P
        },
        // Polling [`Service::poll_ready`] after [`Waiting`] was OK.
        Retrying,
    }
  • request 是 clone_request 留下的副本，None 表示这个请求不能重试。
  • retry 是 call 里复制出来的那份 Retry，重试时用它的 service 和 policy。
  • state 里同一时刻只有一个 future：内层的响应 future，或者策略给的等待 future。

poll 是一个 loop，每个状态做完就切到下一个，直到某处返回 Pending 或最终结果：

        loop {
            match this.state.as_mut().project() {
                StateProj::Called { future } => {
                    let mut result = ready!(future.poll(cx));
                    if let Some(req) = &mut this.request {
                        match this.retry.policy.retry(req, &mut result) {
                            Some(waiting) => {
                                this.state.set(State::Waiting { waiting });
                            }
                            None => return Poll::Ready(result),
                        }
                    } else {
                        // request wasn't cloned, so no way to retry it
                        return Poll::Ready(result);
                    }
                }
                StateProj::Waiting { waiting } => {
                    ready!(waiting.poll(cx));

                    this.state.set(State::Retrying);
                }
                StateProj::Retrying => {
                    // NOTE: we assume here that
                    // ...
                    ready!(this.retry.as_mut().project().service.poll_ready(cx))?;
                    let req = this
                        .request
                        .take()
                        .expect("retrying requires cloned request");
                    *this.request = this.retry.policy.clone_request(&req);
                    this.state.set(State::Called {
                        future: this.retry.as_mut().project().service.call(req),
                    });
                }
            }
        }

逐个状态看：

  • Called：等内层的响应。拿到结果后，有副本就问策略。策略说重试，state.set(Waiting) 把旧的响应 future 换掉，它和这次的 result 一起被丢弃；策略说不重试，结果原样返回。传给 retry 的 req 是副本，策略改的就是下次要发的那个。
  • Waiting：等策略给的 future。这里只关心它完成，不关心输出。
  • Retrying：对副本服务调 poll_ready。Pending 就整个 future Pending，内层就绪时唤醒；报错就用 ? 直接返回这个错误，不问策略；就绪了就取出请求副本，先再复制一份留给下一次，然后 call，回到 Called。
  • expect("retrying requires cloned request") 不会触发：只有 request 是 Some 时才会进入 Waiting，Retrying 是从 Waiting 来的。

Retrying 前面那段 NOTE 注释（上面省略了）说明这里调的是 service.poll_ready 而不是 Retry::poll_ready，理由是「to avoid adding an Unpin bound to the Policy」。这样做的前提是 Retry::poll_ready 一直只是原样转发，所以 mod.rs 的 poll_ready 里有一段对应的 NOTE，提醒以后改那边时要连这里一起改。

下面用 tower-test 的 mock 当内层服务，一步一步推着这个 future 走。测试代码扮演内层：allow(n) 控制它的 poll_ready 放行几次，next_request 取出请求，send_error / send_response 回答。策略最多重试一次，每次重试把请求里的 attempt 加一；等待 future 是一个 Notify，什么时候放行由测试决定。

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

use std::{future::Future, pin::Pin, sync::Arc};
use tokio::sync::Notify;
use tokio_test::task;
use tower::retry::{Policy, Retry};
use tower_test::mock;

type BoxError = Box<dyn std::error::Error + Send + Sync>;

#[derive(Clone, Debug, PartialEq)]
struct Req {
  body: &'static str,
  attempt: u32,
}

#[derive(Clone)]
struct Attempts {
  left: u32,
  gate: Arc<Notify>, // the test decides when the wait is over
}

impl Policy<Req, &'static str, BoxError> for Attempts {
  type Future = Pin<Box<dyn Future<Output = ()>>>;

  fn retry(
    &mut self,
    req: &mut Req,
    result: &mut Result<&'static str, BoxError>,
  ) -> Option<Self::Future> {
    let verdict = match result {
      Err(_) if self.left > 0 => "retry",
      _ => "stop",
    };
    println!("  policy.retry(attempt={}, {:?}) left={} -> {verdict}",
      req.attempt, result.as_ref().map_err(|e| e.to_string()), self.left);
    if verdict == "stop" {
      return None;
    }
    self.left -= 1;
    req.attempt += 1; // the next attempt sends the mutated request
    let gate = self.gate.clone();
    Some(Box::pin(async move { gate.notified().await }))
  }

  fn clone_request(&mut self, req: &Req) -> Option<Req> {
    println!("  policy.clone_request(attempt={})", req.attempt);
    Some(req.clone())
  }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let gate = Arc::new(Notify::new());
  let policy = Attempts { left: 1, gate: gate.clone() };
  let (mut svc, mut handle) = mock::spawn_with(|m| Retry::new(policy.clone(), m));

  println!("A. error, wait, retry");
  handle.allow(1);
  println!("svc.poll_ready -> {:?}", svc.poll_ready().map_err(|e| e.to_string()));
  let mut fut = task::spawn(svc.call(Req { body: "hello", attempt: 0 }));
  let (req, send) = handle.next_request().await.unwrap();
  println!("inner got {req:?}");
  send.send_error("boom");
  println!("fut.poll -> {:?}  (Called -> Waiting)", fut.poll().map(|_| ()));
  gate.notify_one();
  println!("gate opened, fut woken: {}", fut.is_woken());
  println!("fut.poll -> {:?}  (Waiting -> Retrying, inner not ready)", fut.poll().map(|_| ()));
  println!("inner has a request? {}", handle.poll_request().is_ready());
  handle.allow(1);
  println!("allow(1), fut woken: {}", fut.is_woken());
  println!("fut.poll -> {:?}  (Retrying -> Called)", fut.poll().map(|_| ()));
  let (req, send) = handle.next_request().await.unwrap();
  println!("inner got {req:?}");
  send.send_response("world");
  println!("fut.poll -> {:?}", fut.poll().map(|r| r.map_err(|e| e.to_string())));

  println!("B. a second request gets a fresh policy clone");
  handle.allow(1);
  println!("svc.poll_ready -> {:?}", svc.poll_ready().map_err(|e| e.to_string()));
  let mut fut = task::spawn(svc.call(Req { body: "again", attempt: 0 }));
  let (_, send) = handle.next_request().await.unwrap();
  send.send_error("boom");
  println!("fut.poll -> {:?}", fut.poll().map(|_| ()));
  gate.notify_one();
  handle.allow(1);
  println!("fut.poll -> {:?}", fut.poll().map(|_| ()));
  let (_, send) = handle.next_request().await.unwrap();
  send.send_error("boom again");
  println!("fut.poll -> {:?}", fut.poll().map(|r| r.map_err(|e| e.to_string())));

  println!("C. inner poll_ready fails while Retrying");
  handle.allow(1);
  println!("svc.poll_ready -> {:?}", svc.poll_ready().map_err(|e| e.to_string()));
  let mut fut = task::spawn(svc.call(Req { body: "third", attempt: 0 }));
  let (_, send) = handle.next_request().await.unwrap();
  send.send_error("boom");
  println!("fut.poll -> {:?}", fut.poll().map(|_| ()));
  handle.send_error("inner broken");
  gate.notify_one();
  println!("fut.poll -> {:?}", fut.poll().map(|r| r.map_err(|e| e.to_string())));
}

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

A. error, wait, retry
svc.poll_ready -> Ready(Ok(()))
  policy.clone_request(attempt=0)
inner got Req { body: "hello", attempt: 0 }
  policy.retry(attempt=0, Err("boom")) left=1 -> retry
fut.poll -> Pending  (Called -> Waiting)
gate opened, fut woken: true
fut.poll -> Pending  (Waiting -> Retrying, inner not ready)
inner has a request? false
allow(1), fut woken: true
  policy.clone_request(attempt=1)
fut.poll -> Pending  (Retrying -> Called)
inner got Req { body: "hello", attempt: 1 }
  policy.retry(attempt=1, Ok("world")) left=0 -> stop
fut.poll -> Ready(Ok("world"))
B. a second request gets a fresh policy clone
svc.poll_ready -> Ready(Ok(()))
  policy.clone_request(attempt=0)
  policy.retry(attempt=0, Err("boom")) left=1 -> retry
fut.poll -> Pending
  policy.clone_request(attempt=1)
fut.poll -> Pending
  policy.retry(attempt=1, Err("boom again")) left=0 -> stop
fut.poll -> Ready(Err("boom again"))
C. inner poll_ready fails while Retrying
svc.poll_ready -> Ready(Ok(()))
  policy.clone_request(attempt=0)
  policy.retry(attempt=0, Err("boom")) left=1 -> retry
fut.poll -> Pending
fut.poll -> Ready(Err("inner broken"))

对照状态机读：

  • A：allow(1) 只放行一次，第一次 call 用掉了。第一次 fut.poll 拿到 boom，策略说重试，进 Waiting，等待 future 没完成，Pending。打开闸门后 future 被唤醒，第二次 poll 走到 Retrying：副本服务的 poll_ready 是 Pending，内层也确实没收到新请求。allow(1) 唤醒了这个 future（等待内层就绪的是它，不是调用方的 svc），第三次 poll 发出重试，内层收到的请求里 attempt 是 1，就是策略改过的那个。最终成功，策略照样被问了一次，返回 None。
  • B：同一个 svc 的第二个请求，策略的 left 又是 1，因为它用的是一份新复制的策略。第二次失败时 left=0，策略放弃，boom again 原样返回。
  • C：Handle::send_error 让内层下一次 poll_ready 报错。重试前的 poll_ready 拿到这个错，future 直接以它结束，策略没有被调用（没有第二行 policy.retry）。也就是说，策略只看得到内层 call 的结果，看不到 poll_ready 的错误。按 Service trait 的约定，poll_ready 报错表示这个服务实例已经不能用了，重试也没有意义。

重试在 future 里问就绪

Retrying 里的 poll_ready 是在响应 future 里调用的，也就是在等待响应的那个任务里。对调用方来说，它只 poll_ready 过一次、call 过一次，之后所有的重试、等待、再就绪都藏在 future 里。

这决定了别的中间件放在 Retry 外面还是里面，效果完全不同：

  • 放在 Retry 外面：它只看到一个请求。外层的 call 只发生一次，重试不会经过它。
  • 放在 Retry 里面：每次重试都是一次新的 poll_ready + call。

下面拿 ConcurrencyLimit 试两种顺序，上限 1，用 GlobalConcurrencyLimitLayer::with_semaphore 把信号量握在手里，看每一步还剩几个许可：

tower = { version = "=0.5.3", features = ["retry", "limit"] }
tower-test = "0.4"
tokio-test = "0.4"
tokio = { version = "=1.54.1", features = ["macros", "rt", "sync"] }

use std::{future::Future, pin::Pin, sync::Arc};
use tokio::sync::{Notify, Semaphore};
use tokio_test::task;
use tower::{limit::GlobalConcurrencyLimitLayer, retry::{Policy, RetryLayer}, Layer, Service};
use tower_test::mock;

type BoxError = Box<dyn std::error::Error + Send + Sync>;

#[derive(Clone)]
struct OneRetry {
  left: u32,
  gate: Arc<Notify>,
}

impl Policy<&'static str, &'static str, BoxError> for OneRetry {
  type Future = Pin<Box<dyn Future<Output = ()>>>;

  fn retry(
    &mut self,
    _: &mut &'static str,
    result: &mut Result<&'static str, BoxError>,
  ) -> Option<Self::Future> {
    if result.is_ok() || self.left == 0 {
      return None;
    }
    self.left -= 1;
    let gate = self.gate.clone();
    Some(Box::pin(async move { gate.notified().await }))
  }

  fn clone_request(&mut self, req: &&'static str) -> Option<&'static str> {
    Some(*req)
  }
}

async fn run<S>(name: &str, svc: S, sem: Arc<Semaphore>, gate: Arc<Notify>, mut handle: mock::Handle<&'static str, &'static str>)
where
  S: Service<&'static str, Response = &'static str>,
  S::Error: std::fmt::Debug,
{
  println!("{name}");
  let mut svc = mock::Spawn::new(svc);
  assert!(svc.poll_ready().is_ready());
  let mut fut = task::spawn(svc.call("hi"));
  let (_, send) = handle.next_request().await.unwrap();
  println!("  first attempt in flight: available = {}", sem.available_permits());
  send.send_error("boom");
  assert!(fut.poll().is_pending());
  println!("  waiting to retry:        available = {}", sem.available_permits());
  gate.notify_one();
  assert!(fut.poll().is_pending());
  let (_, send) = handle.next_request().await.unwrap();
  println!("  retry in flight:         available = {}", sem.available_permits());
  send.send_response("ok");
  println!("  result: {:?}", fut.poll());
  drop(fut);
  println!("  future dropped:          available = {}", sem.available_permits());
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let gate = Arc::new(Notify::new());
  let policy = OneRetry { left: 1, gate: gate.clone() };

  // Retry outside, limit inside: each attempt takes its own permit
  let sem = Arc::new(Semaphore::new(1));
  let (inner, handle) = mock::pair::<&'static str, &'static str>();
  let svc = RetryLayer::new(policy.clone())
    .layer(GlobalConcurrencyLimitLayer::with_semaphore(sem.clone()).layer(inner));
  run("Retry(ConcurrencyLimit(mock))", svc, sem, gate.clone(), handle).await;

  // Limit outside, Retry inside: the whole retry session holds one permit
  let sem = Arc::new(Semaphore::new(1));
  let (inner, handle) = mock::pair::<&'static str, &'static str>();
  let svc = GlobalConcurrencyLimitLayer::with_semaphore(sem.clone())
    .layer(RetryLayer::new(policy.clone()).layer(inner));
  run("ConcurrencyLimit(Retry(mock))", svc, sem, gate.clone(), handle).await;
}
❗IMPORTANT
动脑筋：Retry(ConcurrencyLimit(mock)) 那一组，waiting to retry 一行的 available 是 0 还是 1？

第一次请求失败，策略说要重试，future 正在 Waiting 里等闸门。这时候上限为 1 的信号量里还剩几个许可？ConcurrencyLimit(Retry(mock)) 那一组呢？完整输出在文末。

同样的道理也适用于 Timeout：Timeout::call 每调用一次就新建一个 Sleep。套在 Retry 外面，整个重试会话（包括所有等待）共用一个期限；套在里面，每次尝试各自计时，超时产生的 Elapsed 错误会交给策略判断要不要重试。

ServiceBuilder 里写 .retry(policy)（builder/mod.rs 第 249 行）就是加一层 RetryLayer，先写的在外层，所以 .retry(p).concurrency_limit(1) 是上面第一种顺序。层的顺序规则见 ServiceBuilder。

动脑筋答案

clones 是几

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

  req 1: clone_request clones=1
  req 1: retry         clones=1 retries=1
  req 1: clone_request clones=2
  req 1: retry         clones=2 retries=2
  req 2: clone_request clones=2
  req 2: retry         clones=2 retries=1
  req 2: clone_request clones=3
  req 2: retry         clones=3 retries=2
  req 3: clone_request clones=3
  req 3: retry         clones=3 retries=1
  req 3: clone_request clones=4
  req 3: retry         clones=4 retries=2

答案是 3。

  • 每个请求的第一次 clone_request 在 Retry::call 里，作用在服务自己的 self.policy 上，所以这一次 +1 留在了服务里：1、2、3。
  • 紧接着 self.clone() 把已经加过的值复制进 future，所以每个请求的会话副本从 1、2、3 起步。
  • 重试前的 clone_request 在 future 里，作用在副本上，+1 随着 future 一起丢掉：第一个请求里变成 2，但第二个请求又从服务上的 2 开始。
  • retries 只在副本上改，所以每个请求都从 1 开始。

结论：只在 retry 里改策略的字段是安全的；在 clone_request 里改，第一次的修改会漏到之后所有请求上。

剩几个许可

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

Retry(ConcurrencyLimit(mock))
  first attempt in flight: available = 0
  waiting to retry:        available = 1
  retry in flight:         available = 0
  result: Ready(Ok("ok"))
  future dropped:          available = 1
ConcurrencyLimit(Retry(mock))
  first attempt in flight: available = 0
  waiting to retry:        available = 0
  retry in flight:         available = 0
  result: Ready(Ok("ok"))
  future dropped:          available = 1
  • 第一组 waiting to retry 是 1。许可在 ConcurrencyLimit 的响应 future 里，这个 future 是 Called 状态里的 future。进入 Waiting 时 state.set 把它换掉，许可随之归还。退避期间不占并发名额，重试前副本服务的 poll_ready 再去排队取一个。
  • 第二组一直是 0。ConcurrencyLimit 在外层，它的许可包着整个 Retry 的响应 future，从第一次发送一直占到 future 被丢弃，等待期间也占着。
  • 两组的 result 那一行之后许可都还没还：最后一次的响应 future 还在 ResponseFuture 里，要等调用方丢掉它。这和 ConcurrencyLimit 里「许可的寿命是响应 future 的寿命」是同一回事。

小结

  • Policy::retry 看每次 call 的结果，返回 None 结束，返回 Some(fut) 等 fut 完成后重试；可以改请求和结果。clone_request 返回 None 的请求不会重试。
  • Retry::poll_ready 原样转发。call 先复制请求、再发请求、最后复制整个 Retry 进 future，所以内层和策略都要 Clone，每个请求都付一次复制的钱。
  • 响应 future 在 Called → Waiting → Retrying → Called 之间循环。Retrying 自己调副本服务的 poll_ready，报错直接返回，不经过策略。
  • 策略每个请求一份副本，retry 里的修改不会跨请求；但第一次 clone_request 改的是服务自己的策略。
  • 重试藏在响应 future 里：Retry 外面的中间件只看到一个请求，里面的看到每一次尝试。ConcurrencyLimit 放里面，退避期间不占许可。

💬 关联讨论

发起讨论
加载中...