reserve_core/limit/
mod.rs

1//! @docgen Pacing is keyed by registry host, never by extension: one operator answers for hundreds of extensions from one endpoint.
2
3mod policy;
4
5use std::collections::HashMap;
6use std::collections::hash_map::RandomState;
7use std::hash::{BuildHasher, Hasher};
8use std::num::NonZeroU32;
9use std::sync::Arc;
10use std::sync::atomic::{AtomicU64, Ordering};
11use std::time::Duration;
12
13use governor::clock::DefaultClock;
14use governor::state::{InMemoryState, NotKeyed};
15use governor::{Jitter, Quota, RateLimiter};
16use tokio::sync::{Mutex, OwnedSemaphorePermit, Semaphore};
17use tokio::time::Instant;
18
19pub use policy::{
20    CAUTIOUS_LIMIT, DEFAULT_RETRY_AFTER, HostLimit, MAX_RETRY_AFTER, clamp_retry_after,
21    published_limit, starting_limit,
22};
23
24const PROBE_DEADLINE: Duration = Duration::from_secs(30);
25
26type TokenBucket = RateLimiter<NotKeyed, InMemoryState, DefaultClock>;
27
28/// @docgen The three refusals need different waits, and a dropped connection is backpressure rather than a network error, so they never collapse.
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub enum Refusal {
31    Throttled { retry_after: Option<Duration> },
32    Blocked,
33    Dropped,
34}
35
36impl Refusal {
37    #[must_use]
38    fn base_wait(&self) -> Duration {
39        match self {
40            Self::Throttled { retry_after } => clamp_retry_after(*retry_after),
41            Self::Blocked => MAX_RETRY_AFTER,
42            Self::Dropped => DEFAULT_RETRY_AFTER,
43        }
44    }
45
46    #[must_use]
47    const fn has_stated_wait(&self) -> bool {
48        matches!(
49            self,
50            Self::Throttled {
51                retry_after: Some(_)
52            }
53        )
54    }
55}
56
57#[derive(Debug, Clone, PartialEq, Eq)]
58pub struct PacingLimits {
59    pub total_concurrency: usize,
60    pub refusal_threshold: u32,
61    pub recovery_threshold: u32,
62    /// @docgen A spread is added before each request so a fan-out does not land on a registry as one spike.
63    pub jitter: Duration,
64    pub max_backoff: Duration,
65    pub is_cautious: bool,
66    pub per_registry: Option<usize>,
67    pub rate: Option<u32>,
68}
69
70impl Default for PacingLimits {
71    fn default() -> Self {
72        Self {
73            total_concurrency: 24,
74            refusal_threshold: 3,
75            recovery_threshold: 8,
76            jitter: Duration::from_millis(120),
77            max_backoff: Duration::from_secs(120),
78            is_cautious: false,
79            per_registry: None,
80            rate: None,
81        }
82    }
83}
84
85impl PacingLimits {
86    #[must_use]
87    pub fn cautious() -> Self {
88        Self {
89            total_concurrency: 8,
90            refusal_threshold: 2,
91            recovery_threshold: 16,
92            jitter: Duration::from_millis(300),
93            is_cautious: true,
94            ..Self::default()
95        }
96    }
97
98    fn limit_for(&self, host: &str) -> HostLimit {
99        let mut limit = if self.is_cautious {
100            CAUTIOUS_LIMIT
101        } else {
102            starting_limit(host)
103        };
104        if let Some(concurrency) = self.per_registry {
105            limit.concurrency = concurrency.max(1);
106        }
107        if let Some(queries) = self.rate {
108            limit.queries = queries.max(1);
109            limit.window = Duration::from_secs(1);
110        }
111        limit
112    }
113}
114
115#[derive(Debug, Clone, Copy, PartialEq, Eq)]
116pub enum BreakerState {
117    Closed,
118    Open,
119    HalfOpen,
120}
121
122#[derive(Debug, Clone, PartialEq, Eq)]
123pub struct PausedHost {
124    pub host: String,
125    pub remaining_wait: Duration,
126    pub refusals: u32,
127}
128
129/// @docgen The two unread permit fields must stay: dropping them is what frees the global and per-host slots.
130#[derive(Debug)]
131pub struct RequestPermit {
132    host: String,
133    is_probe: bool,
134    _global_slot: OwnedSemaphorePermit,
135    _host_slot: OwnedSemaphorePermit,
136}
137
138impl RequestPermit {
139    #[must_use]
140    pub const fn is_probe(&self) -> bool {
141        self.is_probe
142    }
143
144    #[must_use]
145    pub fn host(&self) -> &str {
146        &self.host
147    }
148}
149
150#[derive(Debug)]
151struct HostState {
152    limiter: Arc<TokenBucket>,
153    slots: Arc<Semaphore>,
154    concurrency: usize,
155    max_concurrency: usize,
156    refusals: u32,
157    successes: u32,
158    open_until: Option<Instant>,
159    is_probing: bool,
160}
161
162impl HostState {
163    fn new(limit: HostLimit) -> Self {
164        Self {
165            limiter: Arc::new(RateLimiter::direct(quota_for(limit))),
166            slots: Arc::new(Semaphore::new(limit.concurrency.max(1))),
167            concurrency: limit.concurrency.max(1),
168            max_concurrency: limit.concurrency.max(1),
169            refusals: 0,
170            successes: 0,
171            open_until: None,
172            is_probing: false,
173        }
174    }
175
176    fn breaker(&self, now: Instant) -> BreakerState {
177        match self.open_until {
178            Some(until) if now < until => BreakerState::Open,
179            Some(_) => BreakerState::HalfOpen,
180            None => {
181                if self.is_probing {
182                    BreakerState::HalfOpen
183                } else {
184                    BreakerState::Closed
185                }
186            }
187        }
188    }
189}
190
191fn quota_for(limit: HostLimit) -> Quota {
192    let rate = limit.per_second_rate().max(0.05);
193    let interval = Duration::from_secs_f64(1.0 / rate);
194    let burst = NonZeroU32::new(limit.queries.max(1)).unwrap_or(NonZeroU32::MIN);
195    Quota::with_period(interval).map_or_else(
196        || Quota::per_second(NonZeroU32::MIN),
197        |quota| quota.allow_burst(burst),
198    )
199}
200
201#[derive(Debug)]
202struct JitterRng(AtomicU64);
203
204impl JitterRng {
205    fn new() -> Self {
206        let seed = RandomState::new().build_hasher().finish() | 1;
207        Self(AtomicU64::new(seed))
208    }
209
210    fn next_u64(&self) -> u64 {
211        let mut x = self.0.load(Ordering::Relaxed);
212        x ^= x << 13;
213        x ^= x >> 7;
214        x ^= x << 17;
215        self.0.store(x, Ordering::Relaxed);
216        x
217    }
218
219    /// @docgen Full jitter spreads retries across the whole window instead of clustering them at its end.
220    fn sample_up_to(&self, ceiling: Duration) -> Duration {
221        let nanos = u64::try_from(ceiling.as_nanos()).unwrap_or(u64::MAX);
222        if nanos == 0 {
223            return Duration::ZERO;
224        }
225        Duration::from_nanos(self.next_u64() % nanos.saturating_add(1))
226    }
227}
228
229#[derive(Debug)]
230pub struct Pacer {
231    pacing: PacingLimits,
232    global_slots: Arc<Semaphore>,
233    hosts: Mutex<HashMap<String, HostState>>,
234    jitter: JitterRng,
235}
236
237impl Pacer {
238    #[must_use]
239    pub fn new(pacing: PacingLimits) -> Self {
240        Self {
241            global_slots: Arc::new(Semaphore::new(pacing.total_concurrency.max(1))),
242            hosts: Mutex::new(HashMap::new()),
243            jitter: JitterRng::new(),
244            pacing,
245        }
246    }
247
248    #[must_use]
249    pub const fn pacing(&self) -> &PacingLimits {
250        &self.pacing
251    }
252
253    /// @docgen The breaker is fail-fast, so without waiting out its own pause a single refusal turns a whole zone unknown.
254    pub async fn acquire_patiently(
255        &self,
256        host: &str,
257        budget: Duration,
258    ) -> Result<RequestPermit, PausedHost> {
259        match self.acquire(host).await {
260            Ok(permit) => Ok(permit),
261            Err(paused) => {
262                let wait = paused.remaining_wait;
263                if wait.is_zero() || wait > budget {
264                    return Err(paused);
265                }
266                tokio::time::sleep(wait).await;
267                self.acquire(host).await
268            }
269        }
270    }
271
272    pub async fn acquire(&self, host: &str) -> Result<RequestPermit, PausedHost> {
273        let key = normalize_host(host);
274
275        let (limiter, slots, probe) = {
276            let mut hosts = self.hosts.lock().await;
277            let state = hosts
278                .entry(key.clone())
279                .or_insert_with(|| HostState::new(self.pacing.limit_for(&key)));
280
281            let now = Instant::now();
282            match state.breaker(now) {
283                BreakerState::Open => {
284                    let remaining = state
285                        .open_until
286                        .map_or(Duration::ZERO, |until| until.saturating_duration_since(now));
287                    return Err(PausedHost {
288                        host: key,
289                        remaining_wait: remaining,
290                        refusals: state.refusals,
291                    });
292                }
293                BreakerState::HalfOpen => {
294                    if state.is_probing {
295                        return Err(PausedHost {
296                            host: key,
297                            remaining_wait: Duration::ZERO,
298                            refusals: state.refusals,
299                        });
300                    }
301                    // @docgen A probe whose future is dropped never records an outcome, so the deadline lets the host recover on its own.
302                    state.is_probing = true;
303                    state.open_until = Some(now + PROBE_DEADLINE);
304                    (Arc::clone(&state.limiter), Arc::clone(&state.slots), true)
305                }
306                BreakerState::Closed => {
307                    (Arc::clone(&state.limiter), Arc::clone(&state.slots), false)
308                }
309            }
310        };
311
312        // @docgen The rate wait comes first so a global slot is never held idle while a slow registry's bucket refills.
313        if self.pacing.jitter.is_zero() {
314            limiter.until_ready().await;
315        } else {
316            limiter
317                .until_ready_with_jitter(Jitter::up_to(self.pacing.jitter))
318                .await;
319        }
320
321        let host_permit = slots
322            .acquire_owned()
323            .await
324            .map_err(|_| self.shutdown_pause(&key))?;
325        let global = Arc::clone(&self.global_slots)
326            .acquire_owned()
327            .await
328            .map_err(|_| self.shutdown_pause(&key))?;
329
330        Ok(RequestPermit {
331            host: key,
332            is_probe: probe,
333            _global_slot: global,
334            _host_slot: host_permit,
335        })
336    }
337
338    fn shutdown_pause(&self, host: &str) -> PausedHost {
339        PausedHost {
340            host: host.to_owned(),
341            remaining_wait: Duration::ZERO,
342            refusals: 0,
343        }
344    }
345
346    /// @docgen Concurrency climbs one step per run of successes but is cut hard on any refusal, so recovery cannot re-trigger a block.
347    pub async fn record_success(&self, host: &str) {
348        let key = normalize_host(host);
349        let mut hosts = self.hosts.lock().await;
350        let Some(state) = hosts.get_mut(&key) else {
351            return;
352        };
353
354        state.refusals = 0;
355        state.open_until = None;
356        state.is_probing = false;
357        state.successes = state.successes.saturating_add(1);
358
359        if state.successes >= self.pacing.recovery_threshold
360            && state.concurrency < state.max_concurrency
361        {
362            state.successes = 0;
363            state.concurrency = state
364                .concurrency
365                .saturating_add(1)
366                .min(state.max_concurrency);
367            state.slots.add_permits(1);
368            tracing::debug!(host = %key, concurrency = state.concurrency, "registry concurrency raised");
369        }
370    }
371
372    pub async fn record_refusal(&self, host: &str, refusal: &Refusal) -> Duration {
373        let key = normalize_host(host);
374        let mut hosts = self.hosts.lock().await;
375        let state = hosts
376            .entry(key.clone())
377            .or_insert_with(|| HostState::new(self.pacing.limit_for(&key)));
378
379        state.refusals = state.refusals.saturating_add(1);
380        state.successes = 0;
381        state.is_probing = false;
382
383        let wait = if refusal.has_stated_wait() {
384            refusal.base_wait()
385        } else {
386            let exponent = state.refusals.saturating_sub(1).min(6);
387            let ceiling = refusal
388                .base_wait()
389                .saturating_mul(1_u32 << exponent)
390                .min(self.pacing.max_backoff);
391            self.jitter.sample_up_to(ceiling)
392        };
393        let wait = wait.min(self.pacing.max_backoff);
394
395        if state.concurrency > 1 {
396            // @docgen Three-quarters alone has fixed points at 2 and 3, where most registries sit, so the cut would never fire.
397            let target = state
398                .concurrency
399                .saturating_mul(3)
400                .div_ceil(4)
401                .min(state.concurrency.saturating_sub(1))
402                .max(1);
403            let surplus = state.concurrency.saturating_sub(target);
404            if surplus > 0 {
405                // @docgen Tokio only forgets permits that are free, so the shortfall must carry rather than be assumed applied.
406                let forgotten = state.slots.forget_permits(surplus);
407                state.concurrency = state.concurrency.saturating_sub(forgotten);
408                tracing::debug!(host = %key, concurrency = state.concurrency, "registry concurrency cut");
409            }
410        }
411
412        let should_open =
413            matches!(refusal, Refusal::Blocked) || state.refusals >= self.pacing.refusal_threshold;
414        if should_open {
415            state.open_until = Some(Instant::now() + wait);
416            tracing::debug!(host = %key, refusals = state.refusals, ?wait, "registry paused");
417        }
418
419        wait
420    }
421
422    pub async fn breaker(&self, host: &str) -> BreakerState {
423        let key = normalize_host(host);
424        let hosts = self.hosts.lock().await;
425        hosts
426            .get(&key)
427            .map_or(BreakerState::Closed, |state| state.breaker(Instant::now()))
428    }
429
430    pub async fn paused(&self, host: &str) -> Option<PausedHost> {
431        let key = normalize_host(host);
432        let mut hosts = self.hosts.lock().await;
433        let state = hosts.get_mut(&key)?;
434        let open_until = state.open_until?;
435        let now = Instant::now();
436        if now >= open_until {
437            return None;
438        }
439        Some(PausedHost {
440            host: key,
441            remaining_wait: open_until.saturating_duration_since(now),
442            refusals: state.refusals,
443        })
444    }
445
446    pub async fn paused_hosts(&self) -> Vec<PausedHost> {
447        let now = Instant::now();
448        let hosts = self.hosts.lock().await;
449        let mut paused: Vec<PausedHost> = hosts
450            .iter()
451            .filter_map(|(host, state)| {
452                let open_until = state.open_until?;
453                (open_until > now).then(|| PausedHost {
454                    host: host.clone(),
455                    remaining_wait: open_until.saturating_duration_since(now),
456                    refusals: state.refusals,
457                })
458            })
459            .collect();
460        paused.sort_by(|a, b| a.host.cmp(&b.host));
461        paused
462    }
463
464    pub async fn host_concurrency(&self, host: &str) -> usize {
465        let key = normalize_host(host);
466        let hosts = self.hosts.lock().await;
467        hosts.get(&key).map_or(0, |state| state.concurrency)
468    }
469
470    /// @docgen A long sweep would otherwise keep a map entry for every endpoint it ever touched.
471    pub async fn prune_settled_hosts(&self) {
472        let now = Instant::now();
473        let mut hosts = self.hosts.lock().await;
474        hosts.retain(|_, state| {
475            state.refusals > 0
476                || state.is_probing
477                || state.concurrency < state.max_concurrency
478                || state.open_until.is_some_and(|until| until > now)
479        });
480    }
481}
482
483fn normalize_host(host: &str) -> String {
484    host.trim().trim_end_matches('.').to_lowercase()
485}
486
487#[cfg(test)]
488mod tests {
489    use super::*;
490
491    fn throttled(seconds: u64) -> Refusal {
492        Refusal::Throttled {
493            retry_after: Some(Duration::from_secs(seconds)),
494        }
495    }
496
497    const UNSTATED: Refusal = Refusal::Throttled { retry_after: None };
498
499    fn fast() -> PacingLimits {
500        PacingLimits {
501            jitter: Duration::ZERO,
502            ..PacingLimits::default()
503        }
504    }
505
506    #[tokio::test(start_paused = true)]
507    async fn a_lease_is_granted_and_released() {
508        let pacer = Pacer::new(fast());
509        let lease = pacer
510            .acquire("rdap.example.test")
511            .await
512            .expect("first lease");
513        assert_eq!(lease.host(), "rdap.example.test");
514        assert!(!lease.is_probe());
515        drop(lease);
516        assert!(pacer.acquire("rdap.example.test").await.is_ok());
517    }
518
519    #[tokio::test(start_paused = true)]
520    async fn hosts_are_keyed_case_and_dot_insensitively() {
521        let pacer = Pacer::new(fast());
522        pacer
523            .record_refusal("RDAP.Example.Test.", &throttled(30))
524            .await;
525        pacer
526            .record_refusal("rdap.example.test", &throttled(30))
527            .await;
528        pacer
529            .record_refusal("rdap.example.test", &throttled(30))
530            .await;
531        assert!(pacer.paused("rdap.example.test").await.is_some());
532    }
533
534    #[tokio::test(start_paused = true)]
535    async fn a_published_registry_starts_with_its_published_allowance() {
536        let pacer = Pacer::new(fast());
537        let _lease = pacer
538            .acquire("rdap.identitydigital.services")
539            .await
540            .unwrap();
541        assert_eq!(
542            pacer
543                .host_concurrency("rdap.identitydigital.services")
544                .await,
545            4
546        );
547    }
548
549    #[tokio::test(start_paused = true)]
550    async fn gentle_pacing_ignores_a_generous_published_allowance() {
551        let pacer = Pacer::new(PacingLimits {
552            jitter: Duration::ZERO,
553            ..PacingLimits::cautious()
554        });
555        let _lease = pacer
556            .acquire("rdap.identitydigital.services")
557            .await
558            .unwrap();
559        assert_eq!(
560            pacer
561                .host_concurrency("rdap.identitydigital.services")
562                .await,
563            CAUTIOUS_LIMIT.concurrency
564        );
565    }
566
567    #[tokio::test(start_paused = true)]
568    async fn the_breaker_stays_shut_until_the_threshold_is_reached() {
569        let pacer = Pacer::new(PacingLimits {
570            refusal_threshold: 3,
571            ..fast()
572        });
573        pacer.record_refusal("slow.test", &UNSTATED).await;
574        assert_eq!(pacer.breaker("slow.test").await, BreakerState::Closed);
575        pacer.record_refusal("slow.test", &UNSTATED).await;
576        assert_eq!(pacer.breaker("slow.test").await, BreakerState::Closed);
577        pacer.record_refusal("slow.test", &UNSTATED).await;
578        assert_eq!(pacer.breaker("slow.test").await, BreakerState::Open);
579    }
580
581    #[tokio::test(start_paused = true)]
582    async fn an_outright_block_opens_the_breaker_on_the_first_refusal() {
583        let pacer = Pacer::new(PacingLimits {
584            refusal_threshold: 99,
585            ..fast()
586        });
587        pacer
588            .record_refusal("blocked.test", &Refusal::Blocked)
589            .await;
590        assert_eq!(pacer.breaker("blocked.test").await, BreakerState::Open);
591    }
592
593    #[tokio::test(start_paused = true)]
594    async fn a_dropped_connection_counts_as_backpressure() {
595        let pacer = Pacer::new(PacingLimits {
596            refusal_threshold: 1,
597            ..fast()
598        });
599        pacer.record_refusal("quiet.test", &Refusal::Dropped).await;
600        assert_eq!(pacer.breaker("quiet.test").await, BreakerState::Open);
601    }
602
603    #[tokio::test(start_paused = true)]
604    async fn an_open_breaker_refuses_a_lease_instead_of_blocking() {
605        let pacer = Pacer::new(PacingLimits {
606            refusal_threshold: 1,
607            ..fast()
608        });
609        pacer.record_refusal("busy.test", &throttled(5)).await;
610
611        match pacer.acquire("busy.test").await {
612            Err(paused) => {
613                assert_eq!(paused.host, "busy.test");
614                assert!(paused.remaining_wait <= Duration::from_secs(5));
615            }
616            Ok(_) => panic!("an open breaker must refuse the lease"),
617        }
618    }
619
620    #[tokio::test(start_paused = true)]
621    async fn the_cooldown_lets_exactly_one_probe_through() {
622        let pacer = Pacer::new(PacingLimits {
623            refusal_threshold: 1,
624            ..fast()
625        });
626        pacer.record_refusal("recovering.test", &throttled(2)).await;
627        tokio::time::advance(Duration::from_secs(3)).await;
628
629        assert_eq!(
630            pacer.breaker("recovering.test").await,
631            BreakerState::HalfOpen
632        );
633        let probe = pacer
634            .acquire("recovering.test")
635            .await
636            .expect("probe allowed");
637        assert!(probe.is_probe());
638
639        assert!(
640            pacer.acquire("recovering.test").await.is_err(),
641            "only one probe may be in flight"
642        );
643    }
644
645    #[tokio::test(start_paused = true)]
646    async fn a_successful_probe_closes_the_breaker() {
647        let pacer = Pacer::new(PacingLimits {
648            refusal_threshold: 1,
649            ..fast()
650        });
651        pacer.record_refusal("healing.test", &throttled(2)).await;
652        tokio::time::advance(Duration::from_secs(3)).await;
653        let probe = pacer.acquire("healing.test").await.expect("probe");
654        drop(probe);
655
656        pacer.record_success("healing.test").await;
657        assert_eq!(pacer.breaker("healing.test").await, BreakerState::Closed);
658        assert!(pacer.acquire("healing.test").await.is_ok());
659    }
660
661    #[tokio::test(start_paused = true)]
662    async fn a_failed_probe_reopens_the_breaker() {
663        let pacer = Pacer::new(PacingLimits {
664            refusal_threshold: 1,
665            ..fast()
666        });
667        pacer.record_refusal("stubborn.test", &throttled(2)).await;
668        tokio::time::advance(Duration::from_secs(3)).await;
669        let probe = pacer.acquire("stubborn.test").await.expect("probe");
670        drop(probe);
671
672        pacer.record_refusal("stubborn.test", &throttled(4)).await;
673        assert_eq!(pacer.breaker("stubborn.test").await, BreakerState::Open);
674    }
675
676    #[tokio::test(start_paused = true)]
677    async fn a_stated_wait_is_honored_exactly() {
678        let pacer = Pacer::new(fast());
679        let wait = pacer.record_refusal("polite.test", &throttled(7)).await;
680        assert_eq!(wait, Duration::from_secs(7));
681    }
682
683    #[tokio::test(start_paused = true)]
684    async fn an_absurd_stated_wait_is_capped() {
685        let pacer = Pacer::new(fast());
686        let wait = pacer
687            .record_refusal("hostile.test", &throttled(86_400))
688            .await;
689        assert_eq!(wait, MAX_RETRY_AFTER);
690    }
691
692    #[tokio::test(start_paused = true)]
693    async fn an_unstated_wait_stays_inside_the_growing_ceiling() {
694        let pacer = Pacer::new(PacingLimits {
695            refusal_threshold: 99,
696            ..fast()
697        });
698        for _ in 0..12 {
699            let wait = pacer.record_refusal("steep.test", &UNSTATED).await;
700            assert!(
701                wait <= pacer.pacing().max_backoff,
702                "{wait:?} exceeded the cap"
703            );
704        }
705    }
706
707    #[tokio::test(start_paused = true)]
708    async fn a_refusal_cuts_the_allowance_and_success_earns_it_back() {
709        let pacer = Pacer::new(PacingLimits {
710            refusal_threshold: 99,
711            recovery_threshold: 2,
712            ..fast()
713        });
714        let host = "rdap.identitydigital.services";
715        let lease = pacer.acquire(host).await.unwrap();
716        drop(lease);
717        assert_eq!(pacer.host_concurrency(host).await, 4);
718
719        pacer.record_refusal(host, &UNSTATED).await;
720        assert_eq!(pacer.host_concurrency(host).await, 3);
721
722        for _ in 0..2 {
723            pacer.record_success(host).await;
724        }
725        assert_eq!(pacer.host_concurrency(host).await, 4);
726    }
727
728    #[tokio::test(start_paused = true)]
729    async fn a_refusal_still_cuts_a_host_that_starts_at_two() {
730        // @docgen Most registries start at the cautious limit of two, where a three-quarter rounding cut has a fixed point.
731        let pacer = Pacer::new(PacingLimits {
732            refusal_threshold: 99,
733            ..fast()
734        });
735        let host = "unpublished.test";
736        let lease = pacer.acquire(host).await.unwrap();
737        drop(lease);
738        assert_eq!(pacer.host_concurrency(host).await, 2);
739
740        pacer.record_refusal(host, &UNSTATED).await;
741        assert_eq!(
742            pacer.host_concurrency(host).await,
743            1,
744            "backpressure must reach a host that starts at the cautious limit"
745        );
746    }
747
748    #[tokio::test(start_paused = true)]
749    async fn the_allowance_never_climbs_past_where_it_started() {
750        let pacer = Pacer::new(PacingLimits {
751            recovery_threshold: 1,
752            ..fast()
753        });
754        let host = "rdap.identitydigital.services";
755        let lease = pacer.acquire(host).await.unwrap();
756        drop(lease);
757
758        for _ in 0..50 {
759            pacer.record_success(host).await;
760        }
761        assert_eq!(pacer.host_concurrency(host).await, 4);
762    }
763
764    #[tokio::test(start_paused = true)]
765    async fn paused_hosts_lists_only_the_registries_actually_on_hold() {
766        let pacer = Pacer::new(PacingLimits {
767            refusal_threshold: 1,
768            ..fast()
769        });
770        pacer.record_refusal("one.test", &throttled(10)).await;
771        pacer.record_success("two.test").await;
772
773        let paused = pacer.paused_hosts().await;
774        assert_eq!(paused.len(), 1);
775        assert_eq!(paused.first().map(|p| p.host.as_str()), Some("one.test"));
776    }
777
778    #[tokio::test(start_paused = true)]
779    async fn tidy_keeps_troubled_hosts_and_drops_settled_ones() {
780        let pacer = Pacer::new(PacingLimits {
781            refusal_threshold: 1,
782            ..fast()
783        });
784        let settled = pacer.acquire("calm.test").await.unwrap();
785        drop(settled);
786        pacer.record_success("calm.test").await;
787        pacer.record_refusal("angry.test", &throttled(60)).await;
788
789        pacer.prune_settled_hosts().await;
790        assert_eq!(pacer.host_concurrency("calm.test").await, 0);
791        assert!(pacer.paused("angry.test").await.is_some());
792    }
793
794    #[test]
795    fn full_jitter_draws_inside_the_window_and_actually_varies() {
796        let jitterer = JitterRng::new();
797        let ceiling = Duration::from_secs(10);
798        let draws: Vec<Duration> = (0..64).map(|_| jitterer.sample_up_to(ceiling)).collect();
799
800        assert!(draws.iter().all(|d| *d <= ceiling));
801        let unique = draws
802            .iter()
803            .collect::<std::collections::BTreeSet<_>>()
804            .len();
805        assert!(
806            unique > 32,
807            "jitter is not spreading: {unique} distinct draws"
808        );
809    }
810
811    #[test]
812    fn a_zero_window_yields_no_wait_rather_than_panicking() {
813        let jitterer = JitterRng::new();
814        assert_eq!(jitterer.sample_up_to(Duration::ZERO), Duration::ZERO);
815    }
816
817    #[test]
818    fn a_quota_survives_an_extremely_slow_published_limit() {
819        let quota = quota_for(HostLimit::per_minute(1, 1));
820        assert!(quota.burst_size().get() >= 1);
821    }
822}
reserve_core/limit/mod.rs

reserve_core/limit/
mod.rs
