1use std::num::NonZeroU32;
2use std::sync::Arc;
3use std::time::Duration;
4
5use governor::Quota;
6use reqwest::Method;
7
8type DirectLimiter = governor::RateLimiter<
9 governor::state::NotKeyed,
10 governor::state::InMemoryState,
11 governor::clock::DefaultClock,
12>;
13
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16#[allow(dead_code)]
17enum MatchMode {
18 Prefix,
22 Exact,
24}
25
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31struct RateSpec {
32 count: u32,
33 period: Duration,
34}
35
36struct Bucket {
42 #[cfg_attr(not(test), allow(dead_code))]
45 spec: RateSpec,
46 limiter: DirectLimiter,
47}
48
49impl Bucket {
50 fn new(count: u32, period: Duration) -> Arc<Self> {
51 Arc::new(Self {
52 spec: RateSpec { count, period },
53 limiter: DirectLimiter::direct(quota(count, period)),
54 })
55 }
56}
57
58struct EndpointLimit {
60 path_prefix: &'static str,
61 method: Option<Method>,
62 match_mode: MatchMode,
63 buckets: Vec<Arc<Bucket>>,
65}
66
67impl EndpointLimit {
68 fn matches(&self, path: &str, method: Option<&Method>) -> bool {
73 let path_matches = match self.match_mode {
74 MatchMode::Exact => path == self.path_prefix,
75 MatchMode::Prefix => {
76 match path.strip_prefix(self.path_prefix) {
79 Some(rest) => rest.is_empty() || rest.starts_with('/') || rest.starts_with('?'),
80 None => false,
81 }
82 }
83 };
84 if !path_matches {
85 return false;
86 }
87 match &self.method {
88 Some(expected) => method == Some(expected),
89 None => true,
90 }
91 }
92}
93
94#[derive(Clone)]
99pub struct RateLimiter {
100 inner: Arc<RateLimiterInner>,
101}
102
103impl std::fmt::Debug for RateLimiter {
104 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
105 f.debug_struct("RateLimiter")
106 .field("endpoints", &self.inner.limits.len())
107 .finish()
108 }
109}
110
111struct RateLimiterInner {
112 limits: Vec<EndpointLimit>,
113 default: DirectLimiter,
114}
115
116fn quota(count: u32, period: Duration) -> Quota {
121 let count = count.max(1);
122 let interval = period / count;
123 Quota::with_period(interval)
124 .expect("quota interval must be non-zero")
125 .allow_burst(NonZeroU32::new(count).unwrap())
126}
127
128fn endpoint_limit(
130 path_prefix: &'static str,
131 method: Option<Method>,
132 buckets: Vec<Arc<Bucket>>,
133) -> EndpointLimit {
134 EndpointLimit {
135 path_prefix,
136 method,
137 match_mode: MatchMode::Prefix,
138 buckets,
139 }
140}
141
142fn simple_limit(
144 path_prefix: &'static str,
145 method: Option<Method>,
146 count: u32,
147 period: Duration,
148) -> EndpointLimit {
149 endpoint_limit(path_prefix, method, vec![Bucket::new(count, period)])
150}
151
152fn dual_limit(
154 path_prefix: &'static str,
155 method: Method,
156 burst: (u32, Duration),
157 sustained: (u32, Duration),
158) -> EndpointLimit {
159 endpoint_limit(
160 path_prefix,
161 Some(method),
162 vec![
163 Bucket::new(burst.0, burst.1),
164 Bucket::new(sustained.0, sustained.1),
165 ],
166 )
167}
168
169impl RateLimiter {
170 pub async fn acquire(&self, path: &str, method: Option<&Method>) {
175 self.inner.default.until_ready().await;
176
177 if let Some(limit) = self.inner.limits.iter().find(|l| l.matches(path, method)) {
178 for bucket in &limit.buckets {
179 bucket.limiter.until_ready().await;
180 }
181 }
182 }
183
184 #[cfg(test)]
190 fn resolve_specs(&self, path: &str, method: Option<&Method>) -> Vec<RateSpec> {
191 self.inner
192 .limits
193 .iter()
194 .find(|l| l.matches(path, method))
195 .map(|l| l.buckets.iter().map(|b| b.spec).collect())
196 .unwrap_or_default()
197 }
198
199 pub fn clob_default() -> Self {
219 let ten_sec = Duration::from_secs(10);
220 let ten_min = Duration::from_secs(600);
221 let get = Some(Method::GET);
222
223 let ledger_group = Bucket::new(900, ten_sec);
225
226 Self {
227 inner: Arc::new(RateLimiterInner {
228 default: DirectLimiter::direct(quota(9_000, ten_sec)),
229 limits: vec![
230 simple_limit("/balance-allowance/update", None, 50, ten_sec),
233 simple_limit("/balance-allowance", None, 200, ten_sec),
234 dual_limit("/order", Method::POST, (5_000, ten_sec), (120_000, ten_min)),
236 dual_limit(
237 "/order",
238 Method::DELETE,
239 (5_000, ten_sec),
240 (120_000, ten_min),
241 ),
242 dual_limit("/orders", Method::POST, (2_000, ten_sec), (21_000, ten_min)),
243 dual_limit(
244 "/orders",
245 Method::DELETE,
246 (2_000, ten_sec),
247 (15_000, ten_min),
248 ),
249 dual_limit(
250 "/cancel-all",
251 Method::DELETE,
252 (250, ten_sec),
253 (6_000, ten_min),
254 ),
255 dual_limit(
256 "/cancel-market-orders",
257 Method::DELETE,
258 (1_500, ten_sec),
259 (21_000, ten_min),
260 ),
261 endpoint_limit(
264 "/notifications",
265 None,
266 vec![ledger_group.clone(), Bucket::new(125, ten_sec)],
267 ),
268 endpoint_limit("/trades", get.clone(), vec![ledger_group.clone()]),
269 endpoint_limit("/orders", get.clone(), vec![ledger_group.clone()]),
270 endpoint_limit("/order", get.clone(), vec![ledger_group]),
271 simple_limit("/data/orders", None, 500, ten_sec),
275 simple_limit("/data/trades", None, 500, ten_sec),
276 simple_limit("/data", None, 500, ten_sec),
277 simple_limit("/auth", None, 100, ten_sec),
279 simple_limit("/prices-history", None, 1_000, ten_sec),
283 simple_limit("/book", None, 1_500, ten_sec),
284 simple_limit("/books", None, 500, ten_sec),
285 simple_limit("/price", None, 1_500, ten_sec),
286 simple_limit("/prices", None, 500, ten_sec),
287 simple_limit("/midpoint", None, 1_500, ten_sec),
288 simple_limit("/midpoints", None, 500, ten_sec),
289 simple_limit("/tick-size", None, 200, ten_sec),
290 simple_limit("/ok", None, 100, ten_sec),
292 simple_limit("/markets", None, 1_500, ten_sec),
297 simple_limit("/neg-risk", None, 1_500, ten_sec),
298 ],
299 }),
300 }
301 }
302
303 pub fn gamma_default() -> Self {
323 let ten_sec = Duration::from_secs(10);
324
325 Self {
326 inner: Arc::new(RateLimiterInner {
327 default: DirectLimiter::direct(quota(4_000, ten_sec)),
328 limits: vec![
329 simple_limit("/comments", None, 200, ten_sec),
330 simple_limit("/tags", None, 200, ten_sec),
331 simple_limit("/markets", None, 300, ten_sec),
332 simple_limit("/public-search", None, 350, ten_sec),
333 simple_limit("/events", None, 500, ten_sec),
334 simple_limit("/status", None, 100, ten_sec),
335 ],
336 }),
337 }
338 }
339
340 pub fn data_default() -> Self {
366 let ten_sec = Duration::from_secs(10);
367
368 Self {
369 inner: Arc::new(RateLimiterInner {
370 default: DirectLimiter::direct(quota(1_000, ten_sec)),
371 limits: vec![
372 simple_limit("/closed-positions", None, 150, ten_sec),
373 simple_limit("/positions", None, 150, ten_sec),
374 simple_limit("/trades", None, 200, ten_sec),
375 simple_limit("/user-pnl", None, 200, ten_sec),
376 simple_limit("/", None, 100, ten_sec),
377 ],
378 }),
379 }
380 }
381
382 pub fn relay_default() -> Self {
386 Self {
387 inner: Arc::new(RateLimiterInner {
388 default: DirectLimiter::direct(quota(25, Duration::from_secs(60))),
389 limits: vec![],
390 }),
391 }
392 }
393}
394
395#[derive(Debug, Clone)]
397pub struct RetryConfig {
398 pub max_retries: u32,
400 pub initial_backoff_ms: u64,
402 pub max_backoff_ms: u64,
404}
405
406impl Default for RetryConfig {
407 fn default() -> Self {
408 Self {
409 max_retries: 3,
410 initial_backoff_ms: 500,
411 max_backoff_ms: 10_000,
412 }
413 }
414}
415
416impl RetryConfig {
417 pub fn backoff(&self, attempt: u32) -> Duration {
422 let base = self
423 .initial_backoff_ms
424 .saturating_mul(1u64 << attempt.min(10));
425 let capped = base.min(self.max_backoff_ms);
426 let jitter_factor = 0.75 + (fastrand::f64() * 0.5);
428 let ms = (capped as f64 * jitter_factor) as u64;
429 Duration::from_millis(ms.max(1))
430 }
431}
432
433#[cfg(test)]
434mod agreement {
435 use super::*;
445
446 pub type DocumentedRule = (&'static str, Option<Method>, Vec<(u32, u64)>);
449
450 pub fn assert_matches_published(rl: &RateLimiter, rules: Vec<DocumentedRule>, general: u32) {
456 for (path, method, expected) in rules {
457 let resolved = rl.resolve_specs(path, method.as_ref());
458 assert!(
459 !resolved.is_empty(),
460 "{method:?} {path} matches no endpoint limit — it falls through to the \
461 general {general}/10s bucket, over-permitting by {}x",
462 general / expected[0].0.max(1),
463 );
464 let actual: Vec<(u32, u64)> = resolved
465 .iter()
466 .map(|s| (s.count, s.period.as_secs()))
467 .collect();
468 assert_eq!(
469 actual, expected,
470 "{method:?} {path} resolves to {actual:?}, published limit is {expected:?}"
471 );
472 }
473 }
474
475 pub fn assert_unconfigured(rl: &RateLimiter, path: &str) {
480 assert!(
481 rl.resolve_specs(path, Some(&Method::GET)).is_empty(),
482 "{path} has an endpoint limit configured, but the host answers 404 there — \
483 the entry is dead configuration and the real route is going unlimited"
484 );
485 }
486
487 pub async fn assert_throttles_after(rl: &RateLimiter, path: &str, count: u32) {
492 for _ in 0..count {
493 rl.acquire(path, Some(&Method::GET)).await;
494 }
495
496 let start = std::time::Instant::now();
497 rl.acquire(path, Some(&Method::GET)).await;
498 let waited = start.elapsed();
499
500 assert!(
501 waited >= Duration::from_millis(25),
502 "request {} to {path} returned in {waited:?}; the cap is not being enforced",
503 count + 1,
504 );
505 }
506}
507
508#[cfg(test)]
509mod documented_data_limits {
510 use super::agreement::*;
528 use super::*;
529
530 fn documented() -> Vec<DocumentedRule> {
532 vec![
533 ("/trades", Some(Method::GET), vec![(200, 10)]),
534 ("/positions", Some(Method::GET), vec![(150, 10)]),
535 ("/closed-positions", Some(Method::GET), vec![(150, 10)]),
536 ("/", Some(Method::GET), vec![(100, 10)]),
537 ("/user-pnl", Some(Method::GET), vec![(200, 10)]),
538 ]
539 }
540
541 #[test]
542 fn every_documented_endpoint_resolves_to_its_published_quota() {
543 assert_matches_published(&RateLimiter::data_default(), documented(), 1_000);
544 }
545
546 #[test]
547 fn the_health_cap_is_attached_to_the_route_the_host_answers_on() {
548 assert_unconfigured(&RateLimiter::data_default(), "/ok");
551 }
552
553 #[test]
554 fn the_root_health_rule_does_not_swallow_every_other_route() {
555 let rl = RateLimiter::data_default();
559 for (path, expected) in [
560 ("/positions", 150),
561 ("/closed-positions", 150),
562 ("/trades", 200),
563 ("/", 100),
564 ] {
565 let specs = rl.resolve_specs(path, Some(&Method::GET));
566 assert_eq!(
567 specs[0].count, expected,
568 "{path} resolved through the wrong rule — the `/` entry is over-matching"
569 );
570 }
571 }
572
573 #[tokio::test]
574 async fn the_closed_positions_cap_actually_throttles() {
575 assert_throttles_after(&RateLimiter::data_default(), "/closed-positions", 150).await;
578 }
579
580 #[tokio::test]
581 async fn closed_positions_and_positions_do_not_share_an_allowance() {
582 let rl = RateLimiter::data_default();
586 for _ in 0..150 {
587 rl.acquire("/closed-positions", Some(&Method::GET)).await;
588 }
589
590 let start = std::time::Instant::now();
591 rl.acquire("/positions", Some(&Method::GET)).await;
592 assert!(
593 start.elapsed() < Duration::from_millis(25),
594 "/positions was throttled by /closed-positions draining its own bucket"
595 );
596 }
597}
598
599#[cfg(test)]
600mod documented_gamma_limits {
601 use super::agreement::*;
609 use super::*;
610
611 fn documented() -> Vec<DocumentedRule> {
613 vec![
614 ("/events", Some(Method::GET), vec![(500, 10)]),
615 ("/public-search", Some(Method::GET), vec![(350, 10)]),
616 ("/markets", Some(Method::GET), vec![(300, 10)]),
617 ("/comments", Some(Method::GET), vec![(200, 10)]),
618 ("/tags", Some(Method::GET), vec![(200, 10)]),
619 ("/status", Some(Method::GET), vec![(100, 10)]),
620 ]
621 }
622
623 #[test]
624 fn every_documented_endpoint_resolves_to_its_published_quota() {
625 assert_matches_published(&RateLimiter::gamma_default(), documented(), 4_000);
626 }
627
628 #[test]
629 fn the_health_cap_is_attached_to_the_route_the_host_answers_on() {
630 assert_unconfigured(&RateLimiter::gamma_default(), "/ok");
631 }
632
633 #[test]
634 fn the_markets_plus_events_group_cap_can_never_bind() {
635 let rl = RateLimiter::gamma_default();
640 let markets = rl.resolve_specs("/markets", Some(&Method::GET))[0].count;
641 let events = rl.resolve_specs("/events", Some(&Method::GET))[0].count;
642 assert!(
643 markets + events <= 900,
644 "/markets ({markets}) + /events ({events}) now exceeds the published 900/10s \
645 group cap, which is no longer unreachable and must be modelled"
646 );
647 }
648
649 #[tokio::test]
650 async fn the_markets_cap_actually_throttles() {
651 assert_throttles_after(&RateLimiter::gamma_default(), "/markets", 300).await;
652 }
653}
654
655#[cfg(test)]
656mod documented_limits {
657 use super::agreement::DocumentedRule;
667 use super::*;
668
669 fn documented() -> Vec<DocumentedRule> {
671 vec![
672 ("/balance-allowance", Some(Method::GET), vec![(200, 10)]),
674 (
675 "/balance-allowance/update",
676 Some(Method::GET),
677 vec![(50, 10)],
678 ),
679 (
681 "/order",
682 Some(Method::POST),
683 vec![(5_000, 10), (120_000, 600)],
684 ),
685 (
686 "/order",
687 Some(Method::DELETE),
688 vec![(5_000, 10), (120_000, 600)],
689 ),
690 (
691 "/orders",
692 Some(Method::POST),
693 vec![(2_000, 10), (21_000, 600)],
694 ),
695 (
696 "/orders",
697 Some(Method::DELETE),
698 vec![(2_000, 10), (15_000, 600)],
699 ),
700 (
701 "/cancel-all",
702 Some(Method::DELETE),
703 vec![(250, 10), (6_000, 600)],
704 ),
705 (
706 "/cancel-market-orders",
707 Some(Method::DELETE),
708 vec![(1_500, 10), (21_000, 600)],
709 ),
710 ("/trades", Some(Method::GET), vec![(900, 10)]),
712 ("/orders", Some(Method::GET), vec![(900, 10)]),
713 ("/order", Some(Method::GET), vec![(900, 10)]),
714 (
715 "/notifications",
716 Some(Method::GET),
717 vec![(900, 10), (125, 10)],
718 ),
719 ("/data/orders", Some(Method::GET), vec![(500, 10)]),
720 ("/data/trades", Some(Method::GET), vec![(500, 10)]),
721 ("/book", Some(Method::GET), vec![(1_500, 10)]),
723 ("/books", Some(Method::POST), vec![(500, 10)]),
724 ("/price", Some(Method::GET), vec![(1_500, 10)]),
725 ("/prices", Some(Method::POST), vec![(500, 10)]),
726 ("/midpoint", Some(Method::GET), vec![(1_500, 10)]),
727 ("/midpoints", Some(Method::POST), vec![(500, 10)]),
728 ("/prices-history", Some(Method::GET), vec![(1_000, 10)]),
729 ("/tick-size", Some(Method::GET), vec![(200, 10)]),
730 ("/auth/api-key", Some(Method::POST), vec![(100, 10)]),
732 ("/ok", Some(Method::GET), vec![(100, 10)]),
733 ]
734 }
735
736 #[test]
737 fn every_documented_endpoint_resolves_to_its_published_quota() {
738 let rl = RateLimiter::clob_default();
739
740 for (path, method, expected) in documented() {
741 let resolved = rl.resolve_specs(path, method.as_ref());
742 assert!(
743 !resolved.is_empty(),
744 "{method:?} {path} matches no endpoint limit — it falls through to the \
745 general {}/10s bucket, over-permitting by {}x",
746 9_000,
747 9_000 / expected[0].0.max(1),
748 );
749 let actual: Vec<(u32, u64)> = resolved
750 .iter()
751 .map(|s| (s.count, s.period.as_secs()))
752 .collect();
753 assert_eq!(
754 actual, expected,
755 "{method:?} {path} resolves to {actual:?}, published limit is {expected:?}"
756 );
757 }
758 }
759
760 #[test]
761 fn batch_endpoints_do_not_inherit_their_singular_sibling() {
762 let rl = RateLimiter::clob_default();
765 for (batch, singular) in [
766 ("/books", "/book"),
767 ("/prices", "/price"),
768 ("/midpoints", "/midpoint"),
769 ] {
770 let batch_specs = rl.resolve_specs(batch, Some(&Method::POST));
771 let singular_specs = rl.resolve_specs(singular, Some(&Method::GET));
772 assert_ne!(
773 batch_specs, singular_specs,
774 "{batch} is being limited as if it were {singular}"
775 );
776 assert_eq!(batch_specs[0].count, 500, "{batch} should allow 500/10s");
777 }
778 }
779
780 #[test]
781 fn the_ledger_group_cap_is_one_shared_bucket() {
782 let rl = RateLimiter::clob_default();
786 let group: Vec<_> = ["/trades", "/orders", "/order", "/notifications"]
787 .iter()
788 .map(|p| {
789 rl.inner
790 .limits
791 .iter()
792 .find(|l| l.matches(p, Some(&Method::GET)))
793 .unwrap_or_else(|| panic!("{p} should match a ledger entry"))
794 .buckets[0]
795 .clone()
796 })
797 .collect();
798
799 for other in &group[1..] {
800 assert!(
801 Arc::ptr_eq(&group[0], other),
802 "ledger endpoints must share one bucket, not hold copies"
803 );
804 }
805 }
806
807 #[test]
808 fn balance_allowance_update_is_not_shadowed_by_its_parent_path() {
809 let rl = RateLimiter::clob_default();
813 let update = rl.resolve_specs("/balance-allowance/update", Some(&Method::GET));
814 assert_eq!(
815 update[0].count, 50,
816 "the tighter /balance-allowance/update rule must be ordered first"
817 );
818 }
819
820 #[tokio::test]
821 async fn a_documented_cap_actually_throttles() {
822 let rl = RateLimiter::clob_default();
826 for _ in 0..200 {
827 rl.acquire("/tick-size", Some(&Method::GET)).await;
828 }
829
830 let start = std::time::Instant::now();
831 rl.acquire("/tick-size", Some(&Method::GET)).await;
832 let waited = start.elapsed();
833
834 assert!(
835 waited >= Duration::from_millis(25),
836 "201st /tick-size request returned in {waited:?}; the cap is not being enforced"
837 );
838 }
839
840 #[tokio::test]
841 async fn the_ledger_group_allowance_is_consumed_jointly() {
842 let rl = RateLimiter::clob_default();
846 for _ in 0..900 {
847 rl.acquire("/trades", Some(&Method::GET)).await;
848 }
849
850 let start = std::time::Instant::now();
851 rl.acquire("/orders", Some(&Method::GET)).await;
852 let waited = start.elapsed();
853
854 assert!(
855 waited >= Duration::from_millis(5),
856 "GET /orders returned in {waited:?} after /trades drained the shared 900/10s \
857 allowance — the group cap is not actually shared"
858 );
859 }
860
861 #[test]
862 fn post_order_is_not_throttled_by_the_ledger_group() {
863 let rl = RateLimiter::clob_default();
868 let specs = rl.resolve_specs("/order", Some(&Method::POST));
869 assert_eq!(specs[0].count, 5_000);
870 assert!(
871 !specs.iter().any(|s| s.count == 900),
872 "POST /order must not be caught by the ledger read cap"
873 );
874 }
875}
876
877#[cfg(test)]
878mod tests {
879 use super::*;
880
881 #[test]
884 fn test_retry_config_default() {
885 let cfg = RetryConfig::default();
886 assert_eq!(cfg.max_retries, 3);
887 assert_eq!(cfg.initial_backoff_ms, 500);
888 assert_eq!(cfg.max_backoff_ms, 10_000);
889 }
890
891 #[test]
892 fn test_backoff_attempt_zero() {
893 let cfg = RetryConfig::default();
894 let d = cfg.backoff(0);
895 let ms = d.as_millis() as u64;
898 assert!(
899 (375..=625).contains(&ms),
900 "attempt 0: {ms}ms not in [375, 625]"
901 );
902 }
903
904 #[test]
905 fn test_backoff_exponential_growth() {
906 let cfg = RetryConfig::default();
907 let d0 = cfg.backoff(0);
908 let d1 = cfg.backoff(1);
909 let d2 = cfg.backoff(2);
910 assert!(d0 < d1, "d0={d0:?} should be < d1={d1:?}");
911 assert!(d1 < d2, "d1={d1:?} should be < d2={d2:?}");
912 }
913
914 #[test]
915 fn test_backoff_jitter_bounds() {
916 let cfg = RetryConfig::default();
917 for attempt in 0..20 {
918 let d = cfg.backoff(attempt);
919 let base = cfg
920 .initial_backoff_ms
921 .saturating_mul(1u64 << attempt.min(10));
922 let capped = base.min(cfg.max_backoff_ms);
923 let lower = (capped as f64 * 0.75) as u64;
924 let upper = (capped as f64 * 1.25) as u64;
925 let ms = d.as_millis() as u64;
926 assert!(
927 ms >= lower.max(1) && ms <= upper,
928 "attempt {attempt}: {ms}ms not in [{lower}, {upper}]"
929 );
930 }
931 }
932
933 #[test]
934 fn test_backoff_max_capping() {
935 let cfg = RetryConfig::default();
936 for attempt in 5..=10 {
937 let d = cfg.backoff(attempt);
938 let ceiling = (cfg.max_backoff_ms as f64 * 1.25) as u64;
939 assert!(
940 d.as_millis() as u64 <= ceiling,
941 "attempt {attempt}: {:?} exceeded ceiling {ceiling}ms",
942 d
943 );
944 }
945 }
946
947 #[test]
948 fn test_backoff_very_high_attempt() {
949 let cfg = RetryConfig::default();
950 let d = cfg.backoff(100);
951 let ceiling = (cfg.max_backoff_ms as f64 * 1.25) as u64;
952 assert!(d.as_millis() as u64 <= ceiling);
953 assert!(d.as_millis() >= 1);
954 }
955
956 #[test]
957 fn test_backoff_jitter_distribution() {
958 let cfg = RetryConfig::default();
961 let midpoint = cfg.initial_backoff_ms; let (mut below, mut above) = (0u32, 0u32);
963 for _ in 0..200 {
964 let ms = cfg.backoff(0).as_millis() as u64;
965 if ms < midpoint {
966 below += 1;
967 } else {
968 above += 1;
969 }
970 }
971 assert!(
972 below >= 20 && above >= 20,
973 "jitter looks degenerate: {below} below midpoint, {above} above"
974 );
975 }
976
977 #[test]
980 fn test_quota_creation() {
981 let _ = quota(100, Duration::from_secs(10));
983 let _ = quota(1, Duration::from_secs(60));
984 let _ = quota(9_000, Duration::from_secs(10));
985 }
986
987 #[test]
988 fn test_quota_edge_zero_count() {
989 let _ = quota(0, Duration::from_secs(10));
991 }
992
993 #[test]
996 fn test_clob_default_construction() {
997 let rl = RateLimiter::clob_default();
998 assert_eq!(rl.inner.limits.len(), 27);
999 assert!(format!("{:?}", rl).contains("endpoints"));
1000 }
1001
1002 #[test]
1003 fn test_gamma_default_construction() {
1004 let rl = RateLimiter::gamma_default();
1005 assert_eq!(rl.inner.limits.len(), 6);
1006 }
1007
1008 #[test]
1009 fn test_data_default_construction() {
1010 let rl = RateLimiter::data_default();
1011 assert_eq!(rl.inner.limits.len(), 5);
1012 }
1013
1014 #[test]
1015 fn test_relay_default_construction() {
1016 let rl = RateLimiter::relay_default();
1017 assert_eq!(rl.inner.limits.len(), 0);
1018 }
1019
1020 #[test]
1021 fn test_rate_limiter_debug_format() {
1022 let rl = RateLimiter::clob_default();
1023 let dbg = format!("{:?}", rl);
1024 assert!(dbg.contains("RateLimiter"), "missing struct name: {dbg}");
1025 assert!(dbg.contains("endpoints: 27"), "missing count: {dbg}");
1026 }
1027
1028 #[test]
1031 fn test_clob_tighter_rules_precede_the_prefixes_that_would_shadow_them() {
1032 let rl = RateLimiter::clob_default();
1036 let index_of = |path: &str| {
1037 rl.inner
1038 .limits
1039 .iter()
1040 .position(|l| l.path_prefix == path)
1041 .unwrap_or_else(|| panic!("{path} should be configured"))
1042 };
1043
1044 for (specific, general) in [
1045 ("/balance-allowance/update", "/balance-allowance"),
1046 ("/data/orders", "/data"),
1047 ("/data/trades", "/data"),
1048 ] {
1049 assert!(
1050 index_of(specific) < index_of(general),
1051 "{specific} must be matched before {general} or it can never win"
1052 );
1053 }
1054 }
1055
1056 #[tokio::test]
1059 async fn test_acquire_single_completes_immediately() {
1060 let rl = RateLimiter::clob_default();
1061 let start = std::time::Instant::now();
1062 rl.acquire("/order", Some(&Method::POST)).await;
1063 assert!(start.elapsed() < Duration::from_millis(50));
1064 }
1065
1066 #[tokio::test]
1067 async fn test_acquire_matches_endpoint_by_prefix() {
1068 let rl = RateLimiter::clob_default();
1069 let start = std::time::Instant::now();
1070 rl.acquire("/order/123", Some(&Method::POST)).await;
1072 assert!(start.elapsed() < Duration::from_millis(50));
1073 }
1074
1075 #[tokio::test]
1076 async fn test_acquire_prefix_respects_segment_boundary() {
1077 let rl = RateLimiter::clob_default();
1078 let limits = &rl.inner.limits;
1079
1080 let price_idx = limits
1082 .iter()
1083 .position(|l| l.path_prefix == "/price")
1084 .expect("/price endpoint exists");
1085
1086 let prices_history_idx = limits
1088 .iter()
1089 .position(|l| l.path_prefix == "/prices-history")
1090 .expect("/prices-history endpoint exists");
1091
1092 assert!(
1094 prices_history_idx < price_idx,
1095 "/prices-history (idx {prices_history_idx}) should come before /price (idx {price_idx})"
1096 );
1097 }
1098
1099 #[test]
1100 fn test_match_mode_prefix_segment_boundary() {
1101 let pattern = "/price";
1103
1104 let check = |path: &str| -> bool {
1105 match path.strip_prefix(pattern) {
1106 Some(rest) => rest.is_empty() || rest.starts_with('/') || rest.starts_with('?'),
1107 None => false,
1108 }
1109 };
1110
1111 assert!(check("/price"), "exact match");
1113 assert!(check("/price/foo"), "sub-path");
1114 assert!(check("/price?token=abc"), "query params");
1115
1116 assert!(!check("/prices-history"), "partial word /prices-history");
1118 assert!(!check("/pricelist"), "partial word /pricelist");
1119 assert!(!check("/pricing"), "partial word /pricing");
1120
1121 assert!(!check("/midpoint"), "different prefix");
1123 }
1124
1125 #[test]
1126 fn test_match_mode_exact() {
1127 let pattern = "/trades";
1129
1130 let check = |path: &str| -> bool { path == pattern };
1131
1132 assert!(check("/trades"), "exact match");
1133 assert!(!check("/trades/123"), "sub-path should not match");
1134 assert!(!check("/trades?limit=10"), "query params should not match");
1135 assert!(!check("/traded"), "different word should not match");
1136 }
1137
1138 #[tokio::test]
1139 async fn test_acquire_method_filtering() {
1140 let rl = RateLimiter::clob_default();
1141 let start = std::time::Instant::now();
1142 rl.acquire("/order", Some(&Method::GET)).await;
1144 assert!(start.elapsed() < Duration::from_millis(50));
1145 }
1146
1147 #[tokio::test]
1148 async fn test_acquire_no_endpoint_match_uses_default_only() {
1149 let rl = RateLimiter::clob_default();
1150 let start = std::time::Instant::now();
1151 rl.acquire("/unknown/path", None).await;
1152 assert!(start.elapsed() < Duration::from_millis(50));
1153 }
1154
1155 #[tokio::test]
1156 async fn test_acquire_method_none_matches_any_method() {
1157 let rl = RateLimiter::gamma_default();
1158 let start = std::time::Instant::now();
1159 rl.acquire("/events", Some(&Method::GET)).await;
1161 rl.acquire("/events", Some(&Method::POST)).await;
1162 rl.acquire("/events", None).await;
1163 assert!(start.elapsed() < Duration::from_millis(50));
1164 }
1165
1166 #[test]
1169 fn test_clob_price_and_prices_history_are_distinct() {
1170 let rl = RateLimiter::clob_default();
1171 let limits = &rl.inner.limits;
1172
1173 let price = limits.iter().find(|l| l.path_prefix == "/price").unwrap();
1174 let prices_history = limits
1175 .iter()
1176 .find(|l| l.path_prefix == "/prices-history")
1177 .unwrap();
1178
1179 assert_eq!(price.match_mode, MatchMode::Prefix);
1181 assert_eq!(prices_history.match_mode, MatchMode::Prefix);
1182
1183 if let Some(rest) = "/prices-history".strip_prefix(price.path_prefix) {
1185 assert!(
1186 !rest.is_empty() && !rest.starts_with('/') && !rest.starts_with('?'),
1187 "/prices-history must not match /price pattern, rest = '{rest}'"
1188 );
1189 }
1190 }
1191
1192 #[test]
1193 fn test_data_positions_and_closed_positions_are_distinct() {
1194 let rl = RateLimiter::data_default();
1199
1200 let closed = rl.resolve_specs("/closed-positions", Some(&Method::GET));
1201 let positions = rl.resolve_specs("/positions", Some(&Method::GET));
1202 assert_eq!(closed, positions, "both are published at 150/10s");
1203
1204 let bucket_for = |path: &str| {
1205 rl.inner
1206 .limits
1207 .iter()
1208 .find(|l| l.matches(path, Some(&Method::GET)))
1209 .unwrap_or_else(|| panic!("{path} should match a rule"))
1210 .buckets[0]
1211 .clone()
1212 };
1213 assert!(
1214 !Arc::ptr_eq(&bucket_for("/closed-positions"), &bucket_for("/positions")),
1215 "equal quotas must still be separate buckets — upstream publishes \
1216 150/10s each, not 150/10s combined"
1217 );
1218 }
1219
1220 #[test]
1221 fn test_all_clob_endpoints_have_match_mode() {
1222 let rl = RateLimiter::clob_default();
1223 for limit in &rl.inner.limits {
1224 assert!(
1226 limit.match_mode == MatchMode::Prefix || limit.match_mode == MatchMode::Exact,
1227 "endpoint {} has no valid match mode",
1228 limit.path_prefix
1229 );
1230 }
1231 }
1232
1233 #[tokio::test]
1236 async fn test_acquire_concurrent_tasks_all_complete() {
1237 let rl = RateLimiter::clob_default(); let rl = std::sync::Arc::new(rl);
1240
1241 let mut handles = Vec::new();
1242 for _ in 0..10 {
1243 let rl = rl.clone();
1244 handles.push(tokio::spawn(async move {
1245 rl.acquire("/markets", None).await;
1246 }));
1247 }
1248
1249 let start = std::time::Instant::now();
1250 for handle in handles {
1251 handle.await.unwrap();
1252 }
1253 assert!(
1255 start.elapsed() < Duration::from_millis(100),
1256 "concurrent acquires took too long: {:?}",
1257 start.elapsed()
1258 );
1259 }
1260
1261 #[tokio::test]
1262 async fn test_acquire_concurrent_different_endpoints() {
1263 let rl = std::sync::Arc::new(RateLimiter::clob_default());
1265
1266 let rl1 = rl.clone();
1267 let rl2 = rl.clone();
1268 let rl3 = rl.clone();
1269
1270 let start = std::time::Instant::now();
1271 let (r1, r2, r3) = tokio::join!(
1272 tokio::spawn(async move { rl1.acquire("/markets", None).await }),
1273 tokio::spawn(async move { rl2.acquire("/auth", None).await }),
1274 tokio::spawn(async move { rl3.acquire("/order", Some(&Method::POST)).await }),
1275 );
1276 r1.unwrap();
1277 r2.unwrap();
1278 r3.unwrap();
1279
1280 assert!(
1281 start.elapsed() < Duration::from_millis(50),
1282 "different endpoints should not block: {:?}",
1283 start.elapsed()
1284 );
1285 }
1286
1287 #[test]
1290 fn test_clob_post_order_has_dual_window() {
1291 let rl = RateLimiter::clob_default();
1292 let post_order = rl
1293 .inner
1294 .limits
1295 .iter()
1296 .find(|l| l.path_prefix == "/order" && l.method == Some(Method::POST))
1297 .expect("POST /order endpoint should exist");
1298
1299 assert_eq!(
1300 post_order.buckets.len(),
1301 2,
1302 "POST /order should have a burst and a sustained window"
1303 );
1304 }
1305
1306 #[test]
1307 fn test_clob_delete_order_has_a_sustained_window_too() {
1308 let rl = RateLimiter::clob_default();
1312 let delete_order = rl
1313 .inner
1314 .limits
1315 .iter()
1316 .find(|l| l.path_prefix == "/order" && l.method == Some(Method::DELETE))
1317 .expect("DELETE /order endpoint should exist");
1318
1319 assert_eq!(
1320 delete_order.buckets.len(),
1321 2,
1322 "DELETE /order should have both a burst and a sustained window"
1323 );
1324 }
1325
1326 #[tokio::test]
1327 async fn test_dual_window_both_burst_and_sustained_are_awaited() {
1328 let rl = RateLimiter::clob_default();
1331 let start = std::time::Instant::now();
1332 rl.acquire("/order", Some(&Method::POST)).await;
1333 assert!(
1334 start.elapsed() < Duration::from_millis(50),
1335 "dual window single acquire should be fast: {:?}",
1336 start.elapsed()
1337 );
1338 }
1339
1340 #[test]
1343 fn test_should_retry_exhaustion() {
1344 let client = crate::HttpClientBuilder::new("https://example.com")
1346 .with_retry_config(RetryConfig {
1347 max_retries: 3,
1348 ..RetryConfig::default()
1349 })
1350 .build()
1351 .unwrap();
1352
1353 for attempt in 0..3 {
1355 assert!(
1356 client
1357 .should_retry(reqwest::StatusCode::TOO_MANY_REQUESTS, attempt, None)
1358 .is_some(),
1359 "attempt {attempt} should allow retry"
1360 );
1361 }
1362 assert!(
1364 client
1365 .should_retry(reqwest::StatusCode::TOO_MANY_REQUESTS, 3, None)
1366 .is_none(),
1367 "attempt 3 should exhaust retries"
1368 );
1369 }
1370
1371 #[test]
1372 fn test_should_retry_zero_max_retries_never_retries() {
1373 let client = crate::HttpClientBuilder::new("https://example.com")
1374 .with_retry_config(RetryConfig {
1375 max_retries: 0,
1376 ..RetryConfig::default()
1377 })
1378 .build()
1379 .unwrap();
1380
1381 assert!(
1382 client
1383 .should_retry(reqwest::StatusCode::TOO_MANY_REQUESTS, 0, None)
1384 .is_none(),
1385 "max_retries=0 should never retry"
1386 );
1387 }
1388}