Skip to main content

polyoxide_binance/
weight.rs

1//! Binance's request-weight budget, and the separate limit on the funding routes.
2//!
3//! Binance charges each REST request a *weight* that depends on its route and
4//! its parameters, and refuses an IP whose weight in the current minute passes
5//! `exchangeInfo`'s `REQUEST_WEIGHT` limit of 2400. Core's `RateLimiter` counts
6//! requests per path, so it cannot model a cost that varies with `limit`;
7//! [`WeightBudget`] does. The table is [`Route::cost`]; the measurements behind
8//! it are in `docs/specs/binance/OBSERVED.md`.
9
10use std::sync::{Arc, Mutex, MutexGuard};
11use std::time::{Duration, SystemTime, UNIX_EPOCH};
12
13use tokio::time::Instant;
14
15use crate::usdm::types::DepthLimit;
16
17/// `exchangeInfo`'s `REQUEST_WEIGHT` limit per minute, per IP.
18pub const PUBLISHED_WEIGHT_PER_MINUTE: u32 = 2400;
19
20/// Requests per IP every five minutes that `fundingRate` and `fundingInfo`
21/// share, as documented. Neither route reports a weight.
22pub const PUBLISHED_FUNDING_PER_FIVE_MINUTES: u32 = 500;
23
24/// How long a `418` holds every request when it carries no `Retry-After`: the
25/// shortest ban Binance documents.
26pub const DEFAULT_BAN: Duration = Duration::from_secs(120);
27
28/// The longest a cooldown can last: the longest ban Binance documents. A
29/// `Retry-After` beyond it is clamped.
30pub const MAX_COOLDOWN: Duration = Duration::from_secs(3 * 24 * 60 * 60);
31
32/// Reciprocal of the share of a published limit the client leaves unused, as
33/// in core's `RESERVED_FRACTION`: aiming at a published quota is a bug, because
34/// the server's count and the client's never agree exactly.
35const RESERVED_FRACTION: u32 = 10;
36
37const MINUTE_MS: u64 = 60_000;
38const FUNDING_PERIOD: Duration = Duration::from_secs(300);
39
40fn after_reserve(published: u32) -> u32 {
41    published - published.div_ceil(RESERVED_FRACTION)
42}
43
44/// What one request costs, and which limit it draws on.
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub enum Cost {
47    /// Request weight, against the per-minute budget.
48    Weight(u32),
49    /// One request against the funding routes' own limit.
50    Funding,
51}
52
53/// A REST route, with the parameters its weight depends on.
54#[derive(Debug, Clone, Copy, PartialEq, Eq)]
55#[non_exhaustive]
56pub enum Route {
57    /// `GET /fapi/v1/ping`.
58    Ping,
59    /// `GET /fapi/v1/time`.
60    Time,
61    /// `GET /fapi/v1/exchangeInfo`.
62    ExchangeInfo,
63    /// `GET /fapi/v1/fundingInfo`.
64    FundingInfo,
65    /// `GET /fapi/v1/ticker/24hr`: one symbol, or every symbol when `all`.
66    Ticker24h {
67        /// No `symbol` parameter.
68        all: bool,
69    },
70    /// `GET /fapi/v1/premiumIndex`: one symbol, or every symbol when `all`.
71    PremiumIndex {
72        /// No `symbol` parameter.
73        all: bool,
74    },
75    /// `GET /fapi/v1/klines`.
76    Klines {
77        /// The `limit` parameter, if sent.
78        limit: Option<u32>,
79    },
80    /// `GET /fapi/v1/fundingRate`.
81    FundingRate,
82    /// `GET /fapi/v1/openInterest`.
83    OpenInterest,
84    /// `GET /fapi/v1/aggTrades`.
85    AggTrades,
86    /// `GET /fapi/v1/depth`.
87    Depth {
88        /// The `limit` parameter, if sent.
89        limit: Option<DepthLimit>,
90    },
91}
92
93impl Route {
94    /// The route's path on `fapi.binance.com`.
95    pub fn path(self) -> &'static str {
96        match self {
97            Self::Ping => "/fapi/v1/ping",
98            Self::Time => "/fapi/v1/time",
99            Self::ExchangeInfo => "/fapi/v1/exchangeInfo",
100            Self::FundingInfo => "/fapi/v1/fundingInfo",
101            Self::Ticker24h { .. } => "/fapi/v1/ticker/24hr",
102            Self::PremiumIndex { .. } => "/fapi/v1/premiumIndex",
103            Self::Klines { .. } => "/fapi/v1/klines",
104            Self::FundingRate => "/fapi/v1/fundingRate",
105            Self::OpenInterest => "/fapi/v1/openInterest",
106            Self::AggTrades => "/fapi/v1/aggTrades",
107            Self::Depth { .. } => "/fapi/v1/depth",
108        }
109    }
110
111    /// What the request costs, as measured from `X-MBX-USED-WEIGHT-1M` deltas
112    /// on 2026-10-07.
113    ///
114    /// Two rows differ from Binance's page. `klines` bands are inclusive at the
115    /// top (a limit of 100 costs 1, 500 costs 2, 1000 costs 5), and omitting
116    /// `limit` costs 5 although it returns 500 rows. `depth` without `limit`
117    /// costs 1 and returns 500 levels, where an explicit 500 costs 10.
118    pub fn cost(self) -> Cost {
119        match self {
120            Self::Ping | Self::Time | Self::ExchangeInfo | Self::OpenInterest => Cost::Weight(1),
121            Self::FundingInfo | Self::FundingRate => Cost::Funding,
122            Self::Ticker24h { all } => Cost::Weight(if all { 40 } else { 1 }),
123            Self::PremiumIndex { all } => Cost::Weight(if all { 10 } else { 1 }),
124            Self::Klines { limit: None } => Cost::Weight(5),
125            Self::Klines { limit: Some(limit) } => Cost::Weight(match limit {
126                0..=100 => 1,
127                101..=500 => 2,
128                501..=1000 => 5,
129                _ => 10,
130            }),
131            Self::AggTrades => Cost::Weight(20),
132            Self::Depth { limit: None } => Cost::Weight(1),
133            Self::Depth { limit: Some(limit) } => Cost::Weight(match limit {
134                DepthLimit::Five | DepthLimit::Ten | DepthLimit::Twenty | DepthLimit::Fifty => 2,
135                DepthLimit::Hundred => 5,
136                DepthLimit::FiveHundred => 10,
137                DepthLimit::Thousand => 20,
138            }),
139        }
140    }
141}
142
143/// The request-weight budget of one IP, shared by every client that clones it.
144///
145/// Binance limits weight per IP, not per client, so every
146/// [`Usdm`](crate::Usdm) in a process should be built with one budget
147/// ([`UsdmBuilder::weight_budget`](crate::UsdmBuilder::weight_budget)).
148///
149/// - **Window.** The UTC clock minute: the server's count fell to 1 just after
150///   each minute boundary on 2026-10-07, where a sliding window would have
151///   kept counting.
152/// - **Budget.** The published 2400 less a tenth, 2160. A request waits while
153///   its weight would take the minute past that.
154/// - **Server count.** Every response's `X-MBX-USED-WEIGHT-1M` raises the
155///   minute's count to at least what the server reports, because other
156///   processes on the same IP spend the same budget. The count never falls
157///   within a minute, and a response is applied only to the minute its request
158///   was charged in.
159/// - **Funding.** `fundingRate` and `fundingInfo` carry no weight and share a
160///   documented 500 requests per 5 minutes, paced here at 450 with a depth of
161///   one, so they never burst, not even when a cooldown ends.
162/// - **Cooldown.** A `429` or `418` holds every request, on both limits, for
163///   the delay the server asked for. A cooldown is only ever extended.
164/// - **In flight.** A response's header is applied only to the minute its
165///   request was charged in, so a request in flight across a minute boundary is
166///   counted by the server in a minute this budget no longer sees. The reserve
167///   of 240 absorbs that while the weight in flight stays under it: one `Usdm`
168///   at its default of 4 concurrent requests has at most 160 in flight (4 × the
169///   heaviest route, 40). A higher `max_concurrent`, or several clients sharing
170///   one budget, can pass it.
171#[derive(Debug, Clone)]
172pub struct WeightBudget {
173    inner: Arc<Inner>,
174}
175
176#[derive(Debug)]
177struct Inner {
178    per_minute: u32,
179    funding_interval: Duration,
180    clock: Clock,
181    state: Mutex<State>,
182}
183
184#[derive(Debug)]
185struct State {
186    /// The UTC minute (milliseconds / 60 000) that `used` counts.
187    minute: u64,
188    used: u32,
189    cooldown_until: Option<Instant>,
190    next_funding: Option<Instant>,
191}
192
193/// The minute a request was charged in, so its response's header is applied to
194/// that minute and no other. `None` for the funding routes, which report no
195/// weight.
196#[derive(Debug, Clone, Copy, PartialEq, Eq)]
197pub(crate) struct Charge {
198    minute: Option<u64>,
199}
200
201/// Where the current UTC minute comes from.
202#[derive(Debug, Clone)]
203enum Clock {
204    /// The system clock, which is what Binance's minute follows.
205    System,
206    /// Wall time derived from tokio's clock, so a paused-time test controls it.
207    #[cfg(test)]
208    Tokio { start: Instant, start_ms: u64 },
209    /// A clock a test sets by hand, to step it backwards.
210    #[cfg(test)]
211    Manual(Arc<std::sync::atomic::AtomicU64>),
212}
213
214impl Clock {
215    fn now_ms(&self) -> u64 {
216        match self {
217            Self::System => SystemTime::now()
218                .duration_since(UNIX_EPOCH)
219                .map_or(0, |since| since.as_millis() as u64),
220            #[cfg(test)]
221            Self::Tokio { start, start_ms } => *start_ms + start.elapsed().as_millis() as u64,
222            #[cfg(test)]
223            Self::Manual(ms) => ms.load(std::sync::atomic::Ordering::SeqCst),
224        }
225    }
226}
227
228impl Default for WeightBudget {
229    fn default() -> Self {
230        Self::new()
231    }
232}
233
234impl WeightBudget {
235    /// A budget for Binance's published limits, less the reserve.
236    pub fn new() -> Self {
237        Self::with_clock(
238            PUBLISHED_WEIGHT_PER_MINUTE,
239            PUBLISHED_FUNDING_PER_FIVE_MINUTES,
240            Clock::System,
241        )
242    }
243
244    fn with_clock(weight_per_minute: u32, funding_per_five_minutes: u32, clock: Clock) -> Self {
245        // One slot of the funding target is the bucket's depth; the rest is
246        // paced, so no five-minute window admits more than the target.
247        let funding_slots = after_reserve(funding_per_five_minutes).max(2) - 1;
248        Self {
249            inner: Arc::new(Inner {
250                per_minute: after_reserve(weight_per_minute),
251                funding_interval: FUNDING_PERIOD / funding_slots,
252                clock,
253                state: Mutex::new(State {
254                    minute: 0,
255                    used: 0,
256                    cooldown_until: None,
257                    next_funding: None,
258                }),
259            }),
260        }
261    }
262
263    /// Whether two handles charge one budget.
264    #[cfg(test)]
265    pub(crate) fn is_shared_with(&self, other: &Self) -> bool {
266        Arc::ptr_eq(&self.inner, &other.inner)
267    }
268
269    /// The most weight this budget spends in one minute.
270    pub fn per_minute(&self) -> u32 {
271        self.inner.per_minute
272    }
273
274    /// Weight charged or reported in the current UTC minute.
275    pub fn used(&self) -> u32 {
276        let minute = self.inner.clock.now_ms() / MINUTE_MS;
277        let state = self.lock();
278        if minute > state.minute {
279            0
280        } else {
281            state.used
282        }
283    }
284
285    /// Waits until the request can go, then charges it.
286    pub(crate) async fn acquire(&self, cost: Cost) -> Charge {
287        match cost {
288            Cost::Funding => loop {
289                // Wait out a cooldown before taking a slot: a slot taken first
290                // and slept on through a cooldown is in the past when it ends,
291                // and every request parked that way would go at once.
292                self.await_cooldown().await;
293                let slot = self.reserve_funding_slot();
294                tokio::time::sleep_until(slot).await;
295                if !self.in_cooldown() {
296                    return Charge { minute: None };
297                }
298                // A cooldown began while this request waited for its slot: the
299                // slot is spent, and the request queues again behind it.
300            },
301            Cost::Weight(weight) => loop {
302                self.await_cooldown().await;
303                match self.try_charge(weight) {
304                    Ok(charge) => return charge,
305                    Err(until_next_minute) => tokio::time::sleep(until_next_minute).await,
306                }
307            },
308        }
309    }
310
311    /// Charges `weight` to the current minute, or says how long until the next.
312    ///
313    /// An empty minute admits any weight, so a request heavier than the whole
314    /// budget is sent alone rather than held forever.
315    fn try_charge(&self, weight: u32) -> Result<Charge, Duration> {
316        let now_ms = self.inner.clock.now_ms();
317        let mut state = self.lock();
318        // The minute only moves forward. A read that lands in an earlier minute,
319        // from a stale read racing a boundary or a clock stepped back, charges
320        // the newest minute seen instead of resetting its count.
321        let minute = now_ms / MINUTE_MS;
322        if minute > state.minute {
323            state.minute = minute;
324            state.used = 0;
325        }
326        if state.used == 0 || state.used.saturating_add(weight) <= self.inner.per_minute {
327            state.used = state.used.saturating_add(weight);
328            Ok(Charge {
329                minute: Some(state.minute),
330            })
331        } else {
332            Err(Duration::from_millis(MINUTE_MS - now_ms % MINUTE_MS))
333        }
334    }
335
336    /// Raises the minute's count to what the server reported for it.
337    pub(crate) fn record_used(&self, charge: Charge, used: u32) {
338        let Some(minute) = charge.minute else { return };
339        let mut state = self.lock();
340        if state.minute == minute && used > state.used {
341            state.used = used;
342        }
343    }
344
345    /// Holds every request for `delay`. Extends a cooldown in force, never
346    /// shortens one: concurrent requests see the same `429` milliseconds
347    /// apart, and taking the latest delay would release them all early.
348    pub(crate) fn begin_cooldown(&self, delay: Duration) {
349        let until = Instant::now() + delay.min(MAX_COOLDOWN);
350        let mut state = self.lock();
351        if state.cooldown_until.is_none_or(|current| until > current) {
352            state.cooldown_until = Some(until);
353        }
354    }
355
356    /// Holds every request until the next UTC minute, when the weight window
357    /// resets: what a `429` with no `Retry-After` and no retry left calls for.
358    /// Returns the hold, for the log.
359    pub(crate) fn hold_until_next_minute(&self) -> Duration {
360        let now_ms = self.inner.clock.now_ms();
361        let hold = Duration::from_millis(MINUTE_MS - now_ms % MINUTE_MS);
362        self.begin_cooldown(hold);
363        hold
364    }
365
366    async fn await_cooldown(&self) {
367        loop {
368            // Copy the deadline out so the guard is not held across the await.
369            let until = self.lock().cooldown_until;
370            match until {
371                // Loop rather than return: another response may extend the
372                // cooldown while this one sleeps.
373                Some(until) if until > Instant::now() => tokio::time::sleep_until(until).await,
374                _ => return,
375            }
376        }
377    }
378
379    fn in_cooldown(&self) -> bool {
380        self.lock()
381            .cooldown_until
382            .is_some_and(|until| until > Instant::now())
383    }
384
385    fn reserve_funding_slot(&self) -> Instant {
386        let now = Instant::now();
387        let mut state = self.lock();
388        let slot = state.next_funding.map_or(now, |next| next.max(now));
389        state.next_funding = Some(slot + self.inner.funding_interval);
390        slot
391    }
392
393    /// A poison-tolerant lock: a panic elsewhere must not turn the budget into
394    /// a permanent outage.
395    fn lock(&self) -> MutexGuard<'_, State> {
396        self.inner
397            .state
398            .lock()
399            .unwrap_or_else(|poisoned| poisoned.into_inner())
400    }
401}
402
403#[cfg(test)]
404mod tests {
405    use super::*;
406    use std::sync::atomic::{AtomicU64, Ordering};
407
408    /// 2026-10-07 08:35:00 UTC, a minute boundary.
409    const ON_A_MINUTE: u64 = 1_791_362_100_000;
410
411    fn budget(weight_per_minute: u32, start_ms: u64) -> WeightBudget {
412        WeightBudget::with_clock(
413            weight_per_minute,
414            PUBLISHED_FUNDING_PER_FIVE_MINUTES,
415            Clock::Tokio {
416                start: Instant::now(),
417                start_ms,
418            },
419        )
420    }
421
422    fn manual_budget(start_ms: u64) -> (WeightBudget, Arc<AtomicU64>) {
423        let clock = Arc::new(AtomicU64::new(start_ms));
424        let budget = WeightBudget::with_clock(
425            PUBLISHED_WEIGHT_PER_MINUTE,
426            PUBLISHED_FUNDING_PER_FIVE_MINUTES,
427            Clock::Manual(Arc::clone(&clock)),
428        );
429        (budget, clock)
430    }
431
432    #[test]
433    fn documented_weights() {
434        use DepthLimit::*;
435        let w = Cost::Weight;
436        let table = [
437            (Route::Ping, w(1)),
438            (Route::Time, w(1)),
439            (Route::ExchangeInfo, w(1)),
440            (Route::FundingInfo, Cost::Funding),
441            (Route::Ticker24h { all: false }, w(1)),
442            (Route::Ticker24h { all: true }, w(40)),
443            (Route::PremiumIndex { all: false }, w(1)),
444            (Route::PremiumIndex { all: true }, w(10)),
445            (Route::Klines { limit: None }, w(5)),
446            (Route::Klines { limit: Some(1) }, w(1)),
447            (Route::Klines { limit: Some(100) }, w(1)),
448            (Route::Klines { limit: Some(101) }, w(2)),
449            (Route::Klines { limit: Some(500) }, w(2)),
450            (Route::Klines { limit: Some(501) }, w(5)),
451            (Route::Klines { limit: Some(1000) }, w(5)),
452            (Route::Klines { limit: Some(1001) }, w(10)),
453            (Route::Klines { limit: Some(1500) }, w(10)),
454            (Route::FundingRate, Cost::Funding),
455            (Route::OpenInterest, w(1)),
456            (Route::AggTrades, w(20)),
457            (Route::Depth { limit: None }, w(1)),
458            (Route::Depth { limit: Some(Five) }, w(2)),
459            (Route::Depth { limit: Some(Ten) }, w(2)),
460            (
461                Route::Depth {
462                    limit: Some(Twenty),
463                },
464                w(2),
465            ),
466            (Route::Depth { limit: Some(Fifty) }, w(2)),
467            (
468                Route::Depth {
469                    limit: Some(Hundred),
470                },
471                w(5),
472            ),
473            (
474                Route::Depth {
475                    limit: Some(FiveHundred),
476                },
477                w(10),
478            ),
479            (
480                Route::Depth {
481                    limit: Some(Thousand),
482                },
483                w(20),
484            ),
485        ];
486        for (route, cost) in table {
487            assert_eq!(route.cost(), cost, "{route:?}");
488        }
489    }
490
491    #[test]
492    fn the_budget_aims_a_tenth_below_the_published_limit() {
493        assert_eq!(WeightBudget::new().per_minute(), 2160);
494    }
495
496    #[tokio::test(start_paused = true)]
497    async fn a_charge_past_the_budget_waits_for_the_next_minute() {
498        let budget = budget(PUBLISHED_WEIGHT_PER_MINUTE, ON_A_MINUTE + 15_000);
499        for _ in 0..54 {
500            budget.acquire(Cost::Weight(40)).await;
501        }
502        assert_eq!(budget.used(), 2160);
503
504        let start = Instant::now();
505        budget.acquire(Cost::Weight(1)).await;
506        assert_eq!(
507            start.elapsed(),
508            Duration::from_secs(45),
509            "held to the boundary"
510        );
511        assert_eq!(budget.used(), 1, "the new minute starts from zero");
512    }
513
514    #[tokio::test(start_paused = true)]
515    async fn a_header_raises_the_count_and_never_lowers_it() {
516        let budget = budget(PUBLISHED_WEIGHT_PER_MINUTE, ON_A_MINUTE);
517        let charge = budget.acquire(Cost::Weight(1)).await;
518        budget.record_used(charge, 2000);
519        assert_eq!(budget.used(), 2000, "another process on this IP spent 1999");
520        budget.record_used(charge, 3);
521        assert_eq!(budget.used(), 2000);
522
523        let start = Instant::now();
524        budget.acquire(Cost::Weight(200)).await;
525        assert_eq!(
526            start.elapsed(),
527            Duration::from_secs(60),
528            "2000 + 200 passes 2160"
529        );
530    }
531
532    #[tokio::test(start_paused = true)]
533    async fn a_header_is_applied_only_to_the_minute_its_request_was_charged_in() {
534        let budget = budget(PUBLISHED_WEIGHT_PER_MINUTE, ON_A_MINUTE + 59_900);
535        let in_flight = budget.acquire(Cost::Weight(5)).await;
536        tokio::time::advance(Duration::from_millis(200)).await;
537        budget.acquire(Cost::Weight(5)).await;
538        // The response to the first request arrives in the new minute carrying
539        // the old minute's count. Applying it would hold the new minute for 60 s.
540        budget.record_used(in_flight, 2150);
541        assert_eq!(budget.used(), 5);
542    }
543
544    #[tokio::test(start_paused = true)]
545    async fn a_request_heavier_than_the_budget_is_sent_alone() {
546        let budget = budget(20, ON_A_MINUTE);
547        let start = Instant::now();
548        budget.acquire(Cost::Weight(40)).await;
549        assert_eq!(start.elapsed(), Duration::ZERO);
550        budget.acquire(Cost::Weight(1)).await;
551        assert_eq!(start.elapsed(), Duration::from_secs(60));
552    }
553
554    #[tokio::test(start_paused = true)]
555    async fn a_cooldown_holds_both_limits_and_is_only_extended() {
556        let budget = budget(PUBLISHED_WEIGHT_PER_MINUTE, ON_A_MINUTE);
557        budget.begin_cooldown(Duration::from_secs(10));
558        budget.begin_cooldown(Duration::from_secs(2));
559
560        let start = Instant::now();
561        budget.acquire(Cost::Weight(1)).await;
562        assert_eq!(start.elapsed(), Duration::from_secs(10));
563
564        budget.begin_cooldown(Duration::from_secs(3));
565        let start = Instant::now();
566        budget.acquire(Cost::Funding).await;
567        assert_eq!(start.elapsed(), Duration::from_secs(3));
568    }
569
570    #[tokio::test(start_paused = true)]
571    async fn a_cooldown_is_clamped_to_the_longest_documented_ban() {
572        let budget = budget(PUBLISHED_WEIGHT_PER_MINUTE, ON_A_MINUTE);
573        budget.begin_cooldown(Duration::MAX);
574        let start = Instant::now();
575        budget.acquire(Cost::Weight(1)).await;
576        assert_eq!(start.elapsed(), MAX_COOLDOWN);
577    }
578
579    #[tokio::test(start_paused = true)]
580    async fn a_hold_lasts_until_the_next_minute_for_both_limits() {
581        for cost in [Cost::Weight(1), Cost::Funding] {
582            let budget = budget(PUBLISHED_WEIGHT_PER_MINUTE, ON_A_MINUTE + 15_000);
583            assert_eq!(budget.hold_until_next_minute(), Duration::from_secs(45));
584            let start = Instant::now();
585            budget.acquire(cost).await;
586            assert_eq!(start.elapsed(), Duration::from_secs(45), "{cost:?}");
587        }
588    }
589
590    #[tokio::test(start_paused = true)]
591    async fn the_funding_bucket_admits_450_per_five_minutes() {
592        let budget = budget(PUBLISHED_WEIGHT_PER_MINUTE, ON_A_MINUTE);
593        let start = Instant::now();
594        let mut admitted = 0;
595        while start.elapsed() <= FUNDING_PERIOD {
596            budget.acquire(Cost::Funding).await;
597            if start.elapsed() <= FUNDING_PERIOD {
598                admitted += 1;
599            }
600        }
601        assert_eq!(admitted, 450);
602    }
603
604    #[tokio::test(start_paused = true)]
605    async fn the_funding_bucket_never_bursts() {
606        let budget = budget(PUBLISHED_WEIGHT_PER_MINUTE, ON_A_MINUTE);
607        let start = Instant::now();
608        budget.acquire(Cost::Funding).await;
609        budget.acquire(Cost::Funding).await;
610        assert!(
611            start.elapsed() >= Duration::from_millis(668),
612            "{:?}",
613            start.elapsed()
614        );
615    }
616
617    #[tokio::test(start_paused = true)]
618    async fn a_cooldown_extended_mid_wait_is_waited_out_in_full() {
619        let budget = budget(PUBLISHED_WEIGHT_PER_MINUTE, ON_A_MINUTE);
620        budget.begin_cooldown(Duration::from_secs(2));
621        let extender = budget.clone();
622        tokio::spawn(async move {
623            tokio::time::sleep(Duration::from_secs(1)).await;
624            extender.begin_cooldown(Duration::from_secs(10));
625        });
626        let start = Instant::now();
627        budget.acquire(Cost::Weight(1)).await;
628        assert_eq!(start.elapsed(), Duration::from_secs(11));
629    }
630
631    #[tokio::test(start_paused = true)]
632    async fn funding_requests_parked_by_a_cooldown_are_paced_when_it_ends() {
633        let budget = budget(PUBLISHED_WEIGHT_PER_MINUTE, ON_A_MINUTE);
634        budget.begin_cooldown(Duration::from_secs(60));
635        let start = Instant::now();
636        let tasks: Vec<_> = (0..10)
637            .map(|_| {
638                let budget = budget.clone();
639                tokio::spawn(async move {
640                    budget.acquire(Cost::Funding).await;
641                    start.elapsed()
642                })
643            })
644            .collect();
645        let mut released = Vec::new();
646        for task in tasks {
647            released.push(task.await.unwrap());
648        }
649        released.sort();
650        assert_eq!(released[0], Duration::from_secs(60));
651        for pair in released.windows(2) {
652            assert!(
653                pair[1] - pair[0] >= Duration::from_millis(668),
654                "released together: {released:?}"
655            );
656        }
657    }
658
659    #[tokio::test(start_paused = true)]
660    async fn the_counted_minute_never_moves_backwards() {
661        let (budget, clock) = manual_budget(ON_A_MINUTE + 1_000);
662        budget.acquire(Cost::Weight(2000)).await;
663        // The clock steps back into the previous minute, as an NTP step or a
664        // read racing the boundary can. Resetting the count there would admit
665        // another 2000 in what the server counts as the same minute.
666        clock.store(ON_A_MINUTE - 1_000, Ordering::SeqCst);
667        assert_eq!(budget.used(), 2000);
668        let second =
669            tokio::time::timeout(Duration::from_secs(5), budget.acquire(Cost::Weight(2000))).await;
670        assert!(second.is_err(), "charged into a minute already counted");
671        // A request charged from the stepped-back clock is counted in the
672        // newest minute, so its response's header still applies.
673        let stale = tokio::time::timeout(Duration::from_secs(5), budget.acquire(Cost::Weight(1)))
674            .await
675            .expect("2001 fits under 2160");
676        budget.record_used(stale, 2100);
677        assert_eq!(budget.used(), 2100);
678    }
679
680    #[tokio::test(start_paused = true)]
681    async fn a_cooldown_begun_during_the_slot_wait_holds_the_funding_request() {
682        // Sent anyway, a request during a 418 ban can lengthen the ban.
683        let budget = budget(PUBLISHED_WEIGHT_PER_MINUTE, ON_A_MINUTE);
684        budget.acquire(Cost::Funding).await; // slot 0; the next is ~668 ms away
685        let start = Instant::now();
686        let waiter = {
687            let budget = budget.clone();
688            tokio::spawn(async move {
689                budget.acquire(Cost::Funding).await;
690                start.elapsed()
691            })
692        };
693        tokio::time::sleep(Duration::from_millis(100)).await;
694        budget.begin_cooldown(Duration::from_secs(10));
695        assert_eq!(waiter.await.unwrap(), Duration::from_millis(10_100));
696    }
697}