alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::collections::HashMap;
2use std::hash::Hash;
3use std::net::IpAddr;
4use std::sync::{Arc, Mutex};
5use std::time::Duration;
6
7use knot_types::UnixMicros;
8
9const MAX_TRACKED_PEERS: usize = 100_000;
10
11const MAX_PACED_KEYS: usize = 4_096;
12
13const SWEEP_INTERVAL_MICROS: u64 = 1_000_000;
14
15knot_types::scalar_newtype! {
16 pub struct Burst(u32);
17 pub struct RefillMicros(u64);
18 pub struct PerPeerInflight(usize);
19 pub struct GlobalInflight(usize);
20}
21
22#[derive(Debug, Clone, Copy, PartialEq, Eq)]
23pub struct RateLimit {
24 pub burst: Burst,
25 pub refill: RefillMicros,
26}
27
28impl RateLimit {
29 pub const fn interval(self) -> RefillMicros {
30 match self.refill.get() {
31 0 => RefillMicros::new(1),
32 micros => RefillMicros::new(micros),
33 }
34 }
35}
36
37#[derive(Debug, Clone, Copy)]
38pub struct LimitConfig {
39 pub rate: Option<RateLimit>,
40 pub per_peer_inflight: Option<PerPeerInflight>,
41 pub global_inflight: Option<GlobalInflight>,
42}
43
44impl Default for LimitConfig {
45 fn default() -> Self {
46 Self {
47 rate: Some(RateLimit {
48 burst: Burst::new(20),
49 refill: RefillMicros::new(100_000),
50 }),
51 per_peer_inflight: Some(PerPeerInflight::new(8)),
52 global_inflight: Some(GlobalInflight::new(64)),
53 }
54 }
55}
56
57impl LimitConfig {
58 pub const fn per_peer_only(per_peer_inflight: PerPeerInflight) -> Self {
59 Self {
60 rate: None,
61 per_peer_inflight: Some(per_peer_inflight),
62 global_inflight: None,
63 }
64 }
65
66 pub const fn unmetered() -> Self {
67 Self {
68 rate: None,
69 per_peer_inflight: None,
70 global_inflight: None,
71 }
72 }
73}
74
75struct Bucket {
76 tokens: u32,
77 last_refill: UnixMicros,
78}
79
80impl Bucket {
81 fn new(rate: RateLimit, now: UnixMicros) -> Self {
82 Self {
83 tokens: rate.burst.get(),
84 last_refill: now,
85 }
86 }
87
88 fn tokens_at(&self, rate: RateLimit, now: UnixMicros) -> u32 {
89 let elapsed = now.get().saturating_sub(self.last_refill.get());
90 let gained = (elapsed / rate.interval().get()).min(u64::from(rate.burst.get())) as u32;
91 self.tokens.saturating_add(gained).min(rate.burst.get())
92 }
93
94 fn replenish(&mut self, rate: RateLimit, now: UnixMicros) -> bool {
95 if now.get().saturating_sub(self.last_refill.get()) >= rate.interval().get() {
96 self.tokens = self.tokens_at(rate, now);
97 self.last_refill = now;
98 }
99 self.tokens > 0
100 }
101
102 fn full(&self, rate: RateLimit, now: UnixMicros) -> bool {
103 self.tokens_at(rate, now) >= rate.burst.get()
104 }
105}
106
107struct PeerState {
108 bucket: Option<Bucket>,
109 inflight: usize,
110}
111
112impl PeerState {
113 fn forgettable(&self) -> bool {
114 self.inflight == 0 && self.bucket.is_none()
115 }
116
117 fn worth_tracking(&self, rate: Option<RateLimit>, now: UnixMicros) -> bool {
118 match (self.inflight, &self.bucket, rate) {
119 (0, Some(bucket), Some(rate)) => !bucket.full(rate, now),
120 (0, _, _) => false,
121 _ => true,
122 }
123 }
124}
125
126const CONCENTRATED_REFUSALS: u32 = 1_024;
127
128#[derive(Default)]
129struct RefusalMajority {
130 peer: Option<IpAddr>,
131 votes: u32,
132 reported: bool,
133}
134
135impl RefusalMajority {
136 fn observe(&mut self, peer: IpAddr) -> Option<IpAddr> {
137 match (self.peer == Some(peer), self.votes) {
138 (true, _) => self.votes = self.votes.saturating_add(1),
139 (false, 0) => {
140 self.peer = Some(peer);
141 self.votes = 1;
142 }
143 (false, _) => self.votes -= 1,
144 }
145 let crossed = self.votes >= CONCENTRATED_REFUSALS && !self.reported;
146 self.reported |= crossed;
147 crossed.then_some(peer)
148 }
149}
150
151struct Inner {
152 peers: HashMap<Option<IpAddr>, PeerState>,
153 global_inflight: usize,
154 last_sweep: UnixMicros,
155 refusal_majority: RefusalMajority,
156}
157
158impl Inner {
159 fn has_room_for(
160 &mut self,
161 peer: Option<IpAddr>,
162 rate: Option<RateLimit>,
163 now: UnixMicros,
164 ) -> bool {
165 if rate.is_none() || self.peers.contains_key(&peer) || self.peers.len() < MAX_TRACKED_PEERS
166 {
167 return true;
168 }
169 if now.get().saturating_sub(self.last_sweep.get()) >= SWEEP_INTERVAL_MICROS {
170 self.last_sweep = now;
171 self.peers
172 .retain(|_, state| state.worth_tracking(rate, now));
173 }
174 self.peers.len() < MAX_TRACKED_PEERS
175 }
176}
177
178pub struct PreAuthLimiter {
179 config: LimitConfig,
180 inner: Mutex<Inner>,
181}
182
183#[derive(Debug, Clone, Copy, PartialEq, Eq)]
184pub enum Refusal {
185 RateLimited,
186 Saturated,
187}
188
189impl Default for PreAuthLimiter {
190 fn default() -> Self {
191 Self::with_config(LimitConfig::default())
192 }
193}
194
195impl PreAuthLimiter {
196 pub fn with_config(config: LimitConfig) -> Self {
197 Self {
198 config,
199 inner: Mutex::new(Inner {
200 peers: HashMap::new(),
201 global_inflight: 0,
202 last_sweep: UnixMicros::new(0),
203 refusal_majority: RefusalMajority::default(),
204 }),
205 }
206 }
207
208 fn lock(&self) -> std::sync::MutexGuard<'_, Inner> {
209 self.inner
210 .lock()
211 .unwrap_or_else(|poisoned| poisoned.into_inner())
212 }
213
214 pub fn admit(
215 self: &Arc<Self>,
216 peer: Option<IpAddr>,
217 now: UnixMicros,
218 ) -> Result<AdmitGuard, Refusal> {
219 let mut inner = self.lock();
220 let rate = self.config.rate;
221 let over_global = self
222 .config
223 .global_inflight
224 .is_some_and(|limit| inner.global_inflight >= limit.get());
225
226 let per_peer = self.config.per_peer_inflight;
227 let decision = match inner.has_room_for(peer, rate, now) {
228 false => Err(Refusal::Saturated),
229 true => {
230 let state = inner.peers.entry(peer).or_insert_with(|| PeerState {
231 bucket: rate.map(|rate| Bucket::new(rate, now)),
232 inflight: 0,
233 });
234 let ready = match (rate, state.bucket.as_mut()) {
235 (Some(rate), Some(bucket)) => bucket.replenish(rate, now),
236 _ => true,
237 };
238 let over_peer = per_peer.is_some_and(|limit| state.inflight >= limit.get());
239 match (ready, over_peer || over_global) {
240 (false, _) => Err(Refusal::RateLimited),
241 (_, true) => Err(Refusal::Saturated),
242 (true, false) => {
243 if let Some(bucket) = state.bucket.as_mut() {
244 bucket.tokens -= 1;
245 }
246 state.inflight += 1;
247 Ok(())
248 }
249 }
250 }
251 };
252 match decision {
253 Err(refusal) => {
254 if inner.peers.get(&peer).is_some_and(PeerState::forgettable) {
255 inner.peers.remove(&peer);
256 }
257 let concentrated = peer.and_then(|peer| inner.refusal_majority.observe(peer));
258 drop(inner);
259 if let Some(peer) = concentrated {
260 tracing::warn!(
261 %peer,
262 "one address has taken {CONCENTRATED_REFUSALS} more of the pre-authentication limiter's refusals than every other address combined. If this address is a proxy, set xrpc.trusted_proxy_header to the header it forwards the client address in and add the address to xrpc.trusted_proxies, since every client behind a proxy will share its one rate-limit bucket. This warning reports the first such address only."
263 );
264 }
265 Err(refusal)
266 }
267 Ok(()) => {
268 inner.global_inflight += 1;
269 Ok(AdmitGuard {
270 limiter: Arc::clone(self),
271 peer,
272 })
273 }
274 }
275 }
276
277 fn leave(&self, peer: Option<IpAddr>) {
278 let mut inner = self.lock();
279 inner.global_inflight = inner.global_inflight.saturating_sub(1);
280 let Some(state) = inner.peers.get_mut(&peer) else {
281 return;
282 };
283 state.inflight = state.inflight.saturating_sub(1);
284 if state.forgettable() {
285 inner.peers.remove(&peer);
286 }
287 }
288
289 fn repay(&self, peer: Option<IpAddr>) {
290 let Some(rate) = self.config.rate else {
291 return;
292 };
293 let mut inner = self.lock();
294 if let Some(bucket) = inner
295 .peers
296 .get_mut(&peer)
297 .and_then(|state| state.bucket.as_mut())
298 {
299 bucket.tokens = bucket.tokens.saturating_add(1).min(rate.burst.get());
300 }
301 }
302}
303
304pub struct AdmitGuard {
305 limiter: Arc<PreAuthLimiter>,
306 peer: Option<IpAddr>,
307}
308
309impl AdmitGuard {
310 pub fn refund(self) {
311 self.limiter.repay(self.peer);
312 }
313}
314
315impl Drop for AdmitGuard {
316 fn drop(&mut self) {
317 self.limiter.leave(self.peer);
318 }
319}
320
321#[derive(Debug, Clone, PartialEq, Eq, Hash)]
322pub struct HostKey(String);
323
324impl HostKey {
325 pub fn new(host: &str) -> Self {
326 Self(host.to_ascii_lowercase())
327 }
328}
329
330#[derive(Debug, Clone, PartialEq, Eq, Hash)]
331pub struct SubjectKey(String);
332
333impl SubjectKey {
334 pub fn new(subject: &str) -> Self {
335 Self(subject.to_ascii_lowercase())
336 }
337}
338
339struct Booked<K> {
340 turns: HashMap<K, UnixMicros>,
341 last_sweep: UnixMicros,
342}
343
344impl<K: Eq + Hash> Booked<K> {
345 fn has_room_for(&mut self, key: &K, now: UnixMicros) -> bool {
346 if self.turns.contains_key(key) || self.turns.len() < MAX_PACED_KEYS {
347 return true;
348 }
349 if now.get().saturating_sub(self.last_sweep.get()) >= SWEEP_INTERVAL_MICROS {
350 self.last_sweep = now;
351 self.turns.retain(|_, until| until.get() > now.get());
352 }
353 self.turns.len() < MAX_PACED_KEYS
354 }
355}
356
357pub struct Pacer<K> {
358 rate: RateLimit,
359 inner: Mutex<Booked<K>>,
360}
361
362pub type HostPacer = Pacer<HostKey>;
363pub type SubjectPacer = Pacer<SubjectKey>;
364pub type PeerPacer = Pacer<IpAddr>;
365
366impl<K: Clone + Eq + Hash> Pacer<K> {
367 pub fn new(rate: RateLimit) -> Self {
368 Self {
369 rate,
370 inner: Mutex::new(Booked {
371 turns: HashMap::new(),
372 last_sweep: UnixMicros::new(0),
373 }),
374 }
375 }
376
377 pub fn reserve(&self, key: &K, now: UnixMicros) -> Duration {
378 let mut booked = self.lock();
379 match booked.has_room_for(key, now) {
380 false => Duration::from_micros(self.rate.interval().get()),
381 true => {
382 let wait = self.wait_for(&booked, key, now);
383 self.claim_turn(&mut booked, key, now);
384 wait
385 }
386 }
387 }
388
389 pub fn reserve_now(&self, key: &K, now: UnixMicros) -> bool {
390 let mut booked = self.lock();
391 match booked.has_room_for(key, now) {
392 false => false,
393 true => match self.wait_for(&booked, key, now).is_zero() {
394 false => false,
395 true => {
396 self.claim_turn(&mut booked, key, now);
397 true
398 }
399 },
400 }
401 }
402
403 fn wait_for(&self, booked: &Booked<K>, key: &K, now: UnixMicros) -> Duration {
404 let tolerance = self
405 .rate
406 .interval()
407 .get()
408 .saturating_mul(u64::from(self.rate.burst.get().saturating_sub(1)));
409 Duration::from_micros(
410 self.turn(booked, key, now)
411 .get()
412 .saturating_sub(tolerance)
413 .saturating_sub(now.get()),
414 )
415 }
416
417 fn claim_turn(&self, booked: &mut Booked<K>, key: &K, now: UnixMicros) {
418 let until = self
419 .turn(booked, key, now)
420 .get()
421 .saturating_add(self.rate.interval().get());
422 booked.turns.insert(key.clone(), UnixMicros::new(until));
423 }
424
425 fn turn(&self, booked: &Booked<K>, key: &K, now: UnixMicros) -> UnixMicros {
426 booked
427 .turns
428 .get(key)
429 .map_or(now, |until| UnixMicros::new(until.get().max(now.get())))
430 }
431
432 fn lock(&self) -> std::sync::MutexGuard<'_, Booked<K>> {
433 self.inner
434 .lock()
435 .unwrap_or_else(|poisoned| poisoned.into_inner())
436 }
437}
438
439#[cfg(test)]
440mod tests {
441 use super::*;
442 use std::net::Ipv4Addr;
443
444 fn limiter(config: LimitConfig) -> Arc<PreAuthLimiter> {
445 Arc::new(PreAuthLimiter::with_config(config))
446 }
447
448 fn rate(burst: u32, refill_micros: u64) -> Option<RateLimit> {
449 Some(RateLimit {
450 burst: Burst::new(burst),
451 refill: RefillMicros::new(refill_micros),
452 })
453 }
454
455 fn peer(last: u8) -> Option<IpAddr> {
456 Some(ip(last))
457 }
458
459 fn rotating(index: u64) -> Option<IpAddr> {
460 let octets = (index as u32).to_be_bytes();
461 Some(IpAddr::V4(Ipv4Addr::new(
462 1, octets[1], octets[2], octets[3],
463 )))
464 }
465
466 fn at(micros: u64) -> UnixMicros {
467 UnixMicros::new(micros)
468 }
469
470 fn tracked(limiter: &Arc<PreAuthLimiter>) -> usize {
471 limiter.lock().peers.len()
472 }
473
474 fn ip(last: u8) -> IpAddr {
475 IpAddr::V4(Ipv4Addr::new(127, 0, 0, last))
476 }
477
478 #[test]
479 fn only_an_address_taking_most_of_the_refusals_is_reported_and_only_once() {
480 let reported = |rounds, peer: fn(u32) -> u8| {
481 let mut majority = RefusalMajority::default();
482 (0..rounds)
483 .filter_map(|round| majority.observe(ip(peer(round))))
484 .collect::<Vec<IpAddr>>()
485 };
486 [
487 (CONCENTRATED_REFUSALS * 3, (|round| (round % 4 == 0) as u8) as fn(u32) -> u8, vec![ip(0)],
488 "the warning reports that address once, since three refusals in every four come from it"),
489 (CONCENTRATED_REFUSALS * 8, |round| (round % 251) as u8, vec![],
490 "a knot under scattered load doesn't have a proxy to point the operator at"),
491 (CONCENTRATED_REFUSALS - 1, |_| 1, vec![],
492 "a burst under the threshold is ordinary rate limiting and doesn't point at a proxy"),
493 (CONCENTRATED_REFUSALS, |_| 1, vec![ip(1)],
494 "the threshold itself is where an address earns the warning"),
495 ]
496 .into_iter()
497 .for_each(|(rounds, peer, expected, why)| {
498 assert_eq!(reported(rounds, peer), expected, "{why}");
499 });
500 }
501
502 #[test]
503 fn a_rate_budget_refuses_a_flood_refills_over_time_and_stays_per_peer() {
504 let limiter = limiter(LimitConfig {
505 rate: rate(2, 1_000),
506 per_peer_inflight: Some(PerPeerInflight::new(100)),
507 global_inflight: Some(GlobalInflight::new(100)),
508 });
509 assert!(limiter.admit(peer(1), at(0)).is_ok());
510 assert!(limiter.admit(peer(1), at(0)).is_ok());
511 assert_eq!(
512 limiter.admit(peer(1), at(0)).err(),
513 Some(Refusal::RateLimited),
514 "the limiter refuses a third request inside the same instant even when nothing is in flight"
515 );
516 assert!(
517 limiter.admit(peer(2), at(0)).is_ok(),
518 "a second peer keeps its own rate budget"
519 );
520 assert_eq!(
521 limiter.admit(peer(1), at(600)).err(),
522 Some(Refusal::RateLimited)
523 );
524 assert!(
525 limiter.admit(peer(1), at(1_000)).is_ok(),
526 "a refusal partway through the interval mustn't reset the clock the refill measures from"
527 );
528 }
529
530 #[test]
531 fn inflight_shedding_frees_on_drop_and_spends_no_rate_budget() {
532 let limiter = limiter(LimitConfig {
533 rate: rate(4, 1_000_000),
534 per_peer_inflight: Some(PerPeerInflight::new(1)),
535 global_inflight: Some(GlobalInflight::new(2)),
536 });
537 let held = limiter
538 .admit(peer(1), at(0))
539 .expect("the limiter admits the first request");
540 let other = limiter
541 .admit(peer(2), at(0))
542 .expect("a second peer fills the global budget");
543 assert_eq!(
544 limiter.admit(peer(1), at(0)).err(),
545 Some(Refusal::Saturated),
546 "the limiter sheds a second concurrent request from one peer"
547 );
548 assert_eq!(
549 limiter.admit(peer(3), at(0)).err(),
550 Some(Refusal::Saturated),
551 "the limiter sheds a third peer once the global in-flight limit is reached"
552 );
553 drop(other);
554 drop(
555 limiter
556 .admit(peer(3), at(0))
557 .expect("freeing a global slot admits the peer that was shed"),
558 );
559 drop(held);
560 (0..50).for_each(|_| {
561 limiter
562 .admit(peer(1), at(0))
563 .expect("a refunded admission must leave the full budget available")
564 .refund();
565 });
566 (0..3).for_each(|_| {
567 limiter
568 .admit(peer(1), at(0))
569 .expect("a shed request mustn't spend the rate budget it never used");
570 });
571 assert_eq!(
572 limiter.admit(peer(1), at(0)).err(),
573 Some(Refusal::RateLimited),
574 "a dropped guard without a refund keeps its token spent"
575 );
576 }
577
578 #[test]
579 fn a_budget_without_a_rate_never_rate_limits_and_keeps_no_idle_state() {
580 let overflowing = MAX_TRACKED_PEERS as u64 + 1_000;
581
582 let per_peer = limiter(LimitConfig::per_peer_only(PerPeerInflight::new(2)));
583 (0..1_000).for_each(|_| {
584 per_peer
585 .admit(peer(1), at(0))
586 .expect("a budget with no rate has nothing for a sequential flood to exhaust");
587 });
588 let concurrent: Vec<AdmitGuard> = (0..2)
589 .map(|_| {
590 per_peer
591 .admit(peer(1), at(0))
592 .expect("the limiter admits both concurrent operations from one peer")
593 })
594 .collect();
595 assert_eq!(
596 per_peer.admit(peer(1), at(0)).err(),
597 Some(Refusal::Saturated),
598 "only the per-peer count refuses a request in this budget"
599 );
600 drop(concurrent);
601 let held: Vec<AdmitGuard> = (0..overflowing)
602 .map(|index| {
603 per_peer
604 .admit(rotating(index), at(0))
605 .expect("a budget that keeps no idle state has no table to overflow")
606 })
607 .collect();
608 assert_eq!(tracked(&per_peer), held.len());
609 drop(held);
610 assert_eq!(
611 tracked(&per_peer),
612 0,
613 "with no tokens to remember, an idle peer leaves no entry, \
614 so address rotation mustn't fill the table and start shedding newcomers"
615 );
616
617 let global = limiter(LimitConfig {
618 rate: None,
619 per_peer_inflight: None,
620 global_inflight: Some(GlobalInflight::new(1)),
621 });
622 let _saturating = global
623 .admit(peer(1), at(0))
624 .expect("the limiter admits the first peer");
625 (0..overflowing).for_each(|index| {
626 assert_eq!(
627 global.admit(rotating(index), at(0)).err(),
628 Some(Refusal::Saturated)
629 );
630 });
631 assert_eq!(
632 tracked(&global),
633 1,
634 "a refusal returns no guard, so an entry it left behind would never be freed, \
635 and the sweep that bounds the table only reclaims idle rate state"
636 );
637
638 let unmetered = limiter(LimitConfig::unmetered());
639 let guards: Vec<AdmitGuard> = (0..512)
640 .map(|_| {
641 unmetered
642 .admit(peer(1), at(0))
643 .expect("an unmetered budget admits every request from every peer")
644 })
645 .collect();
646 drop(guards);
647 assert_eq!(tracked(&unmetered), 0);
648 }
649
650 #[test]
651 fn a_rate_budget_bounds_its_peer_map_and_sweeps_at_most_once_per_interval() {
652 let limiter = limiter(LimitConfig {
653 rate: rate(1, 1_000),
654 per_peer_inflight: Some(PerPeerInflight::new(8)),
655 global_inflight: Some(GlobalInflight::new(64)),
656 });
657 (0..MAX_TRACKED_PEERS as u64).for_each(|index| {
658 let _ = limiter.admit(rotating(index), at(0));
659 });
660 assert_eq!(
661 limiter
662 .admit(peer(201), at(SWEEP_INTERVAL_MICROS - 1))
663 .err(),
664 Some(Refusal::Saturated),
665 "inside the sweep interval a full map sheds unseen peers without rescanning"
666 );
667 assert!(
668 limiter.admit(peer(202), at(SWEEP_INTERVAL_MICROS)).is_ok(),
669 "once the interval elapses the sweep evicts replenished entries and admits the newcomer"
670 );
671 (0..(MAX_TRACKED_PEERS as u64 + 50_000)).for_each(|index| {
672 let _ = limiter.admit(
673 rotating(MAX_TRACKED_PEERS as u64 + index),
674 at(SWEEP_INTERVAL_MICROS + index),
675 );
676 });
677 let tracked = tracked(&limiter);
678 assert!(
679 tracked <= MAX_TRACKED_PEERS,
680 "a flood of distinct source addresses mustn't grow the peer map past its limit, saw {tracked}"
681 );
682 }
683
684 #[test]
685 fn a_pacer_spends_a_hosts_burst_at_once_then_spaces_it_and_leaves_every_other_host_alone() {
686 let pacer = HostPacer::new(RateLimit {
687 burst: Burst::new(3),
688 refill: RefillMicros::new(100),
689 });
690 let plc = HostKey::new("plc.directory");
691 let pds = HostKey::new("PDS.Nel.Pet");
692 let waits: Vec<u128> = (0..5)
693 .map(|_| pacer.reserve(&plc, at(0)).as_micros())
694 .collect();
695 assert_eq!(
696 waits,
697 vec![0, 0, 0, 100, 200],
698 "a cold host takes its whole burst without waiting. Every visit after that is one \
699 refill interval further out"
700 );
701 assert_eq!(
702 pacer.reserve(&pds, at(0)),
703 Duration::ZERO,
704 "a knot whose accounts spread over many PDSes must fill at the sum of their rates, \
705 so one busy host mustn't delay another"
706 );
707 assert_eq!(
708 HostKey::new("PDS.Nel.Pet"),
709 HostKey::new("pds.nel.pet"),
710 "a PDS endpoint written in mixed case is the same host and must share its schedule"
711 );
712 assert_eq!(
713 pacer.reserve(&plc, at(10_000)),
714 Duration::ZERO,
715 "a host the caller hasn't visited since its last turn is due immediately"
716 );
717 }
718
719 #[test]
720 fn a_turn_that_isnt_due_is_refused_without_pushing_the_schedule_further_out() {
721 let pacer = SubjectPacer::new(RateLimit {
722 burst: Burst::new(1),
723 refill: RefillMicros::new(1_000),
724 });
725 let nel = SubjectKey::new("did:plc:nel");
726 assert!(
727 pacer.reserve_now(&nel, at(0)),
728 "a subject the caller hasn't read takes its turn straight away"
729 );
730 assert!(
731 !pacer.reserve_now(&nel, at(500)),
732 "a caller that mustn't wait is refused inside the interval"
733 );
734 assert!(
735 !pacer.reserve_now(&nel, at(999)),
736 "refusals must leave the booking alone, or whoever keeps trying pushes the turn \
737 further out every time and the knot never reads the subject again"
738 );
739 assert!(pacer.reserve_now(&nel, at(1_000)));
740 assert_eq!(
741 SubjectKey::new("DID:PLC:NEL"),
742 SubjectKey::new("did:plc:nel"),
743 "a DID a client typed in mixed case is the same account and shares its schedule"
744 );
745 }
746
747 #[test]
748 fn a_flood_of_distinct_hosts_doesnt_grow_the_schedule_past_its_limit_or_delay_a_newcomer() {
749 let pacer = HostPacer::new(RateLimit {
750 burst: Burst::new(1),
751 refill: RefillMicros::new(1_000_000),
752 });
753 (0..MAX_PACED_KEYS as u64 + 5_000).for_each(|index| {
754 let _ = pacer.reserve(&HostKey::new(&format!("{index}.nel.pet")), at(0));
755 });
756 let tracked = pacer.lock().turns.len();
757 assert!(
758 tracked <= MAX_PACED_KEYS,
759 "a grant set spread over more did:web hosts than the schedule can track mustn't \
760 grow it past its limit, saw {tracked}"
761 );
762 assert_eq!(
763 pacer.reserve(&HostKey::new("plc.directory"), at(0)),
764 Duration::from_micros(1_000_000),
765 "a host the full schedule can't track waits one refill interval, so a caller that \
766 outgrows the schedule slows itself down"
767 );
768 let settled = at(SWEEP_INTERVAL_MICROS + 2_000_000);
769 assert_eq!(
770 pacer.reserve(&HostKey::new("plc.directory"), settled),
771 Duration::ZERO
772 );
773 assert_eq!(
774 pacer.lock().turns.len(),
775 1,
776 "the sweep reclaims the schedule and tracks the newcomer once every booked turn \
777 has passed"
778 );
779 }
780
781 #[test]
782 fn the_majority_counter_sees_a_refusal_from_a_full_peer_table() {
783 let limiter = limiter(LimitConfig {
784 rate: rate(1, 1_000),
785 per_peer_inflight: Some(PerPeerInflight::new(8)),
786 global_inflight: Some(GlobalInflight::new(64)),
787 });
788 (0..MAX_TRACKED_PEERS as u64).for_each(|index| {
789 let _ = limiter.admit(rotating(index), at(0));
790 });
791 assert_eq!(
792 limiter
793 .admit(peer(201), at(SWEEP_INTERVAL_MICROS - 1))
794 .err(),
795 Some(Refusal::Saturated)
796 );
797 let inner = limiter.lock();
798 assert_eq!(
799 (inner.refusal_majority.peer, inner.refusal_majority.votes),
800 (peer(201), 1),
801 "that refusal has to reach the counter like every other refusal, since a peer shed by a full table is what the warning most needs to report"
802 );
803 }
804}