nautilus_derive/common/
rate_limit.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Derive v3 request budgets and per-wallet/instrument order pacing.
17//!
18//! Request budgets reset in five-second fixed windows. Instrument buckets
19//! refill continuously and hold reserved capacity until a signed request is
20//! dispatched. HTTP and WebSocket clients on the same API host share wallet
21//! state across session keys and subaccounts. Public requests share a separate
22//! process-local per-host budget, conservatively assuming one source IP.
23//!
24//! Published defaults are a reference snapshot. Read deployed request budgets
25//! with `public/getRateLimits`; it does not report instrument allowances.
26//! Client windows cannot observe the venue's phase, and transport queues can
27//! delay departure, so definitive venue throttling remains possible.
28//!
29//! See <https://docs.derive.xyz/rate-limits>.
30
31use std::{
32    collections::HashMap,
33    sync::{Arc, OnceLock},
34    time::Duration,
35};
36
37use nautilus_network::{
38    http::Url,
39    ratelimiter::clock::{Clock, MonotonicClock, Reference},
40};
41use parking_lot::Mutex;
42use ustr::Ustr;
43
44/// Identifier for the matching request budget.
45pub const DERIVE_MATCHING_RATE_KEY: &str = "derive:matching";
46/// Identifier for the non-matching request budget.
47pub const DERIVE_NON_MATCHING_RATE_KEY: &str = "derive:non-matching";
48/// Identifier for the `private/cancel_all` request budget.
49pub const DERIVE_CANCEL_ALL_RATE_KEY: &str = "derive:cancel-all";
50/// Identifier for the `private/cancel_by_label` request budget.
51pub const DERIVE_CANCEL_BY_LABEL_RATE_KEY: &str = "derive:cancel-by-label";
52/// Reference wallet matching requests per second.
53pub const DERIVE_DEFAULT_MATCHING_TPS: u32 = 1;
54/// Reference per-instrument refill rate for perpetuals and spot.
55pub const DERIVE_DEFAULT_PER_INSTRUMENT_MATCHING_TPS: u32 = 1;
56/// Reference public request allowance per IP, in requests per second.
57pub const DERIVE_NON_MATCHING_TPS: u32 = 5;
58/// Reference authenticated non-matching requests per second.
59pub const DERIVE_WEBSOCKET_NON_MATCHING_TPS: u32 = DERIVE_NON_MATCHING_TPS;
60/// Reference `private/cancel_all` requests per second.
61pub const DERIVE_CANCEL_ALL_TPS: u32 = 1;
62/// Adapter `private/cancel_by_label` endpoint allowance, retaining the
63/// integrator reference's 10 TPS for unscoped calls.
64pub const DERIVE_CANCEL_BY_LABEL_TPS: u32 = 10;
65/// Request-budget fixed-window duration in seconds.
66pub const DERIVE_RATE_WINDOW_SECS: u64 = 5;
67/// Instrument capacity multiplier and requests per fixed window per TPS.
68pub const DERIVE_RATE_BURST_MULTIPLIER: u32 = 5;
69
70const DERIVE_DEFAULT_OPTION_MATCHING_TPS: u32 = 1;
71const TOKEN_UNITS: u64 = 1_000_000_000;
72const RATE_WINDOW_NANOS: u64 = DERIVE_RATE_WINDOW_SECS * TOKEN_UNITS;
73
74// Strong references preserve budget debt through client recreation and reconnects
75static RATE_STATES: OnceLock<Mutex<HashMap<RateScope, Arc<RateState<MonotonicClock>>>>> =
76    OnceLock::new();
77
78#[derive(Debug)]
79pub(crate) struct RequestLimiter<C: Clock> {
80    state: Arc<RateState<C>>,
81    public: Arc<RateState<C>>,
82}
83
84impl<C: Clock> RequestLimiter<C> {
85    pub(crate) fn new(limits: RateLimits, clock: C) -> Self {
86        let state = Arc::new(RateState::new(limits, clock));
87
88        Self {
89            public: Arc::clone(&state),
90            state,
91        }
92    }
93
94    pub(crate) fn try_class_ready(
95        &self,
96        class: RateClass,
97        instrument: Option<&Ustr>,
98    ) -> Result<u64, Duration> {
99        self.state.try_request(class, instrument, false)
100    }
101
102    pub(crate) async fn await_class_ready(
103        &self,
104        class: RateClass,
105        instrument: Option<&Ustr>,
106    ) -> u64 {
107        loop {
108            match self.try_class_ready(class, instrument) {
109                Ok(window) => return window,
110                Err(wait) => self.state.clock.sleep(wait).await,
111            }
112        }
113    }
114
115    pub(crate) async fn await_public_ready(&self) {
116        loop {
117            match self.public.try_request(RateClass::NonMatching, None, false) {
118                Ok(_) => return,
119                Err(wait) => self.public.clock.sleep(wait).await,
120            }
121        }
122    }
123
124    pub(crate) async fn reserve_matching_ready(&self, instrument: &Ustr) -> RateReservation<C> {
125        loop {
126            match self
127                .state
128                .try_request(RateClass::Matching, Some(instrument), true)
129            {
130                Ok(window) => {
131                    return RateReservation {
132                        state: Arc::clone(&self.state),
133                        instrument: *instrument,
134                        window,
135                        committed: false,
136                    };
137                }
138                Err(wait) => self.state.clock.sleep(wait).await,
139            }
140        }
141    }
142}
143
144impl RequestLimiter<MonotonicClock> {
145    pub(crate) fn shared(
146        url: &str,
147        wallet: Option<&str>,
148        matching: Option<u32>,
149        instrument: Option<u32>,
150    ) -> Self {
151        let authority = Url::parse(url).map_or_else(
152            |_| url.to_ascii_lowercase(),
153            |url| {
154                format!(
155                    "{}:{}",
156                    url.host_str().unwrap_or_default(),
157                    url.port_or_known_default().unwrap_or_default()
158                )
159            },
160        );
161
162        let public_scope = RateScope {
163            authority,
164            wallet: None,
165        };
166
167        let mut states = RATE_STATES
168            .get_or_init(|| Mutex::new(HashMap::new()))
169            .lock();
170        let public = Arc::clone(
171            states
172                .entry(public_scope.clone())
173                .or_insert_with(|| Self::new(RateLimits::new(None, None), MonotonicClock).state),
174        );
175
176        let state = match wallet {
177            Some(wallet) => {
178                let scope = RateScope {
179                    wallet: Some(wallet.to_ascii_lowercase()),
180                    ..public_scope
181                };
182
183                Arc::clone(states.entry(scope).or_insert_with(|| {
184                    Self::new(RateLimits::new(None, None), MonotonicClock).state
185                }))
186            }
187            None => Arc::clone(&public),
188        };
189
190        state.configure(matching, instrument);
191        Self { state, public }
192    }
193}
194
195pub(crate) type DeriveRateLimiter = RequestLimiter<MonotonicClock>;
196
197#[derive(Debug)]
198pub(crate) struct RateReservation<C: Clock> {
199    state: Arc<RateState<C>>,
200    instrument: Ustr,
201    window: u64,
202    committed: bool,
203}
204
205impl<C: Clock> RateReservation<C> {
206    // Commit never sleeps after the caller has generated its nonce and signature
207    pub(crate) fn commit(mut self) -> Result<(), Duration> {
208        let mut ledger = self.state.ledger.lock();
209        let now = self.state.elapsed_nanos();
210        let window = now / RATE_WINDOW_NANOS;
211        if window != self.window {
212            ledger.check_requests(RateClass::Matching, window, now)?;
213            ledger.consume_requests(RateClass::Matching, window);
214        }
215
216        let tps = ledger.limits.instrument_tps(&self.instrument);
217        let bucket = ledger
218            .instruments
219            .get_mut(&self.instrument)
220            .expect("reserved instrument exists");
221        bucket.refill(now, tps);
222        bucket.pending -= 1;
223        self.committed = true;
224        Ok(())
225    }
226}
227
228impl<C: Clock> Drop for RateReservation<C> {
229    fn drop(&mut self) {
230        if self.committed {
231            return;
232        }
233
234        let mut ledger = self.state.ledger.lock();
235        let now = self.state.elapsed_nanos();
236        let tps = ledger.limits.instrument_tps(&self.instrument);
237        let bucket = ledger
238            .instruments
239            .get_mut(&self.instrument)
240            .expect("reserved instrument exists");
241        bucket.refill(now, tps);
242        bucket.pending -= 1;
243        bucket.tokens = bucket
244            .tokens
245            .saturating_add(TOKEN_UNITS)
246            .min(bucket.capacity(tps));
247
248        for key in request_buckets(RateClass::Matching) {
249            let cell = &mut ledger.requests[*key as usize];
250            if cell.window == self.window {
251                cell.consumed -= 1;
252            }
253        }
254    }
255}
256
257#[derive(Debug, Clone, Copy, PartialEq, Eq)]
258pub(crate) enum RateClass {
259    NonMatching,
260    Matching,
261    CancelAll,
262    CancelByLabel,
263}
264
265#[derive(Debug, Clone, Copy)]
266pub(crate) struct RateLimits {
267    matching: u32,
268    instrument: u32,
269    option: u32,
270}
271
272impl RateLimits {
273    pub(crate) fn new(matching: Option<u32>, instrument: Option<u32>) -> Self {
274        Self {
275            matching: resolve_tps(matching, DERIVE_DEFAULT_MATCHING_TPS),
276            instrument: resolve_tps(instrument, DERIVE_DEFAULT_PER_INSTRUMENT_MATCHING_TPS),
277            option: resolve_tps(instrument, DERIVE_DEFAULT_OPTION_MATCHING_TPS),
278        }
279    }
280
281    fn instrument_tps(&self, instrument: &Ustr) -> u32 {
282        match instrument.as_str().split('-').nth(3) {
283            Some("C" | "P") => self.option,
284            _ => self.instrument,
285        }
286    }
287
288    fn request_limit(&self, bucket: RequestBucket) -> u32 {
289        let tps = match bucket {
290            RequestBucket::NonMatching => DERIVE_WEBSOCKET_NON_MATCHING_TPS,
291            RequestBucket::Matching => self.matching,
292            RequestBucket::CancelAll => DERIVE_CANCEL_ALL_TPS,
293            RequestBucket::CancelByLabel => DERIVE_CANCEL_BY_LABEL_TPS,
294        };
295
296        window_limit(tps)
297    }
298}
299
300#[derive(Debug)]
301struct RateState<C: Clock> {
302    clock: C,
303    start: C::Instant,
304    ledger: Mutex<RateLedger>,
305}
306
307impl<C: Clock> RateState<C> {
308    fn new(limits: RateLimits, clock: C) -> Self {
309        let start = clock.now();
310
311        Self {
312            clock,
313            start,
314            ledger: Mutex::new(RateLedger {
315                limits,
316                matching_configured: None,
317                instrument_configured: None,
318                requests: std::array::from_fn(|_| RequestCell::default()),
319                instruments: HashMap::new(),
320            }),
321        }
322    }
323
324    fn configure(&self, matching: Option<u32>, instrument: Option<u32>) {
325        let mut ledger = self.ledger.lock();
326        if let Some(tps) = matching.filter(|&v| v > 0) {
327            match ledger.matching_configured {
328                None => {
329                    ledger.limits.matching = tps;
330                    ledger.matching_configured = Some(tps);
331                }
332                Some(existing) if existing != tps => {
333                    log::warn!(
334                        "Conflicting Derive wallet matching limits; retaining {existing} TPS"
335                    );
336                }
337                _ => {}
338            }
339        }
340
341        if let Some(tps) = instrument.filter(|&v| v > 0) {
342            match ledger.instrument_configured {
343                None => {
344                    let now = self.elapsed_nanos();
345                    let old = ledger.limits;
346                    for (name, bucket) in &mut ledger.instruments {
347                        let old_tps = old.instrument_tps(name);
348                        bucket.refill(now, old_tps);
349                        let extra = window_limit(tps).saturating_sub(window_limit(old_tps));
350                        bucket.tokens = bucket
351                            .tokens
352                            .saturating_add(u64::from(extra) * TOKEN_UNITS)
353                            .min(bucket.capacity(tps));
354                    }
355
356                    ledger.limits.instrument = tps;
357                    ledger.limits.option = tps;
358                    ledger.instrument_configured = Some(tps);
359                }
360                Some(existing) if existing != tps => {
361                    log::warn!("Conflicting Derive instrument limits; retaining {existing} TPS");
362                }
363                _ => {}
364            }
365        }
366    }
367
368    fn try_request(
369        &self,
370        class: RateClass,
371        instrument: Option<&Ustr>,
372        reserve: bool,
373    ) -> Result<u64, Duration> {
374        let mut ledger = self.ledger.lock();
375        let now = self.elapsed_nanos();
376        let window = now / RATE_WINDOW_NANOS;
377        ledger.check_requests(class, window, now)?;
378        if matches!(class, RateClass::Matching | RateClass::CancelByLabel)
379            && let Some(instrument) = instrument
380        {
381            let tps = ledger.limits.instrument_tps(instrument);
382
383            let bucket =
384                ledger
385                    .instruments
386                    .entry(*instrument)
387                    .or_insert_with(|| InstrumentBucket {
388                        tokens: u64::from(window_limit(tps)) * TOKEN_UNITS,
389                        updated: now,
390                        pending: 0,
391                    });
392
393            bucket.refill(now, tps);
394            if bucket.tokens < TOKEN_UNITS {
395                let deficit = TOKEN_UNITS - bucket.tokens;
396                return Err(Duration::from_nanos(deficit.div_ceil(u64::from(tps))));
397            }
398
399            bucket.tokens -= TOKEN_UNITS;
400            if reserve {
401                bucket.pending += 1;
402            }
403        }
404
405        ledger.consume_requests(class, window);
406        Ok(window)
407    }
408
409    fn elapsed_nanos(&self) -> u64 {
410        self.clock.now().duration_since(self.start).as_u64()
411    }
412}
413
414#[derive(Debug, Clone, PartialEq, Eq, Hash)]
415struct RateScope {
416    authority: String,
417    wallet: Option<String>,
418}
419
420#[derive(Debug)]
421struct RateLedger {
422    limits: RateLimits,
423    matching_configured: Option<u32>,
424    instrument_configured: Option<u32>,
425    requests: [RequestCell; 4],
426    instruments: HashMap<Ustr, InstrumentBucket>,
427}
428
429impl RateLedger {
430    fn check_requests(&self, class: RateClass, window: u64, now: u64) -> Result<(), Duration> {
431        for key in request_buckets(class) {
432            let cell = self.requests[*key as usize];
433            if cell.window == window && cell.consumed >= self.limits.request_limit(*key) {
434                return Err(Duration::from_nanos(
435                    RATE_WINDOW_NANOS - now % RATE_WINDOW_NANOS,
436                ));
437            }
438        }
439
440        Ok(())
441    }
442
443    fn consume_requests(&mut self, class: RateClass, window: u64) {
444        for key in request_buckets(class) {
445            let cell = &mut self.requests[*key as usize];
446            if cell.window != window {
447                *cell = RequestCell {
448                    window,
449                    consumed: 0,
450                };
451            }
452
453            cell.consumed += 1;
454        }
455    }
456}
457
458#[derive(Debug, Clone, Copy, Default)]
459struct RequestCell {
460    window: u64,
461    consumed: u32,
462}
463
464#[derive(Debug)]
465struct InstrumentBucket {
466    tokens: u64,
467    updated: u64,
468    pending: u32,
469}
470
471impl InstrumentBucket {
472    fn refill(&mut self, now: u64, tps: u32) {
473        let added = u128::from(now.saturating_sub(self.updated)) * u128::from(tps);
474        self.tokens =
475            u64::try_from((u128::from(self.tokens) + added).min(u128::from(self.capacity(tps))))
476                .expect("instrument credit fits u64");
477        self.updated = now;
478    }
479
480    fn capacity(&self, tps: u32) -> u64 {
481        u64::from(window_limit(tps).saturating_sub(self.pending)) * TOKEN_UNITS
482    }
483}
484
485pub(crate) fn rate_class_for_method(method: &str) -> RateClass {
486    // The legacy trigger caller remains outside the order-lifecycle migration
487    match method.trim_start_matches('/') {
488        "private/order"
489        | "private/replace"
490        | "private/cancel"
491        | "private/cancel_by_instrument"
492        | "private/cancel_by_nonce" => RateClass::Matching,
493        "private/cancel_all" => RateClass::CancelAll,
494        "private/cancel_by_label" => RateClass::CancelByLabel,
495        _ => RateClass::NonMatching,
496    }
497}
498
499#[derive(Debug, Clone, Copy)]
500enum RequestBucket {
501    NonMatching,
502    Matching,
503    CancelAll,
504    CancelByLabel,
505}
506
507fn request_buckets(class: RateClass) -> &'static [RequestBucket] {
508    match class {
509        RateClass::NonMatching => &[RequestBucket::NonMatching],
510        RateClass::Matching => &[RequestBucket::Matching],
511        RateClass::CancelAll => &[RequestBucket::CancelAll],
512        RateClass::CancelByLabel => &[RequestBucket::Matching, RequestBucket::CancelByLabel],
513    }
514}
515
516fn resolve_tps(configured: Option<u32>, default_tps: u32) -> u32 {
517    configured.filter(|&v| v > 0).unwrap_or(default_tps)
518}
519
520fn window_limit(tps: u32) -> u32 {
521    tps.saturating_mul(DERIVE_RATE_BURST_MULTIPLIER)
522}
523
524#[cfg(test)]
525mod tests {
526    use std::{sync::Barrier, thread};
527
528    use nautilus_network::ratelimiter::clock::FakeRelativeClock;
529    use rstest::rstest;
530
531    use super::*;
532
533    fn limiter(
534        matching: Option<u32>,
535        instrument: Option<u32>,
536    ) -> RequestLimiter<FakeRelativeClock> {
537        RequestLimiter::new(
538            RateLimits::new(matching, instrument),
539            FakeRelativeClock::default(),
540        )
541    }
542
543    #[rstest]
544    #[case("private/order", RateClass::Matching)]
545    #[case("/private/replace", RateClass::Matching)]
546    #[case("private/cancel", RateClass::Matching)]
547    #[case("private/cancel_by_instrument", RateClass::Matching)]
548    #[case("private/cancel_by_nonce", RateClass::Matching)]
549    #[case("private/cancel_trigger_order", RateClass::NonMatching)]
550    #[case("private/get_trigger_orders", RateClass::NonMatching)]
551    #[case("private/get_open_orders", RateClass::NonMatching)]
552    #[case("private/get_subaccount", RateClass::NonMatching)]
553    #[case("private/cancel_all", RateClass::CancelAll)]
554    #[case("private/cancel_by_label", RateClass::CancelByLabel)]
555    #[case("public/login", RateClass::NonMatching)]
556    #[case("subscribe", RateClass::NonMatching)]
557    fn method_classification(#[case] method: &str, #[case] expected: RateClass) {
558        assert_eq!(rate_class_for_method(method), expected);
559    }
560
561    #[rstest]
562    fn defaults_and_independent_overrides() {
563        let limits = RateLimits::new(None, None);
564        assert_eq!(limits.matching, 1);
565        assert_eq!(limits.instrument, 1);
566        assert_eq!(limits.option, 1);
567        assert_eq!(limits.request_limit(RequestBucket::NonMatching), 25);
568        assert_eq!(limits.request_limit(RequestBucket::CancelAll), 5);
569        assert_eq!(limits.request_limit(RequestBucket::CancelByLabel), 50);
570        let wallet = RateLimits::new(Some(500), None);
571        assert_eq!(wallet.matching, 500);
572        assert_eq!(wallet.instrument, 1);
573        assert_eq!(wallet.option, 1);
574        let instrument = RateLimits::new(None, Some(10));
575        assert_eq!(instrument.matching, 1);
576        assert_eq!(instrument.instrument, 10);
577        assert_eq!(instrument.option, 10);
578        let zero = RateLimits::new(Some(0), Some(0));
579        assert_eq!(zero.matching, limits.matching);
580        assert_eq!(zero.instrument, limits.instrument);
581        assert_eq!(zero.option, limits.option);
582    }
583
584    #[rstest]
585    fn request_windows_reset_only_at_boundary() {
586        let limiter = limiter(None, None);
587
588        for _ in 0..5 {
589            assert_eq!(limiter.try_class_ready(RateClass::Matching, None), Ok(0));
590        }
591
592        limiter.state.clock.advance(Duration::from_millis(4_999));
593        assert_eq!(
594            limiter.try_class_ready(RateClass::Matching, None),
595            Err(Duration::from_millis(1))
596        );
597        limiter.state.clock.advance(Duration::from_millis(1));
598
599        for _ in 0..5 {
600            assert_eq!(limiter.try_class_ready(RateClass::Matching, None), Ok(1));
601        }
602
603        assert_eq!(
604            limiter.try_class_ready(RateClass::Matching, None),
605            Err(Duration::from_secs(5))
606        );
607    }
608
609    #[rstest]
610    fn instrument_refills_continuously_with_fractional_credit() {
611        let limiter = limiter(Some(100), None);
612        let name = Ustr::from("ETH-PERP");
613
614        for _ in 0..5 {
615            assert_eq!(
616                limiter.try_class_ready(RateClass::Matching, Some(&name)),
617                Ok(0)
618            );
619        }
620
621        limiter.state.clock.advance(Duration::from_millis(999));
622        assert_eq!(
623            limiter.try_class_ready(RateClass::Matching, Some(&name)),
624            Err(Duration::from_millis(1))
625        );
626        limiter.state.clock.advance(Duration::from_millis(1));
627        assert_eq!(
628            limiter.try_class_ready(RateClass::Matching, Some(&name)),
629            Ok(0)
630        );
631        assert_eq!(
632            limiter.try_class_ready(RateClass::Matching, Some(&name)),
633            Err(Duration::from_secs(1))
634        );
635        limiter.state.clock.advance(Duration::from_secs(10));
636
637        for _ in 0..5 {
638            assert_eq!(
639                limiter.try_class_ready(RateClass::Matching, Some(&name)),
640                Ok(2)
641            );
642        }
643
644        assert_eq!(
645            limiter.try_class_ready(RateClass::Matching, Some(&name)),
646            Err(Duration::from_secs(1))
647        );
648    }
649
650    #[rstest]
651    fn instrument_has_no_window_boundary_double_burst() {
652        let limiter = limiter(Some(100), None);
653        let name = Ustr::from("ETH-PERP");
654        limiter.state.clock.advance(Duration::from_millis(4_999));
655
656        for _ in 0..5 {
657            assert_eq!(
658                limiter.try_class_ready(RateClass::Matching, Some(&name)),
659                Ok(0)
660            );
661        }
662
663        limiter.state.clock.advance(Duration::from_millis(1));
664        assert_eq!(
665            limiter.try_class_ready(RateClass::Matching, Some(&name)),
666            Err(Duration::from_millis(999))
667        );
668    }
669
670    #[rstest]
671    fn instrument_denial_does_not_consume_wallet_points() {
672        let limiter = limiter(Some(2), None);
673        let eth = Ustr::from("ETH-PERP");
674        let btc = Ustr::from("BTC-PERP");
675
676        for _ in 0..5 {
677            assert_eq!(
678                limiter.try_class_ready(RateClass::Matching, Some(&eth)),
679                Ok(0)
680            );
681        }
682
683        for _ in 0..10 {
684            assert_eq!(
685                limiter.try_class_ready(RateClass::Matching, Some(&eth)),
686                Err(Duration::from_secs(1))
687            );
688        }
689
690        for _ in 0..5 {
691            assert_eq!(
692                limiter.try_class_ready(RateClass::Matching, Some(&btc)),
693                Ok(0)
694            );
695        }
696
697        assert_eq!(
698            limiter.state.ledger.lock().requests[RequestBucket::Matching as usize].consumed,
699            10
700        );
701    }
702
703    #[rstest]
704    fn wallet_denial_does_not_consume_instrument_credit() {
705        let limiter = limiter(None, Some(2));
706        let name = Ustr::from("ETH-PERP");
707
708        for _ in 0..5 {
709            assert_eq!(
710                limiter.try_class_ready(RateClass::Matching, Some(&name)),
711                Ok(0)
712            );
713        }
714
715        assert_eq!(
716            limiter.try_class_ready(RateClass::Matching, Some(&name)),
717            Err(Duration::from_secs(5))
718        );
719        assert_eq!(
720            limiter.state.ledger.lock().instruments[&name].tokens,
721            5 * TOKEN_UNITS
722        );
723    }
724
725    #[rstest]
726    fn asset_types_and_instruments_keep_independent_credit() {
727        let limiter = RequestLimiter::new(
728            RateLimits {
729                matching: 100,
730                instrument: 1,
731                option: 2,
732            },
733            FakeRelativeClock::default(),
734        );
735
736        for (name, burst) in [
737            ("ETH-PERP", 5),
738            ("ETH-USDC", 5),
739            ("ETH-20261030-3000-C", 10),
740        ] {
741            let name = Ustr::from(name);
742
743            for _ in 0..burst {
744                assert_eq!(
745                    limiter.try_class_ready(RateClass::Matching, Some(&name)),
746                    Ok(0)
747                );
748            }
749
750            assert_eq!(
751                limiter.try_class_ready(RateClass::Matching, Some(&name)),
752                Err(Duration::from_millis(5_000 / burst))
753            );
754        }
755    }
756
757    #[rstest]
758    fn cancellation_classes_charge_only_required_budgets() {
759        let limiter = limiter(Some(100), None);
760        let name = Ustr::from("ETH-PERP");
761        assert_eq!(
762            limiter.try_class_ready(RateClass::CancelByLabel, None),
763            Ok(0)
764        );
765        assert_eq!(
766            limiter.try_class_ready(RateClass::CancelByLabel, Some(&name)),
767            Ok(0)
768        );
769        assert_eq!(
770            limiter.try_class_ready(RateClass::CancelAll, Some(&name)),
771            Ok(0)
772        );
773        assert_eq!(
774            limiter.try_class_ready(RateClass::NonMatching, Some(&name)),
775            Ok(0)
776        );
777        let ledger = limiter.state.ledger.lock();
778        assert_eq!(
779            ledger.requests[RequestBucket::Matching as usize].consumed,
780            2
781        );
782        assert_eq!(
783            ledger.requests[RequestBucket::CancelByLabel as usize].consumed,
784            2
785        );
786        assert_eq!(
787            ledger.requests[RequestBucket::CancelAll as usize].consumed,
788            1
789        );
790        assert_eq!(
791            ledger.requests[RequestBucket::NonMatching as usize].consumed,
792            1
793        );
794        assert_eq!(ledger.instruments[&name].tokens, 4 * TOKEN_UNITS);
795    }
796
797    #[rstest]
798    fn label_endpoint_denial_does_not_consume_matching_points() {
799        let limiter = limiter(Some(100), None);
800
801        for _ in 0..50 {
802            assert_eq!(
803                limiter.try_class_ready(RateClass::CancelByLabel, None),
804                Ok(0)
805            );
806        }
807
808        assert_eq!(
809            limiter.try_class_ready(RateClass::CancelByLabel, None),
810            Err(Duration::from_secs(5))
811        );
812        assert_eq!(
813            limiter.state.ledger.lock().requests[RequestBucket::Matching as usize].consumed,
814            50
815        );
816        assert_eq!(limiter.try_class_ready(RateClass::Matching, None), Ok(0));
817    }
818
819    #[tokio::test]
820    async fn held_reservations_cannot_accumulate_a_second_burst() {
821        let limiter = limiter(Some(100), None);
822        let name = Ustr::from("ETH-PERP");
823        let mut held = Vec::new();
824        for _ in 0..5 {
825            held.push(limiter.reserve_matching_ready(&name).await);
826        }
827
828        limiter.state.clock.advance(Duration::from_secs(5));
829
830        for reservation in held {
831            assert_eq!(reservation.commit(), Ok(()));
832        }
833
834        assert_eq!(
835            limiter.try_class_ready(RateClass::Matching, Some(&name)),
836            Err(Duration::from_secs(1))
837        );
838        let ledger = limiter.state.ledger.lock();
839        assert_eq!(
840            ledger.requests[RequestBucket::Matching as usize].consumed,
841            5
842        );
843        assert_eq!(ledger.instruments[&name].pending, 0);
844        assert_eq!(ledger.instruments[&name].tokens, 0);
845    }
846
847    #[tokio::test]
848    async fn abandoned_reservation_refunds_each_budget_once() {
849        let limiter = limiter(None, None);
850        let name = Ustr::from("ETH-PERP");
851        let reservation = limiter.reserve_matching_ready(&name).await;
852        drop(reservation);
853        let ledger = limiter.state.ledger.lock();
854        assert_eq!(
855            ledger.requests[RequestBucket::Matching as usize].consumed,
856            0
857        );
858        assert_eq!(ledger.instruments[&name].pending, 0);
859        assert_eq!(ledger.instruments[&name].tokens, 5 * TOKEN_UNITS);
860    }
861
862    #[tokio::test]
863    async fn rolled_reservation_never_waits_after_signing() {
864        let limiter = limiter(None, None);
865        let name = Ustr::from("ETH-PERP");
866        let reservation = limiter.reserve_matching_ready(&name).await;
867        limiter.state.clock.advance(Duration::from_secs(5));
868
869        for _ in 0..5 {
870            assert_eq!(limiter.try_class_ready(RateClass::Matching, None), Ok(1));
871        }
872
873        assert_eq!(reservation.commit(), Err(Duration::from_secs(5)));
874        assert_eq!(limiter.state.elapsed_nanos(), RATE_WINDOW_NANOS);
875        let ledger = limiter.state.ledger.lock();
876        assert_eq!(
877            ledger.requests[RequestBucket::Matching as usize].consumed,
878            5
879        );
880        assert_eq!(ledger.instruments[&name].pending, 0);
881        assert_eq!(ledger.instruments[&name].tokens, 5 * TOKEN_UNITS);
882    }
883
884    #[tokio::test]
885    async fn reservation_commit_does_not_charge_twice() {
886        let limiter = limiter(None, None);
887        let name = Ustr::from("ETH-PERP");
888        let reservation = limiter.reserve_matching_ready(&name).await;
889        assert_eq!(reservation.commit(), Ok(()));
890        let ledger = limiter.state.ledger.lock();
891        assert_eq!(
892            ledger.requests[RequestBucket::Matching as usize].consumed,
893            1
894        );
895        assert_eq!(ledger.instruments[&name].pending, 0);
896        assert_eq!(ledger.instruments[&name].tokens, 4 * TOKEN_UNITS);
897    }
898
899    #[rstest]
900    fn first_explicit_override_preserves_debt_and_conflicts_do_not_reset_it() {
901        let limiter = limiter(None, None);
902        let name = Ustr::from("ETH-PERP");
903        assert_eq!(
904            limiter.try_class_ready(RateClass::Matching, Some(&name)),
905            Ok(0)
906        );
907        limiter.state.configure(Some(2), Some(10));
908        limiter.state.configure(None, None);
909        limiter.state.configure(Some(100), Some(100));
910        let ledger = limiter.state.ledger.lock();
911        assert_eq!(ledger.limits.matching, 2);
912        assert_eq!(ledger.limits.instrument, 10);
913        assert_eq!(ledger.limits.option, 10);
914        assert_eq!(
915            ledger.requests[RequestBucket::Matching as usize].consumed,
916            1
917        );
918        assert_eq!(ledger.instruments[&name].tokens, 49 * TOKEN_UNITS);
919    }
920
921    #[rstest]
922    fn registry_shares_transports_session_keys_and_subaccounts_but_not_wallets() {
923        let http =
924            DeriveRateLimiter::shared("https://pacing.example/v3", Some("0xAbC"), None, None);
925        let ws =
926            DeriveRateLimiter::shared("wss://pacing.example:443/v3/ws", Some("0xabc"), None, None);
927        let other =
928            DeriveRateLimiter::shared("https://pacing.example/v3", Some("0xdef"), None, None);
929        let testnet = DeriveRateLimiter::shared(
930            "https://testnet.pacing.example/v3",
931            Some("0xabc"),
932            None,
933            None,
934        );
935        assert!(Arc::ptr_eq(&http.state, &ws.state));
936        assert!(Arc::ptr_eq(&http.public, &ws.public));
937        assert!(Arc::ptr_eq(&http.public, &other.public));
938        assert!(!Arc::ptr_eq(&http.state, &other.state));
939        assert!(!Arc::ptr_eq(&http.state, &testnet.state));
940        assert!(!Arc::ptr_eq(&http.state, &http.public));
941    }
942
943    #[tokio::test]
944    async fn shared_handles_consume_the_same_fixed_window() {
945        let limiter = limiter(None, None);
946
947        let second = RequestLimiter {
948            state: Arc::clone(&limiter.state),
949            public: Arc::clone(&limiter.public),
950        };
951
952        for _ in 0..5 {
953            limiter.await_class_ready(RateClass::Matching, None).await;
954        }
955
956        assert_eq!(second.await_class_ready(RateClass::Matching, None).await, 1);
957        assert_eq!(limiter.state.elapsed_nanos(), RATE_WINDOW_NANOS);
958        let independent =
959            RequestLimiter::new(RateLimits::new(None, None), limiter.state.clock.clone());
960        assert_eq!(
961            independent.try_class_ready(RateClass::Matching, None),
962            Ok(0)
963        );
964    }
965
966    #[rstest]
967    #[expect(
968        clippy::needless_collect,
969        reason = "all threads must start before any can pass the barrier"
970    )]
971    fn concurrent_admission_never_overspends_or_loses_wallet_points() {
972        let limiter = Arc::new(limiter(Some(10), None));
973        let barrier = Arc::new(Barrier::new(8));
974
975        let threads = (0..8)
976            .map(|_| {
977                let limiter = Arc::clone(&limiter);
978                let barrier = Arc::clone(&barrier);
979                thread::spawn(move || {
980                    barrier.wait();
981                    limiter.try_class_ready(RateClass::Matching, Some(&Ustr::from("ETH-PERP")))
982                })
983            })
984            .collect::<Vec<_>>();
985
986        let outcomes = threads
987            .into_iter()
988            .map(|thread| thread.join().unwrap())
989            .collect::<Vec<_>>();
990        assert_eq!(
991            outcomes.iter().filter(|result| **result == Ok(0)).count(),
992            5
993        );
994        assert_eq!(
995            outcomes
996                .iter()
997                .filter(|result| **result == Err(Duration::from_secs(1)))
998                .count(),
999            3
1000        );
1001        let ledger = limiter.state.ledger.lock();
1002        assert_eq!(
1003            ledger.requests[RequestBucket::Matching as usize].consumed,
1004            5
1005        );
1006        assert_eq!(ledger.instruments[&Ustr::from("ETH-PERP")].tokens, 0);
1007    }
1008
1009    #[tokio::test]
1010    async fn delayed_abandonment_refunds_only_the_held_credit() {
1011        let limiter = limiter(Some(100), None);
1012        let name = Ustr::from("ETH-PERP");
1013        let mut held = Vec::new();
1014        for _ in 0..5 {
1015            held.push(limiter.reserve_matching_ready(&name).await);
1016        }
1017
1018        limiter.state.clock.advance(Duration::from_secs(5));
1019        drop(held.pop());
1020        assert_eq!(
1021            limiter.try_class_ready(RateClass::Matching, Some(&name)),
1022            Ok(1)
1023        );
1024        assert_eq!(
1025            limiter.try_class_ready(RateClass::Matching, Some(&name)),
1026            Err(Duration::from_secs(1))
1027        );
1028        drop(held);
1029
1030        for _ in 0..4 {
1031            assert_eq!(
1032                limiter.try_class_ready(RateClass::Matching, Some(&name)),
1033                Ok(1)
1034            );
1035        }
1036
1037        assert_eq!(
1038            limiter.try_class_ready(RateClass::Matching, Some(&name)),
1039            Err(Duration::from_secs(1))
1040        );
1041    }
1042
1043    #[rstest]
1044    fn depleted_matching_budget_keeps_reads_and_cancel_all_independent() {
1045        let limiter = limiter(None, None);
1046
1047        for _ in 0..5 {
1048            assert_eq!(limiter.try_class_ready(RateClass::Matching, None), Ok(0));
1049        }
1050
1051        assert_eq!(
1052            limiter.try_class_ready(RateClass::CancelByLabel, None),
1053            Err(Duration::from_secs(5))
1054        );
1055
1056        for (class, allowance) in [(RateClass::NonMatching, 25), (RateClass::CancelAll, 5)] {
1057            for _ in 0..allowance {
1058                assert_eq!(limiter.try_class_ready(class, None), Ok(0));
1059            }
1060
1061            assert_eq!(
1062                limiter.try_class_ready(class, None),
1063                Err(Duration::from_secs(5))
1064            );
1065        }
1066
1067        let ledger = limiter.state.ledger.lock();
1068        assert_eq!(
1069            ledger.requests[RequestBucket::Matching as usize].consumed,
1070            5
1071        );
1072        assert_eq!(
1073            ledger.requests[RequestBucket::CancelByLabel as usize].consumed,
1074            0
1075        );
1076    }
1077}
nautilus_derive/common/rate_limit.rs

nautilus_derive/common/
rate_limit.rs
