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

polyoxide_binance/usdm/
weight.rs
