Clone is not satisfied：RateLimit 放不进 axum 和 Retry

tower = { version = "=0.5.3", features = ["limit"] }
tokio = { version = "=1.54.1", features = ["full"] }
axum = "=0.8.9"

use axum::{Router, routing::get};
use std::time::Duration;
use tower::limit::RateLimitLayer;

#[tokio::main]
async fn main() {
  let app: Router = Router::new()
    .route("/", get(|| async { "hi" }))
    .layer(RateLimitLayer::new(5, Duration::from_secs(1)));
  let _ = app;
}

error[E0277]: the trait bound `RateLimit<Route>: Clone` is not satisfied
   --> src/bin/clone_axum_bad.rs:9:12
    |
  9 |     .layer(RateLimitLayer::new(5, Duration::from_secs(1)));
    |      ----- ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ the trait `Clone` is not implemented for `RateLimit<Route>`
    |      |
    |      required by a bound introduced by this call
    |
note: required by a bound in `Router::<S>::layer`
   --> ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/axum-0.8.9/src/routing/mod.rs:306:40
    |
303 |     pub fn layer<L>(self, layer: L) -> Router<S>
    |            ----- required by a bound in this associated function
...
306 |         L::Service: Service<Request> + Clone + Send + Sync + 'static,
    |                                        ^^^^^ required by this bound in `Router::<S>::layer`

  let app: Router = Router::new()
    .route("/", get(|| async { "hi" }))
    .layer(
      ServiceBuilder::new()
        .buffer(64)
        .rate_limit(5, Duration::from_secs(1)),
    );

error[E0277]: the trait bound `Infallible: From<Box<(dyn std::error::Error + Send + Sync + 'static)>>` is not satisfied
   --> src/bin/clone_axum_buffer.rs:10:7
    |
  9 |       .layer(
    |        ----- required by a bound introduced by this call
 10 | /       ServiceBuilder::new()
 11 | |         .buffer(64)
 12 | |         .rate_limit(5, Duration::from_secs(1)),
    | |______________________________________________^ the trait `From<Box<(dyn std::error::Error + Send + Sync + 'static)>>` is not implemented for `Infallible`
    |
    = note: required for `Box<(dyn std::error::Error + Send + Sync + 'static)>` to implement `Into<Infallible>`
note: required by a bound in `Router::<S>::layer`
   --> ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/axum-0.8.9/src/routing/mod.rs:308:50
    |
308 |         <L::Service as Service<Request>>::Error: Into<Infallible> + 'static,
    |                                                  ^^^^^^^^^^^^^^^^ required by this bound in `Router::<S>::layer`

tower = { version = "=0.5.3", features = ["buffer", "limit", "util"] }
tokio = { version = "=1.54.1", features = ["full", "test-util"] }
axum = "=0.8.9"

use axum::{
  Router,
  body::Body,
  error_handling::HandleErrorLayer,
  http::{Request, StatusCode},
  routing::get,
};
use std::time::Duration;
use tokio::time::Instant;
use tower::{BoxError, ServiceBuilder, ServiceExt};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  let app: Router = Router::new()
    .route("/a", get(|| async { "a" }))
    .route("/b", get(|| async { "b" }))
    .layer(
      ServiceBuilder::new()
        .layer(HandleErrorLayer::new(|e: BoxError| async move {
          (StatusCode::INTERNAL_SERVER_ERROR, e.to_string())
        }))
        .buffer(64)
        .rate_limit(2, Duration::from_secs(1)),
    );

  let start = Instant::now();
  for uri in ["/a", "/a", "/a", "/b", "/b"] {
    let req = Request::builder().uri(uri).body(Body::empty()).unwrap();
    let rsp = app.clone().oneshot(req).await.unwrap();
    println!("{uri}: {} at {}ms", rsp.status(), start.elapsed().as_millis());
  }
}

/a: 200 OK at 0ms
/a: 200 OK at 0ms
/a: 200 OK at 0ms
/b: 200 OK at 0ms
/b: 200 OK at 0ms

/a: 200 OK at 0ms
/a: 200 OK at 0ms
/a: 200 OK at 1000ms
/b: 200 OK at 1000ms
/b: 200 OK at 1000ms

tower = { version = "=0.5.3", features = ["buffer", "limit"] }
tokio = { version = "=1.54.1", features = ["full"] }
axum = "=0.8.9"

use axum::{Router, error_handling::HandleErrorLayer, http::StatusCode, routing::get};
use std::time::{Duration, Instant};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tower::{BoxError, ServiceBuilder};

// Send one GET on an open connection and return the status line.
async fn http_get(conn: &mut TcpStream, path: &str) -> String {
  let req = format!("GET {path} HTTP/1.1\r\nHost: x\r\n\r\n");
  conn.write_all(req.as_bytes()).await.unwrap();
  let mut buf = Vec::new();
  let mut byte = [0u8; 1];
  while !buf.ends_with(b"\r\n\r\n") {
    conn.read_exact(&mut byte).await.unwrap();
    buf.push(byte[0]);
  }
  let head = String::from_utf8(buf).unwrap();
  let len: usize = head
    .lines()
    .find_map(|l| l.strip_prefix("content-length: "))
    .unwrap()
    .parse()
    .unwrap();
  let mut body = vec![0u8; len];
  conn.read_exact(&mut body).await.unwrap();
  head.lines().next().unwrap().to_string()
}

#[tokio::main]
async fn main() {
  let app: Router = Router::new().route("/a", get(|| async { "a" })).layer(
    ServiceBuilder::new()
      .layer(HandleErrorLayer::new(|e: BoxError| async move {
        (StatusCode::INTERNAL_SERVER_ERROR, e.to_string())
      }))
      .buffer(64)
      .rate_limit(2, Duration::from_secs(1)),
  );
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });

  let start = Instant::now();
  println!("one connection per request:");
  for _ in 0..3 {
    let mut conn = TcpStream::connect(addr).await.unwrap();
    let status = http_get(&mut conn, "/a").await;
    println!("  {status} at {}ms", start.elapsed().as_millis());
  }

  let start = Instant::now();
  println!("three requests on one connection:");
  let mut conn = TcpStream::connect(addr).await.unwrap();
  for _ in 0..3 {
    let status = http_get(&mut conn, "/a").await;
    println!("  {status} at {}ms", start.elapsed().as_millis());
  }
}

one connection per request:
  HTTP/1.1 200 OK at 0ms
  HTTP/1.1 200 OK at 0ms
  HTTP/1.1 200 OK at 1ms
three requests on one connection:
  HTTP/1.1 200 OK at 0ms
  HTTP/1.1 200 OK at 0ms
  HTTP/1.1 200 OK at 1001ms

tower = { version = "=0.5.3", features = ["buffer", "limit", "make"] }

use tower::{BoxError, ServiceBuilder, make::Shared};

// http_get 同上

#[tokio::main]
async fn main() {
  let app: Router = Router::new()
    .route("/a", get(|| async { "a" }))
    .route("/b", get(|| async { "b" }));
  // Build the limited service once, around the whole router.
  let svc = ServiceBuilder::new()
    .layer(HandleErrorLayer::new(|e: BoxError| async move {
      (StatusCode::INTERNAL_SERVER_ERROR, e.to_string())
    }))
    .buffer(64)
    .rate_limit(2, Duration::from_secs(1))
    .service(app);
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(async move { axum::serve(listener, Shared::new(svc)).await.unwrap() });

  let start = Instant::now();
  for path in ["/a", "/b", "/a"] {
    let mut conn = TcpStream::connect(addr).await.unwrap();
    let status = http_get(&mut conn, path).await;
    println!("{path} on a new connection: {status} at {}ms", start.elapsed().as_millis());
  }
}

/a on a new connection: HTTP/1.1 200 OK at 0ms
/b on a new connection: HTTP/1.1 200 OK at 0ms
/a on a new connection: HTTP/1.1 200 OK at 1003ms

tower = { version = "=0.5.3", features = ["retry", "limit", "util"] }
tokio = { version = "=1.54.1", features = ["full"] }

use std::future::{Ready, ready};
use std::time::Duration;
use tower::retry::Policy;
use tower::{BoxError, ServiceBuilder, ServiceExt};

// Retry every error up to `left` more times.
#[derive(Clone)]
struct Attempts(usize);

impl<Req: Clone, Res, E> Policy<Req, Res, E> for Attempts {
  type Future = Ready<()>;

  fn retry(&mut self, _: &mut Req, result: &mut Result<Res, E>) -> Option<Self::Future> {
    if result.is_err() && self.0 > 0 {
      self.0 -= 1;
      Some(ready(()))
    } else {
      None
    }
  }

  fn clone_request(&mut self, req: &Req) -> Option<Req> {
    Some(req.clone())
  }
}

#[tokio::main]
async fn main() {
  let svc = ServiceBuilder::new()
    .retry(Attempts(2))
    .rate_limit(10, Duration::from_secs(1))
    .service_fn(|x: u32| async move { Ok::<_, BoxError>(x) });
  let rsp = svc.oneshot(1).await.unwrap();
  println!("{rsp}");
}

error[E0599]: the method `oneshot` exists for struct `Retry<Attempts, RateLimit<ServiceFn<{closure@src/bin/clone_retry_bad.rs:32:17: 32:25}>>>`, but its trait bounds were not satisfied
   --> src/bin/clone_retry_bad.rs:33:17
    |
 33 |     let rsp = svc.oneshot(1).await.unwrap();
    |                   ^^^^^^^ method cannot be called due to unsatisfied trait bounds
    |
    = note: the following trait bounds were not satisfied:
            `Retry<Attempts, RateLimit<ServiceFn<{closure@src/bin/clone_retry_bad.rs:32:17: 32:25}>>>: Service<_>`
            which is required by `Retry<Attempts, RateLimit<ServiceFn<{closure@src/bin/clone_retry_bad.rs:32:17: 32:25}>>>: ServiceExt<_>`

// Asks the compiler the exact question: is this a Service?
fn assert_service<S: Service<u32>>(_: &S) {}

// in main, before oneshot:
  assert_service(&svc);

error[E0277]: the trait bound `RateLimit<ServiceFn<{closure@src/bin/clone_retry_assert.rs:35:17: 35:25}>>: Clone` is not satisfied
  --> src/bin/clone_retry_assert.rs:36:18
   |
36 |   assert_service(&svc);
   |   -------------- ^^^^ the trait `Clone` is not implemented for `RateLimit<ServiceFn<{closure@src/bin/clone_retry_assert.rs:35:17: 35:25}>>`
   |   |
   |   required by a bound introduced by this call
   |
help: the trait `Service<Request>` is conditionally implemented for `Retry<P, S>`
  --> ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tower-0.5.3/src/retry/mod.rs:72:1
   |
72 | / impl<P, S, Request> Service<Request> for Retry<P, S>
73 | | where
74 | |     P: Policy<Request, S::Response, S::Error> + Clone,
75 | |     S: Service<Request> + Clone,
   | |___________________________-----^
   |                             |
   |                             unsatisfied requirement introduced here: `RateLimit<ServiceFn<{closure@src/bin/clone_retry_assert.rs:35:17: 35:25}>>: Clone`
   = note: required for `Retry<Attempts, RateLimit<ServiceFn<{closure@src/bin/clone_retry_assert.rs:35:17: 35:25}>>>` to implement `Service<u32>`

tower = { version = "=0.5.3", features = ["retry", "buffer", "limit", "util"] }
tokio = { version = "=1.54.1", features = ["full", "test-util"] }

// Attempts 同上；另外 use std::sync::Arc、std::sync::atomic::{AtomicU32, Ordering}、tokio::time::Instant

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  let start = Instant::now();
  let tries = Arc::new(AtomicU32::new(0));
  let svc = ServiceBuilder::new()
    .retry(Attempts(2))
    .buffer(8)
    .rate_limit(2, Duration::from_secs(1))
    .service_fn(move |x: u32| {
      let n = tries.fetch_add(1, Ordering::SeqCst) + 1;
      async move {
        println!("attempt {n} at {}ms", start.elapsed().as_millis());
        if n < 3 { Err::<u32, BoxError>("flaky".into()) } else { Ok(x) }
      }
    });
  let rsp = svc.oneshot(7).await.unwrap();
  println!("response {rsp} at {}ms", start.elapsed().as_millis());
}

attempt 1 at 0ms
attempt 2 at 0ms
attempt 3 at 1000ms
response 7 at 1000ms