RateLimit

#[derive(Debug)]
pub struct RateLimit<T> {
    inner: T,
    rate: Rate,
    state: State,
    sleep: Pin<Box<Sleep>>,
}

#[derive(Debug)]
enum State {
    // The service has hit its limit
    Limited,
    Ready { until: Instant, rem: u64 },
}

    fn call(&mut self, request: Request) -> Self::Future {
        match self.state {
            State::Ready { mut until, mut rem } => {
                let now = Instant::now();

                // If the period has elapsed, reset it.
                if now >= until {
                    until = now + self.rate.per();
                    rem = self.rate.num();
                }

                if rem > 1 {
                    rem -= 1;
                    self.state = State::Ready { until, rem };
                } else {
                    // The service is disabled until further notice
                    // Reset the sleep future in place, so that we don't have to
                    // deallocate the existing box and allocate a new one.
                    self.sleep.as_mut().reset(until);
                    self.state = State::Limited;
                }

                // Call the inner future
                self.inner.call(request)
            }
            State::Limited => panic!("service not ready; poll_ready must be called first"),
        }
    }

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        match self.state {
            State::Ready { .. } => return self.inner.poll_ready(cx),
            State::Limited => {
                if Pin::new(&mut self.sleep).poll(cx).is_pending() {
                    tracing::trace!("rate limit exceeded; sleeping.");
                    return Poll::Pending;
                }
            }
        }

        self.state = State::Ready {
            until: Instant::now() + self.rate.per(),
            rem: self.rate.num(),
        };

        self.inner.poll_ready(cx)
    }

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

use std::{convert::Infallible, time::Duration};
use tokio::time::{sleep, Instant};
use tower::{limit::RateLimitLayer, service_fn, Layer, Service, ServiceExt};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  let start = Instant::now();
  let ms = move || start.elapsed().as_millis();
  let inner = service_fn(|x: u32| async move { Ok::<_, Infallible>(x) });
  // 3 requests per second
  let mut svc = RateLimitLayer::new(3, Duration::from_secs(1)).layer(inner);

  svc.ready().await.unwrap().call(0).await.unwrap();
  println!("req 0 at {}ms", ms());

  sleep(Duration::from_millis(900)).await;
  for id in 1..=5 {
    svc.ready().await.unwrap().call(id).await.unwrap();
    println!("req {id} at {}ms", ms());
  }
}

use std::{convert::Infallible, time::Duration};
use tower::{limit::RateLimitLayer, service_fn, Layer};

fn main() {
  let inner = service_fn(|x: u32| async move { Ok::<_, Infallible>(x) });
  let svc = RateLimitLayer::new(2, Duration::from_secs(1)).layer(inner);
  let _copy = svc.clone();
}

error[E0599]: no method named `clone` found for struct `RateLimit<T>` in the current scope
 --> src/bin/rate_clone.rs:7:19
  |
7 |   let _copy = svc.clone();
  |                   ^^^^^ method not found in `RateLimit<ServiceFn<{closure@src/bin/rate_clone.rs:5:26: 5:34}>>`

For more information about this error, try `rustc --explain E0599`.
error: could not compile `exp-tower-mwa` (bin "rate_clone") due to 1 previous error

use std::{
  convert::Infallible,
  panic::{catch_unwind, AssertUnwindSafe},
  time::Duration,
};
use tower::{limit::RateLimitLayer, service_fn, Layer, Service};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  let inner = service_fn(|x: u32| async move { Ok::<_, Infallible>(x) });
  let layer = RateLimitLayer::new(2, Duration::from_secs(1));
  let mut a = layer.layer(inner);
  let mut b = layer.layer(inner);

  // never call poll_ready: fine while the state is Ready
  println!("a.call(1) -> {:?}", a.call(1).await);
  println!("a.call(2) -> {:?}", a.call(2).await);
  // two instances made by the same layer count separately
  println!("b.call(3) -> {:?}", b.call(3).await);

  std::panic::set_hook(Box::new(|_| {})); // keep the default hook quiet
  let r = catch_unwind(AssertUnwindSafe(|| {
    let _ = a.call(4); // a is Limited now
  }));
  let p = r.unwrap_err();
  println!("a.call(4) panics: {:?}", p.downcast_ref::<&str>());
}

a.call(1) -> Ok(1)
a.call(2) -> Ok(2)
b.call(3) -> Ok(3)
a.call(4) panics: Some("service not ready; poll_ready must be called first")

req 0 at 0ms
req 1 at 900ms
req 2 at 900ms
req 3 at 1000ms
req 4 at 1000ms
req 5 at 1000ms