use std::{collections::VecDeque, fmt, time::Duration};
use parking_lot::Mutex;
use tokio::{
sync::Notify,
time::{Instant, sleep},
};
use crate::Error;
const DEFAULT_HISTORY_REQUESTS: usize = 50;
const DEFAULT_HISTORY_WINDOW: Duration = Duration::from_secs(30);
const DEFAULT_GENERAL_REQUESTS: usize = 200;
const DEFAULT_GENERAL_WINDOW: Duration = Duration::from_mins(1);
const MAX_INITIAL_CAPACITY: usize = 1_024;
pub(crate) const MAX_COOLDOWN: Duration = Duration::from_hours(24);
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
#[non_exhaustive]
pub enum RateLimitKind {
History,
General,
}
impl fmt::Display for RateLimitKind {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::History => formatter.write_str("history"),
Self::General => formatter.write_str("general"),
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub struct RateLimit {
max_requests: usize,
window: Duration,
}
impl RateLimit {
pub fn new(max_requests: usize, window: Duration) -> Result<Self, Error> {
if max_requests == 0
|| window.is_zero()
|| std::time::Instant::now().checked_add(window).is_none()
{
return Err(Error::Configuration(
"rate limits require a positive request count and representable window".to_owned(),
));
}
Ok(Self {
max_requests,
window,
})
}
const fn trusted(max_requests: usize, window: Duration) -> Self {
Self {
max_requests,
window,
}
}
#[must_use]
pub const fn max_requests(self) -> usize {
self.max_requests
}
#[must_use]
pub const fn window(self) -> Duration {
self.window
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub struct RateLimitConfig {
history: RateLimit,
general: RateLimit,
exit_reserve: usize,
}
impl RateLimitConfig {
#[must_use]
pub const fn new(history: RateLimit, general: RateLimit) -> Self {
Self {
history,
general,
exit_reserve: 0,
}
}
pub fn with_exit_reserve(mut self, reserve: usize) -> Result<Self, Error> {
if reserve >= self.general.max_requests {
return Err(Error::Configuration(
"exit reserve must be smaller than the general request limit".to_owned(),
));
}
self.exit_reserve = reserve;
Ok(self)
}
#[must_use]
pub const fn exit_reserve(self) -> usize {
self.exit_reserve
}
#[must_use]
pub const fn history(self) -> RateLimit {
self.history
}
#[must_use]
pub const fn general(self) -> RateLimit {
self.general
}
pub(crate) const fn limit(self, kind: RateLimitKind) -> RateLimit {
match kind {
RateLimitKind::History => self.history,
RateLimitKind::General => self.general,
}
}
}
impl Default for RateLimitConfig {
fn default() -> Self {
Self {
history: RateLimit::trusted(DEFAULT_HISTORY_REQUESTS, DEFAULT_HISTORY_WINDOW),
general: RateLimit::trusted(DEFAULT_GENERAL_REQUESTS, DEFAULT_GENERAL_WINDOW),
exit_reserve: 0,
}
}
}
pub(crate) struct RateLimits {
config: Option<RateLimitConfig>,
history: Option<WindowLimiter>,
general: Option<WindowLimiter>,
}
impl RateLimits {
pub(crate) fn new(config: Option<RateLimitConfig>) -> Self {
let history = config.map(|limits| WindowLimiter::new(limits.history, 0));
let general = config.map(|limits| WindowLimiter::new(limits.general, limits.exit_reserve));
Self {
config,
history,
general,
}
}
pub(crate) fn config(&self) -> Option<RateLimitConfig> {
self.config.map(|mut config| {
if let Some(limiter) = &self.general {
config.exit_reserve = limiter.state.lock().exit_reserve;
}
config
})
}
pub(crate) fn set_exit_reserve(&self, reserve: usize) -> Result<(), Error> {
let limiter = self.general.as_ref().ok_or_else(|| {
Error::Configuration("exit reserve requires enabled rate limits".to_owned())
})?;
let mut state = limiter.state.lock();
if reserve >= limiter.limit.max_requests {
return Err(Error::Configuration(
"exit reserve must be smaller than the general request limit".to_owned(),
));
}
let changed = state.exit_reserve != reserve;
state.exit_reserve = reserve;
drop(state);
if changed {
limiter.policy_changed.notify_waiters();
}
Ok(())
}
pub(crate) fn limit(&self, kind: RateLimitKind) -> RateLimit {
self.config.unwrap_or_default().limit(kind)
}
pub(crate) async fn wait(&self, kind: RateLimitKind) {
let Some(limiter) = self.limiter(kind) else {
return;
};
loop {
let changed = limiter.policy_changed.notified();
tokio::pin!(changed);
changed.as_mut().enable();
match limiter.try_acquire(false) {
Ok(()) => return,
Err(retry_after) => {
tokio::select! {
() = &mut changed => {},
() = sleep(retry_after) => {},
}
}
}
}
}
pub(crate) fn try_acquire(&self, kind: RateLimitKind) -> Result<(), Duration> {
self.limiter(kind)
.map_or(Ok(()), |limiter| limiter.try_acquire(false))
}
pub(crate) fn try_acquire_exit(&self) -> Result<(), Duration> {
self.general
.as_ref()
.map_or(Ok(()), |limiter| limiter.try_acquire(true))
}
pub(crate) fn cool_down(&self, kind: RateLimitKind, retry_after: Duration) {
if let Some(limiter) = self.limiter(kind) {
limiter.cool_down(retry_after);
}
}
fn limiter(&self, kind: RateLimitKind) -> Option<&WindowLimiter> {
match kind {
RateLimitKind::History => self.history.as_ref(),
RateLimitKind::General => self.general.as_ref(),
}
}
}
struct WindowLimiter {
limit: RateLimit,
policy_changed: Notify,
state: Mutex<WindowState>,
}
impl WindowLimiter {
fn new(limit: RateLimit, exit_reserve: usize) -> Self {
Self {
limit,
policy_changed: Notify::new(),
state: Mutex::new(WindowState::with_capacity(
limit.max_requests.min(MAX_INITIAL_CAPACITY),
exit_reserve,
)),
}
}
fn try_acquire(&self, exit: bool) -> Result<(), Duration> {
let mut state = self.state.lock();
let capacity = if exit {
self.limit.max_requests
} else {
self.limit.max_requests - state.exit_reserve
};
let now = Instant::now();
state.expire(now, self.limit.window);
let retry_after = state.retry_after(now, self.limit.window, capacity);
if retry_after.is_zero() {
state.admitted.push_back(now);
Ok(())
} else {
Err(retry_after)
}
}
fn cool_down(&self, retry_after: Duration) {
let mut state = self.state.lock();
let deadline = saturating_deadline(Instant::now(), retry_after.min(MAX_COOLDOWN));
if state
.cooldown_until
.is_none_or(|current| deadline > current)
{
state.cooldown_until = Some(deadline);
}
}
}
struct WindowState {
exit_reserve: usize,
admitted: VecDeque<Instant>,
cooldown_until: Option<Instant>,
}
impl WindowState {
fn with_capacity(capacity: usize, exit_reserve: usize) -> Self {
Self {
exit_reserve,
admitted: VecDeque::with_capacity(capacity),
cooldown_until: None,
}
}
fn expire(&mut self, now: Instant, window: Duration) {
while self
.admitted
.front()
.is_some_and(|admitted| now.saturating_duration_since(*admitted) >= window)
{
self.admitted.pop_front();
}
if self.cooldown_until.is_some_and(|deadline| deadline <= now) {
self.cooldown_until = None;
}
}
fn retry_after(&self, now: Instant, window: Duration, capacity: usize) -> Duration {
let cooldown = self.cooldown_until.map_or(Duration::ZERO, |deadline| {
deadline.saturating_duration_since(now)
});
let capacity_delay = if self.admitted.len() < capacity {
Duration::ZERO
} else {
self.admitted
.get(self.admitted.len() - capacity)
.map_or(window, |admitted| {
admitted
.checked_add(window)
.map_or(window, |deadline| deadline.saturating_duration_since(now))
})
};
cooldown.max(capacity_delay)
}
}
fn saturating_deadline(now: Instant, duration: Duration) -> Instant {
if let Some(deadline) = now.checked_add(duration) {
return deadline;
}
let mut low = 0_u64;
let mut high = u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX);
let mut deadline = now;
while high - low > 1 {
let mid = u64::midpoint(low, high);
if let Some(probe) = now.checked_add(Duration::from_nanos(mid)) {
low = mid;
deadline = probe;
} else {
high = mid;
}
}
deadline
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use tokio::time::{Instant, advance};
use super::*;
fn limits(requests: usize, window: Duration) -> Arc<RateLimits> {
let limit = RateLimit::new(requests, window)
.unwrap_or_else(|error| panic!("test rate limit must be valid: {error}"));
Arc::new(RateLimits::new(Some(RateLimitConfig::new(limit, limit))))
}
#[tokio::test(start_paused = true)]
async fn wait_resumes_when_the_rolling_window_releases_capacity() {
let window = Duration::from_secs(10);
let limits = limits(1, window);
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
let started = Instant::now();
let waiting_limits = Arc::clone(&limits);
let waiter = tokio::spawn(async move {
waiting_limits.wait(RateLimitKind::General).await;
Instant::now()
});
tokio::task::yield_now().await;
assert!(!waiter.is_finished());
advance(window).await;
let admitted = waiter
.await
.unwrap_or_else(|error| panic!("waiting task must finish: {error}"));
assert_eq!(admitted.saturating_duration_since(started), window);
}
#[tokio::test(start_paused = true)]
async fn provider_cooldown_blocks_admission_until_its_deadline() {
let cooldown = Duration::from_secs(15);
let limits = limits(2, Duration::from_secs(30));
limits.cool_down(RateLimitKind::General, cooldown);
assert_eq!(limits.try_acquire(RateLimitKind::General), Err(cooldown));
advance(cooldown).await;
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
}
#[tokio::test(start_paused = true)]
async fn an_oversized_cooldown_is_clamped_not_skipped() {
let limits = limits(2, Duration::from_secs(30));
limits.cool_down(RateLimitKind::General, Duration::MAX);
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_hours(24)),
"an unrepresentable provider cooldown must still block admission"
);
}
#[test]
fn an_unrepresentable_deadline_search_terminates_and_still_blocks() {
let now = Instant::now();
let deadline = saturating_deadline(now, Duration::MAX);
let offset = deadline.saturating_duration_since(now);
assert!(
offset > Duration::from_hours(24) && offset <= Duration::from_nanos(u64::MAX),
"the saturated deadline must still block admission within the \
searched nanosecond domain, found {offset:?}"
);
}
#[tokio::test(start_paused = true)]
async fn rolling_window_expires_staggered_attempts_individually() {
let window = Duration::from_secs(10);
let limits = limits(2, window);
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
advance(Duration::from_secs(5)).await;
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
advance(Duration::from_secs(5)).await;
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_secs(5))
);
}
#[tokio::test(start_paused = true)]
async fn cancelled_waiter_leaves_no_ghost_reservation() {
let window = Duration::from_secs(10);
let limits = limits(1, window);
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
let waiting_limits = Arc::clone(&limits);
let waiter = tokio::spawn(async move {
waiting_limits.wait(RateLimitKind::General).await;
});
tokio::task::yield_now().await;
waiter.abort();
advance(window).await;
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
}
#[tokio::test(start_paused = true)]
async fn default_history_budget_reaches_its_limit_before_throttling() {
let limits = Arc::new(RateLimits::new(Some(RateLimitConfig::default())));
for request_number in 1..=50 {
assert_eq!(
limits.try_acquire(RateLimitKind::History),
Ok(()),
"history request {request_number} must be admitted"
);
}
assert_eq!(
limits.try_acquire(RateLimitKind::History),
Err(Duration::from_secs(30))
);
}
#[tokio::test(start_paused = true)]
async fn default_general_budget_reaches_its_limit_before_throttling() {
let limits = Arc::new(RateLimits::new(Some(RateLimitConfig::default())));
for request_number in 1..=200 {
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Ok(()),
"general request {request_number} must be admitted"
);
}
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_mins(1))
);
}
#[tokio::test(start_paused = true)]
async fn later_provider_cooldown_cannot_be_shortened() {
let limits = limits(200, Duration::from_mins(1));
limits.cool_down(RateLimitKind::General, Duration::from_secs(30));
limits.cool_down(RateLimitKind::General, Duration::from_secs(5));
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_secs(30))
);
}
#[tokio::test(start_paused = true)]
async fn admission_uses_the_later_of_window_and_provider_cooldown() {
let limits = limits(1, Duration::from_secs(20));
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
limits.cool_down(RateLimitKind::General, Duration::from_secs(5));
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_secs(20))
);
}
#[test]
fn history_and_general_budgets_are_independent() {
let limits = limits(1, Duration::from_secs(30));
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
assert_eq!(limits.try_acquire(RateLimitKind::History), Ok(()));
assert!(limits.try_acquire(RateLimitKind::General).is_err());
assert!(limits.try_acquire(RateLimitKind::History).is_err());
}
#[test]
fn public_defaults_match_the_provider_contract() {
let config = RateLimitConfig::default();
assert_eq!(config.history().max_requests(), 50);
assert_eq!(config.history().window(), Duration::from_secs(30));
assert_eq!(config.general().max_requests(), 200);
assert_eq!(config.general().window(), Duration::from_mins(1));
}
#[test]
fn public_rate_limit_rejects_zero_values() {
assert!(RateLimit::new(0, Duration::from_secs(1)).is_err());
assert!(RateLimit::new(1, Duration::ZERO).is_err());
}
fn reserved_limits() -> Arc<RateLimits> {
let config = RateLimitConfig::new(
RateLimit::new(5, Duration::from_secs(10)).unwrap_or_else(|error| panic!("{error}")),
RateLimit::new(5, Duration::from_secs(10)).unwrap_or_else(|error| panic!("{error}")),
)
.with_exit_reserve(2)
.unwrap_or_else(|error| panic!("{error}"));
Arc::new(RateLimits::new(Some(config)))
}
#[tokio::test(start_paused = true)]
async fn exit_reserve_keeps_ordinary_waiters_below_the_shared_aggregate_cap() {
let limits = reserved_limits();
for _ in 0..3 {
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
}
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_secs(10))
);
let waiting_limits = Arc::clone(&limits);
let waiter = tokio::spawn(async move {
waiting_limits.wait(RateLimitKind::General).await;
});
tokio::task::yield_now().await;
assert!(!waiter.is_finished());
assert_eq!(limits.try_acquire_exit(), Ok(()));
tokio::task::yield_now().await;
assert!(!waiter.is_finished());
assert_eq!(limits.try_acquire_exit(), Ok(()));
assert_eq!(limits.try_acquire_exit(), Err(Duration::from_secs(10)));
waiter.abort();
assert!(waiter.await.is_err_and(|error| error.is_cancelled()));
advance(Duration::from_secs(10)).await;
for _ in 0..3 {
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
}
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_secs(10))
);
assert_eq!(limits.try_acquire_exit(), Ok(()));
}
#[tokio::test(start_paused = true)]
async fn ordinary_retry_after_waits_for_enough_aggregate_entries_to_expire() {
let limits = reserved_limits();
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
advance(Duration::from_secs(1)).await;
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
advance(Duration::from_secs(1)).await;
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
advance(Duration::from_secs(1)).await;
assert_eq!(limits.try_acquire_exit(), Ok(()));
advance(Duration::from_secs(1)).await;
assert_eq!(limits.try_acquire_exit(), Ok(()));
assert_eq!(limits.try_acquire_exit(), Err(Duration::from_secs(6)));
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_secs(8))
);
advance(Duration::from_secs(6)).await;
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_secs(2))
);
advance(Duration::from_secs(2)).await;
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_secs(1))
);
}
#[tokio::test(start_paused = true)]
async fn concurrent_exit_admission_never_exceeds_shared_capacity() {
let limits = reserved_limits();
for _ in 0..3 {
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
}
let mut tasks = tokio::task::JoinSet::new();
for _ in 0..20 {
let clone = Arc::clone(&limits);
tasks.spawn(async move { clone.try_acquire_exit() });
}
let mut accepted = 0;
let mut refused = 0;
while let Some(result) = tasks.join_next().await {
match result.unwrap_or_else(|error| panic!("{error}")) {
Ok(()) => accepted += 1,
Err(delay) => {
assert_eq!(delay, Duration::from_secs(10));
refused += 1;
}
}
}
assert_eq!(accepted, 2);
assert_eq!(refused, 18);
}
#[tokio::test(start_paused = true)]
async fn provider_cooldown_blocks_ordinary_and_reserved_capacity() {
let limits = reserved_limits();
limits.cool_down(RateLimitKind::General, Duration::from_secs(15));
assert_eq!(limits.try_acquire_exit(), Err(Duration::from_secs(15)));
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_secs(15))
);
advance(Duration::from_secs(14)).await;
assert_eq!(limits.try_acquire_exit(), Err(Duration::from_secs(1)));
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_secs(1))
);
advance(Duration::from_secs(1)).await;
assert_eq!(limits.try_acquire_exit(), Ok(()));
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
}
#[test]
fn exit_reserve_retains_ordinary_capacity_and_defaults_to_zero() {
let config = RateLimitConfig::default();
assert_eq!(config.exit_reserve(), 0);
assert_eq!(
RateLimitConfig::new(config.history(), config.general()).exit_reserve(),
0
);
assert_eq!(
config
.with_exit_reserve(199)
.unwrap_or_else(|error| panic!("{error}"))
.exit_reserve(),
199
);
assert!(matches!(
config.with_exit_reserve(200),
Err(Error::Configuration(_))
));
assert!(matches!(
config.with_exit_reserve(usize::MAX),
Err(Error::Configuration(_))
));
}
#[tokio::test(start_paused = true)]
async fn runtime_release_wakes_all_registered_waiters_without_advancing_time() {
let limits = reserved_limits();
for _ in 0..3 {
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
}
let first = limits.wait(RateLimitKind::General);
let second = limits.wait(RateLimitKind::General);
tokio::pin!(first, second);
assert!(futures_util::poll!(first.as_mut()).is_pending());
assert!(futures_util::poll!(second.as_mut()).is_pending());
let started = Instant::now();
let clone = Arc::clone(&limits);
clone
.set_exit_reserve(0)
.unwrap_or_else(|error| panic!("{error}"));
assert!(futures_util::poll!(first.as_mut()).is_ready());
assert!(futures_util::poll!(second.as_mut()).is_ready());
assert_eq!(Instant::now(), started);
assert_eq!(limits.config().map(RateLimitConfig::exit_reserve), Some(0));
assert_eq!(limits.try_acquire_exit(), Err(Duration::from_secs(10)));
}
#[tokio::test(start_paused = true)]
async fn runtime_growth_rechecks_waiters_and_preserves_spent_capacity() {
let limits = reserved_limits();
for _ in 0..3 {
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
}
let read = limits.wait(RateLimitKind::General);
tokio::pin!(read);
assert!(futures_util::poll!(read.as_mut()).is_pending());
limits
.set_exit_reserve(0)
.unwrap_or_else(|error| panic!("{error}"));
limits
.set_exit_reserve(4)
.unwrap_or_else(|error| panic!("{error}"));
assert!(futures_util::poll!(read.as_mut()).is_pending());
assert_eq!(limits.config().map(RateLimitConfig::exit_reserve), Some(4));
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_secs(10))
);
assert_eq!(limits.try_acquire_exit(), Ok(()));
assert_eq!(limits.try_acquire_exit(), Ok(()));
assert_eq!(limits.try_acquire_exit(), Err(Duration::from_secs(10)));
limits
.set_exit_reserve(0)
.unwrap_or_else(|error| panic!("{error}"));
limits
.set_exit_reserve(4)
.unwrap_or_else(|error| panic!("{error}"));
assert_eq!(limits.try_acquire_exit(), Err(Duration::from_secs(10)));
limits
.set_exit_reserve(0)
.unwrap_or_else(|error| panic!("{error}"));
assert!(futures_util::poll!(read.as_mut()).is_pending());
advance(Duration::from_secs(10)).await;
assert!(futures_util::poll!(read.as_mut()).is_ready());
}
#[tokio::test(start_paused = true)]
async fn runtime_changes_preserve_timestamps_cooldown_and_invalid_policy_state() {
let limits = reserved_limits();
assert_eq!(limits.try_acquire_exit(), Ok(()));
advance(Duration::from_secs(2)).await;
assert_eq!(limits.try_acquire_exit(), Ok(()));
limits.cool_down(RateLimitKind::General, Duration::from_secs(3));
for reserve in [4, 0, 2] {
limits
.set_exit_reserve(reserve)
.unwrap_or_else(|error| panic!("{error}"));
assert_eq!(limits.try_acquire_exit(), Err(Duration::from_secs(3)));
}
for reserve in [5, usize::MAX] {
assert!(matches!(
limits.set_exit_reserve(reserve),
Err(Error::Configuration(_))
));
assert_eq!(limits.config().map(RateLimitConfig::exit_reserve), Some(2));
assert_eq!(limits.try_acquire_exit(), Err(Duration::from_secs(3)));
}
advance(Duration::from_secs(3)).await;
for _ in 0..3 {
assert_eq!(limits.try_acquire_exit(), Ok(()));
}
assert_eq!(limits.try_acquire_exit(), Err(Duration::from_secs(5)));
advance(Duration::from_secs(5)).await;
assert_eq!(limits.try_acquire_exit(), Ok(()));
assert_eq!(limits.try_acquire_exit(), Err(Duration::from_secs(2)));
}
#[tokio::test(start_paused = true)]
async fn runtime_waiter_rechecks_each_replacement_including_release_then_growth() {
let limits = reserved_limits();
limits
.set_exit_reserve(4)
.unwrap_or_else(|error| panic!("{error}"));
assert_eq!(limits.try_acquire(RateLimitKind::General), Ok(()));
let read = limits.wait(RateLimitKind::General);
tokio::pin!(read);
assert!(futures_util::poll!(read.as_mut()).is_pending());
limits
.set_exit_reserve(0)
.unwrap_or_else(|error| panic!("{error}"));
limits
.set_exit_reserve(4)
.unwrap_or_else(|error| panic!("{error}"));
assert!(futures_util::poll!(read.as_mut()).is_pending());
limits
.set_exit_reserve(3)
.unwrap_or_else(|error| panic!("{error}"));
assert!(futures_util::poll!(read.as_mut()).is_ready());
assert_eq!(
limits.try_acquire(RateLimitKind::General),
Err(Duration::from_secs(10))
);
}
}