rusty_time_core/server.rs
1//! Server-side policy: client tracking, rate limiting, and interleaved mode.
2//!
3//! All decisions, no I/O. The daemon owns sockets and NTS; this module owns
4//! *what the answer should be*, which is what makes it testable without a
5//! network and portable to wasm.
6//!
7//! The client key is generic on purpose: the core must not know what an IP
8//! address is (mission plan §4 — the core knows bytes and timestamps, never a
9//! product type). The daemon instantiates it with the peer address.
10
11use crate::ntp::NtpTimestamp;
12use std::hash::{Hash, Hasher};
13
14/// Rate-limit policy, mirroring chrony's `ratelimit interval/burst/leak`.
15#[derive(Clone, Copy, Debug, PartialEq)]
16pub struct RateLimitConfig {
17 /// log2 of the mean seconds between responses to one client. 3 = one
18 /// response per 8 s, chrony's default.
19 pub interval_log2: i8,
20 /// How many responses a client may take back to back after being quiet.
21 pub burst: u32,
22 /// Emit a Kiss-o'-Death to one dropped request in 2^leak_shift. Never 0:
23 /// answering *every* dropped request turns the limiter into the
24 /// amplifier it exists to prevent.
25 pub leak_shift: u8,
26 /// Ceiling on responses per second **across all clients**. 0 disables it.
27 ///
28 /// Per-client limiting alone is defeated by table churn: once the client
29 /// population exceeds the table, every request arrives from an address we
30 /// have forgotten, gets a fresh bucket, and is answered. TIMECORP S12b
31 /// (100k addresses into a 16k table) showed the reply ratio climbing back
32 /// to 100% for exactly that reason. A global bucket is the backstop that
33 /// bounds total output no matter how the address space is shuffled.
34 ///
35 /// Set high enough not to hinder a busy legitimate server; it exists to
36 /// bound the worst case, not to shape normal traffic.
37 pub global_rate_hz: f64,
38 /// Global burst allowance, in responses.
39 pub global_burst: f64,
40}
41
42impl Default for RateLimitConfig {
43 fn default() -> Self {
44 RateLimitConfig {
45 interval_log2: 3,
46 burst: 8,
47 leak_shift: 2,
48 global_rate_hz: 20_000.0,
49 global_burst: 40_000.0,
50 }
51 }
52}
53
54/// What the server should do with one request.
55#[derive(Clone, Copy, Debug, PartialEq, Eq)]
56pub enum Disposition {
57 /// Answer normally.
58 Respond,
59 /// Over the limit: answer with a Kiss-o'-Death RATE so a well-behaved
60 /// client backs off.
61 KissOfDeath,
62 /// Over the limit and not this client's turn for a KoD: say nothing.
63 Drop,
64}
65
66/// Mark a server response's timestamps so the two are distinguishable by their
67/// lowest bit: **receive has bit 0 set, transmit has it clear** (RFC 9769 and
68/// chrony's `ntp_core.c`).
69///
70/// Two things depend on this. A server can then recognise an interleaved
71/// request statelessly — an origin field with bit 0 set is echoing a *receive*
72/// timestamp — and receive can never accidentally equal transmit, which would
73/// make the mode ambiguous. The cost is the bottom bit of a 232-picosecond
74/// unit, far below any clock's resolution.
75pub fn mark_server_timestamps(receive: &mut NtpTimestamp, transmit: &mut NtpTimestamp) {
76 receive.0 |= 1;
77 transmit.0 &= !1;
78}
79
80/// Which timestamps a response should carry.
81#[derive(Clone, Copy, Debug, PartialEq, Eq)]
82pub enum ResponseMode {
83 /// The ordinary exchange: receive and transmit from *this* exchange.
84 Basic,
85 /// Interleaved: the reply carries **this** exchange's receive timestamp
86 /// (so the client can interleave again next time) together with the
87 /// *actual, post-send* transmit timestamp of the earlier response the
88 /// client named in its origin field. That transmit value is the whole
89 /// point — a basic reply must write its transmit field before the packet
90 /// leaves, so it can only ever be an estimate.
91 ///
92 /// Two field rules, both of which fail silently rather than loudly when
93 /// broken (RFC 9769; chrony `ntp_core.c` lines 1241/1251/1290, 2744-2754):
94 ///
95 /// * **origin** echoes the request's *receive* field, not its transmit —
96 /// that is how the client recognises an interleaved reply.
97 /// * **receive** is the current exchange's, not the previous one's.
98 /// Reporting the previous receive pairs the client's T1/T4 with a
99 /// mismatched T2/T3 and shows up as a delay inflated by exactly one
100 /// poll interval — measured against chrony as 4.009 s on a 4 s poll.
101 Interleaved {
102 /// The true transmit timestamp of the response the client's origin
103 /// field identifies.
104 prev_transmit: NtpTimestamp,
105 },
106}
107
108/// Per-client state: enough for rate limiting, interleaved mode and the MRU
109/// report, and no more — this is multiplied by every client that has ever
110/// spoken to us.
111#[derive(Clone, Copy, Debug)]
112pub struct ClientRecord {
113 pub last_seen: f64,
114 /// Token bucket level, in responses.
115 tokens: f64,
116 pub requests: u64,
117 pub responses: u64,
118 pub dropped: u64,
119 /// Dropped requests since the last Kiss-o'-Death, for deterministic leak.
120 drops_since_kod: u32,
121 /// T2 of the last request we accepted. [`NtpTimestamp::ZERO`] until there
122 /// has been one.
123 ///
124 /// Zero rather than `Option` deliberately: `NtpTimestamp` is a bare `u64`
125 /// with no niche, so `Option<NtpTimestamp>` is SIXTEEN bytes for eight
126 /// bytes of payload and every read tests a discriminant. An all-zero
127 /// timestamp is already how RFC 5905 spells "unset" on the wire, so the
128 /// sentinel is the protocol's own, not one invented here
129 /// (`rusty-compiler-leverage` B5: the one win in that table was a
130 /// representation change, not a bit trick).
131 pub last_receive: NtpTimestamp,
132 /// The true transmit timestamp of our last response, once the driver
133 /// reports it. [`NtpTimestamp::ZERO`] until then, which is why interleaved
134 /// mode cannot answer the very first request.
135 pub last_transmit: NtpTimestamp,
136 /// The receive timestamp we put in our last response. A client asks for
137 /// interleaved mode by echoing exactly this. [`NtpTimestamp::ZERO`] until
138 /// we have answered once.
139 pub last_receive_sent: NtpTimestamp,
140 /// Whether the last answered request was actually served interleaved.
141 ///
142 /// Distinct from "we *could* serve it interleaved": every client we have
143 /// answered once is capable, but only a client that asks is using it, and
144 /// an operator reading a client log needs the second fact.
145 pub interleaved_now: bool,
146}
147
148impl ClientRecord {
149 fn new(now: f64, burst: u32) -> Self {
150 ClientRecord {
151 last_seen: now,
152 tokens: burst as f64,
153 requests: 0,
154 responses: 0,
155 dropped: 0,
156 drops_since_kod: 0,
157 last_receive: NtpTimestamp::ZERO,
158 last_transmit: NtpTimestamp::ZERO,
159 last_receive_sent: NtpTimestamp::ZERO,
160 interleaved_now: false,
161 }
162 }
163}
164
165/// Aggregate counters for the `status.serverstats` op.
166#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
167pub struct ServerStats {
168 pub requests: u64,
169 pub responses: u64,
170 pub dropped_rate_limit: u64,
171 pub kiss_of_death: u64,
172 pub interleaved_responses: u64,
173 /// Requests refused before any per-client work (bad mode, too short).
174 pub refused: u64,
175 /// Clients evicted from the table because it was full.
176 pub evicted: u64,
177}
178
179/// End-of-list sentinel for the intrusive recency links.
180const NIL: u32 = u32::MAX;
181
182/// Odd 64-bit multiplier for the key mixer. Any odd constant with a good bit
183/// spread works; this one is xxHash's prime 1.
184const MIX: u64 = 0x9e37_79b1_85eb_ca87;
185
186/// A fast, **seeded** hasher for short keys.
187///
188/// The default `HashMap` hasher is SipHash-1-3, chosen for resistance to
189/// collision floods. That resistance is not optional here — the key is a
190/// client's source address, which an attacker picks — but SipHash's cost is
191/// out of proportion to a 4-to-17-byte key: with everything else in the client
192/// table fixed, callgrind still attributed ~32% of the server's per-request
193/// instructions to hashing one address once.
194///
195/// So this keeps the property and drops the price. The seed is drawn from the
196/// OS once per process via `RandomState`, exactly as SipHash's keys are, so an
197/// attacker cannot compute which addresses collide without first learning a
198/// secret they never see. What is given up is SipHash's *proof* against an
199/// adversary who somehow does learn the seed; what is kept is the practical
200/// defence, on a table that is additionally bounded to a fixed capacity with
201/// LRU eviction, so even a successful collision attack cannot grow a chain
202/// without bound.
203#[derive(Clone, Copy)]
204pub struct ClientHashBuilder {
205 seed: u64,
206}
207
208impl Default for ClientHashBuilder {
209 fn default() -> Self {
210 // `RUSTY_TIME_HASH_SEED` pins the seed. It exists for measurement: a
211 // random seed changes which keys collide, which moves the probe count,
212 // which makes an instruction-count harness reproducible only to about
213 // 0.002% instead of exactly. That is far below any effect worth acting
214 // on, but an instrument that is exact is worth more than one that is
215 // nearly exact, and the pin costs one environment read per table.
216 //
217 // It is emphatically NOT for production: a known seed is a known
218 // collision set, which is the property this hasher is seeded to deny.
219 if let Ok(pinned) = std::env::var("RUSTY_TIME_HASH_SEED")
220 && let Ok(seed) = pinned.parse::<u64>()
221 {
222 return ClientHashBuilder { seed };
223 }
224 // One OS-random draw per table, reusing std's entropy source rather
225 // than adding a dependency for it.
226 use std::hash::BuildHasher as _;
227 let seed = std::collections::hash_map::RandomState::new().hash_one(0xA5A5_5A5Au64);
228 ClientHashBuilder { seed }
229 }
230}
231
232impl std::hash::BuildHasher for ClientHashBuilder {
233 type Hasher = ClientHasher;
234 fn build_hasher(&self) -> ClientHasher {
235 ClientHasher { state: self.seed }
236 }
237}
238
239pub struct ClientHasher {
240 state: u64,
241}
242
243impl ClientHasher {
244 #[inline]
245 fn mix(&mut self, value: u64) {
246 self.state = (self.state ^ value).wrapping_mul(MIX);
247 }
248}
249
250impl Hasher for ClientHasher {
251 #[inline]
252 fn write(&mut self, bytes: &[u8]) {
253 // `as_chunks` rather than `chunks_exact(8)`: the width is a constant, so
254 // this yields `&[u8; 8]` directly and the fallible conversion in the
255 // loop disappears.
256 let (chunks, rest) = bytes.as_chunks::<8>();
257 for chunk in chunks {
258 self.mix(u64::from_le_bytes(*chunk));
259 }
260 if !rest.is_empty() {
261 let mut buf = [0u8; 8];
262 buf[..rest.len()].copy_from_slice(rest);
263 // Fold the length in so trailing zero bytes cannot alias a shorter
264 // key against a longer one.
265 self.mix(u64::from_le_bytes(buf) ^ (rest.len() as u64) << 56);
266 }
267 }
268
269 #[inline]
270 fn write_u8(&mut self, value: u8) {
271 self.mix(value as u64);
272 }
273
274 #[inline]
275 fn write_u32(&mut self, value: u32) {
276 self.mix(value as u64);
277 }
278
279 #[inline]
280 fn write_u64(&mut self, value: u64) {
281 self.mix(value);
282 }
283
284 #[inline]
285 fn write_usize(&mut self, value: usize) {
286 self.mix(value as u64);
287 }
288
289 #[inline]
290 fn finish(&self) -> u64 {
291 // splitmix64's finalizer: full avalanche in a handful of instructions,
292 // which is what stops near-adjacent addresses landing in near-adjacent
293 // buckets.
294 let mut z = self.state;
295 z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
296 z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
297 z ^ (z >> 31)
298 }
299}
300
301/// An open-addressed index from client key to slot number.
302///
303/// Replaces `HashMap<K, u32, ClientHashBuilder>`, measured on the churn arm
304/// (`HOT_PATH_EVICT`): hashbrown was **13.7M Ir, 18.1% of that arm** against
305/// 3.5M / 7.4% on the steady arm, which never removes a key and so priced the
306/// vein at a quarter of its size.
307///
308/// Four properties, all of them facts about what a client table IS rather than
309/// cleverness:
310///
311/// 1. **Fixed capacity, so no resize.** The table evicts its least recently
312/// seen client instead of growing, so the bucket array is allocated once and
313/// never rehashed. The whole load-factor/regrow path does not exist.
314/// 2. **Backward-shift deletion, so no tombstones.** A removal walks the rest
315/// of the probe run and pulls back any entry that may legally occupy the
316/// hole. Tombstones are the one failure mode an open-addressed table has
317/// under churn -- they accumulate until a probe degrades to a full scan --
318/// and this design has none. That is what the churn arm exists to check.
319/// 3. **Load factor <= 0.25**, so probe runs stay short. Not a free knob: at
320/// 0.5 this WON the steady arm (-789,020) and LOST under churn
321/// (+7,455,786), because backward-shift deletion walks the probe run. At
322/// 0.25 both arms win.
323/// 4. **The bucket does not store the key.** It is already in
324/// `slots[handle].key`, stored exactly once -- where `HashMap` had to keep
325/// its own copy. A bucket is two `u32`s, so the lower load factor costs
326/// almost nothing: the corpus priced the key-carrying version at +106
327/// bytes/client and this one at +2.
328///
329/// The low 32 bits of the hash live in the bucket, which is what lets the key
330/// stay out: a non-match is rejected on the fragment, so the dependent load
331/// into `slots` happens on a true match or a 1-in-4-billion collision.
332struct Index {
333 buckets: Box<[Bucket]>,
334 mask: usize,
335 len: usize,
336 hasher: ClientHashBuilder,
337}
338
339/// `handle == EMPTY` is a free bucket. There is deliberately no tombstone.
340#[derive(Clone, Copy)]
341struct Bucket {
342 /// Low 32 bits of the hash. Only two things are ever asked of it -- reject
343 /// a non-match before touching the key, and recover the ideal bucket during
344 /// backward-shift deletion (`hash & mask`) -- and the low word answers
345 /// both, because the mask cannot approach 2^32.
346 hash: u32,
347 handle: u32,
348}
349
350/// Sentinel for an empty bucket. Distinct from `NIL`'s role in the recency
351/// list only by where it is used; the value is the same and both mean "none".
352const EMPTY: u32 = u32::MAX;
353
354impl Index {
355 fn with_capacity(capacity: usize, hasher: ClientHashBuilder) -> Self {
356 let n = capacity.max(1).saturating_mul(4).next_power_of_two();
357 // Compare the MASK, not the count. `(u32::MAX as usize) + 1` overflows
358 // at compile time on a 32-bit `usize` -- wasm32 is a shipping target --
359 // and `arithmetic_overflow` is deny-by-default, so it is a build error
360 // rather than a runtime one. Phrased against `n - 1` it is always true
361 // on a 32-bit target (nothing can exceed `usize::MAX` buckets) and a
362 // real bound on a 64-bit one.
363 assert!(
364 n - 1 <= u32::MAX as usize,
365 "bucket count exceeds the 32-bit hash word"
366 );
367 Index {
368 buckets: vec![
369 Bucket {
370 hash: 0,
371 handle: EMPTY
372 };
373 n
374 ]
375 .into_boxed_slice(),
376 mask: n - 1,
377 len: 0,
378 hasher,
379 }
380 }
381
382 #[inline]
383 fn hash_of<K: Hash>(&self, key: &K) -> u32 {
384 use std::hash::BuildHasher as _;
385 // The finaliser avalanches high bits into low, so the low word
386 // discriminates as well as any other 32 bits of it.
387 self.hasher.hash_one(key) as u32
388 }
389
390 /// Bucket holding `key`, or `None`.
391 ///
392 /// `i & self.mask` is provably in range for any `i` because `x & m <= m`
393 /// and the array is `m + 1` long, so every bucket read here loses its
394 /// bounds check without a line of `unsafe`.
395 #[inline]
396 fn probe<K: Eq>(&self, hash: u32, key: &K, slots: &[Slot<K>]) -> Option<usize> {
397 let mut i = (hash as usize) & self.mask;
398 loop {
399 let b = self.buckets[i];
400 if b.handle == EMPTY {
401 return None;
402 }
403 if b.hash == hash
404 && slots
405 .get(b.handle as usize)
406 .is_some_and(|sl| sl.key == *key)
407 {
408 return Some(i);
409 }
410 i = (i + 1) & self.mask;
411 }
412 }
413
414 #[inline]
415 fn get<K: Eq + Hash>(&self, key: &K, slots: &[Slot<K>]) -> Option<u32> {
416 let h = self.hash_of(key);
417 let i = self.probe(h, key, slots)?;
418 Some(self.buckets[i].handle)
419 }
420
421 fn insert<K: Eq + Hash>(&mut self, key: &K, handle: u32, slots: &[Slot<K>]) {
422 let hash = self.hash_of(key);
423 if let Some(i) = self.probe(hash, key, slots) {
424 self.buckets[i].handle = handle;
425 return;
426 }
427 let mut i = (hash as usize) & self.mask;
428 while self.buckets[i].handle != EMPTY {
429 i = (i + 1) & self.mask;
430 }
431 self.buckets[i] = Bucket { hash, handle };
432 self.len += 1;
433 }
434
435 fn remove<K: Eq + Hash>(&mut self, key: &K, slots: &[Slot<K>]) {
436 let hash = self.hash_of(key);
437 let Some(mut hole) = self.probe(hash, key, slots) else {
438 return;
439 };
440 self.buckets[hole].handle = EMPTY;
441 self.len -= 1;
442
443 // Backward-shift deletion. Walk the remainder of the probe run and move
444 // an entry back into the hole whenever the hole lies between that
445 // entry's ideal bucket and where it sits. That preserves the only
446 // invariant `probe` needs -- every key reachable by a forward scan from
447 // its own ideal bucket -- which is why no tombstone is required.
448 //
449 // Distances are measured forward from `ideal` and masked, so wrapping
450 // needs no signed comparison.
451 let mut i = (hole + 1) & self.mask;
452 loop {
453 let b = self.buckets[i];
454 if b.handle == EMPTY {
455 break;
456 }
457 let ideal = (b.hash as usize) & self.mask;
458 let to_hole = hole.wrapping_sub(ideal) & self.mask;
459 let to_here = i.wrapping_sub(ideal) & self.mask;
460 if to_hole < to_here {
461 self.buckets[hole] = b;
462 self.buckets[i].handle = EMPTY;
463 hole = i;
464 }
465 i = (i + 1) & self.mask;
466 }
467 }
468
469 #[inline]
470 fn len(&self) -> usize {
471 self.len
472 }
473
474 #[inline]
475 fn is_empty(&self) -> bool {
476 self.len == 0
477 }
478}
479
480/// One client's storage: its key, its record, and its place in the recency
481/// list. Slots are stable — an index handed out stays valid until the client
482/// is evicted — which is what makes the list links safe as plain integers.
483struct Slot<K> {
484 key: K,
485 record: ClientRecord,
486 /// Toward the most-recently-used end.
487 prev: u32,
488 /// Toward the least-recently-used end.
489 next: u32,
490 /// Bumped every time this slot is handed to a new client, so a handle kept
491 /// across an eviction is detected rather than silently addressing whoever
492 /// took the slot over.
493 generation: u32,
494}
495
496/// A resolved position in the table.
497///
498/// One request touches the same client four times — admit, choose a response
499/// mode, record what was sent, then record the true transmit time after the
500/// packet leaves. Looking the key up each time cost four hashes of the same
501/// address, which callgrind put at ~51% of what the server does per request
502/// once the recency index was fixed. A handle turns the three follow-ups into
503/// array indexing.
504///
505/// It is deliberately not a raw index: the generation makes a handle that
506/// outlived its client detectably stale instead of quietly wrong.
507#[derive(Clone, Copy, Debug, PartialEq, Eq)]
508pub struct ClientHandle {
509 slot: u32,
510 generation: u32,
511}
512
513impl ClientHandle {
514 /// A handle that resolves to nothing — what a request rejected by the
515 /// global limiter gets, having never reached a per-client record.
516 pub const INVALID: ClientHandle = ClientHandle {
517 slot: NIL,
518 generation: 0,
519 };
520}
521
522/// Bounded most-recently-used client table.
523///
524/// The bound is the point: an unbounded map is a memory-exhaustion lever for
525/// anyone willing to spoof source addresses. When full, the least recently
526/// seen client is evicted — losing its interleaved state, which costs it one
527/// exchange, not correctness.
528///
529/// **Eviction is indexed, not scanned.** The obvious implementation — walk the
530/// map for the oldest `last_seen` — is O(capacity) per admission, and a public
531/// server facing more clients than the table holds evicts on nearly every
532/// packet. TIMECORP S12b (100k clients into a 16k table) went from "runs" to
533/// "does not finish" on exactly that, which is what the scenario is for. The
534/// `order` index makes it O(log n) *and* deterministic, where a `HashMap` scan
535/// depends on iteration order that differs between instances.
536pub struct ClientTable<K: Eq + Hash + Ord + Clone> {
537 /// Key to slot. The only hashed structure, and the reason a request costs
538 /// one hash instead of several.
539 index: Index,
540 /// Records in stable storage. Slots are handed out from `free` and never
541 /// move, which is what lets the ordering below be pointers rather than
542 /// comparisons.
543 slots: Vec<Slot<K>>,
544 free: Vec<u32>,
545 /// Ends of the intrusive most-recently-used list threaded through `slots`.
546 mru: u32,
547 lru: u32,
548 /// Global token bucket: level and the time it was last refilled.
549 global_tokens: f64,
550 global_last: Option<f64>,
551 /// Global drops since the last global Kiss-o'-Death.
552 global_drops_since_kod: u32,
553 capacity: usize,
554 config: RateLimitConfig,
555 /// Tokens a client earns per second — `2^-interval_log2`, precomputed.
556 ///
557 /// It derives only from `config`, which is fixed at construction, so
558 /// recomputing it per request was a `powi` call on the hot path for a
559 /// constant. Callgrind put the server's per-request cost at 3053 Ir; this
560 /// is one of the cheaper pieces of that, and it is free to remove.
561 refill_per_s: f64,
562 pub stats: ServerStats,
563}
564
565impl<K: Eq + Hash + Ord + Clone> ClientTable<K> {
566 pub fn new(capacity: usize, config: RateLimitConfig) -> Self {
567 ClientTable {
568 index: Index::with_capacity(capacity.max(1), ClientHashBuilder::default()),
569 slots: Vec::with_capacity(capacity.max(1)),
570 free: Vec::new(),
571 mru: NIL,
572 lru: NIL,
573 global_tokens: config.global_burst,
574 global_last: None,
575 global_drops_since_kod: 0,
576 capacity: capacity.max(1),
577 config,
578 refill_per_s: 2f64.powi(-(config.interval_log2 as i32)),
579 stats: ServerStats::default(),
580 }
581 }
582
583 /// Detach a slot from the recency list.
584 fn unlink(&mut self, i: u32) {
585 let (prev, next) = {
586 let slot = &self.slots[i as usize];
587 (slot.prev, slot.next)
588 };
589 if prev == NIL {
590 self.mru = next;
591 } else {
592 self.slots[prev as usize].next = next;
593 }
594 if next == NIL {
595 self.lru = prev;
596 } else {
597 self.slots[next as usize].prev = prev;
598 }
599 }
600
601 /// Put a detached slot at the most-recent end.
602 fn link_front(&mut self, i: u32) {
603 let old = self.mru;
604 {
605 let slot = &mut self.slots[i as usize];
606 slot.prev = NIL;
607 slot.next = old;
608 }
609 if old == NIL {
610 self.lru = i;
611 } else {
612 self.slots[old as usize].prev = i;
613 }
614 self.mru = i;
615 }
616
617 /// Move a client to the most-recent end of the eviction order.
618 ///
619 /// Six pointer writes, no hashing, no comparisons, no allocation. The
620 /// previous form kept a `BTreeSet<(seq, K)>` and did a remove plus an
621 /// insert — two O(log n) tree walks and two key clones — on *every*
622 /// request. Callgrind attributed 23% of the whole server hot path to that
623 /// tree's `search_tree`, for an index that is only ever read when the
624 /// table is full and something has to be evicted.
625 fn touch(&mut self, i: u32) {
626 if self.mru == i {
627 return; // already most-recent; the common case for a chatty client
628 }
629 self.unlink(i);
630 self.link_front(i);
631 }
632
633 /// The real per-client footprint: the record, the slot links that order
634 /// it, and the index entry that finds it.
635 ///
636 /// Reported by the type rather than estimated by the caller, because the
637 /// corpus quotes this number and an estimate drifts silently when the
638 /// structure changes. It did: the figure used to be `size_of::<ClientRecord>()`
639 /// alone, which stopped being the whole story the moment records moved
640 /// into slots.
641 pub fn bytes_per_client() -> usize {
642 // The index is open-addressed at a load factor of 0.25, so it holds
643 // FOUR buckets per client, not one entry. That is the honest figure and
644 // it is a real cost: the lower load factor is what makes backward-shift
645 // deletion cheap (measured -6.45% Ir on the churn arm against 0.5), and
646 // it is paid in memory. A caller sizing a table for a large population
647 // needs to see it here rather than discover it in RSS.
648 core::mem::size_of::<Slot<K>>() + 4 * core::mem::size_of::<Bucket>()
649 }
650
651 pub fn len(&self) -> usize {
652 self.index.len()
653 }
654
655 pub fn is_empty(&self) -> bool {
656 self.index.is_empty()
657 }
658
659 pub fn get(&self, key: &K) -> Option<&ClientRecord> {
660 let i = self.index.get(key, &self.slots)?;
661 Some(&self.slots[i as usize].record)
662 }
663
664 /// The MRU report: most recently seen first, at most `limit` entries.
665 ///
666 /// Walks the recency list, which is **already** in this order — `admit`
667 /// moves a client to the front and sets `last_seen` in the same breath, so
668 /// list order and descending `last_seen` are the same thing.
669 ///
670 /// The previous form cloned every record in the table into a `Vec`, sorted
671 /// all of them, and threw away all but `limit`. At the daemon's capacity
672 /// of 16384 that is ~1.5 MiB copied and an O(n log n) sort to answer a
673 /// ten-row status query. This is O(limit) and allocates once, for exactly
674 /// the rows returned.
675 pub fn most_recent(&self, limit: usize) -> Vec<(K, ClientRecord)> {
676 let mut out = Vec::with_capacity(limit.min(self.index.len()));
677 let mut at = self.mru;
678 while at != NIL && out.len() < limit {
679 let slot = &self.slots[at as usize];
680 out.push((slot.key.clone(), slot.record));
681 at = slot.next;
682 }
683 out
684 }
685
686 /// Drop the least recently seen client. O(1): it is the list's tail.
687 fn evict_one(&mut self) {
688 let victim = self.lru;
689 if victim == NIL {
690 return;
691 }
692 self.unlink(victim);
693 let ClientTable { index, slots, .. } = self;
694 // No clone: `index` is borrowed mutably and `slots` immutably, and they
695 // are different fields, so destructuring `self` lets both live at once.
696 index.remove(&slots[victim as usize].key, slots);
697 self.free.push(victim);
698 self.stats.evicted += 1;
699 }
700
701 /// Take a slot for a new client, reusing an evicted one where possible.
702 fn alloc_slot(&mut self, key: K, record: ClientRecord) -> u32 {
703 let i = match self.free.pop() {
704 Some(i) => {
705 let slot = &mut self.slots[i as usize];
706 slot.key = key;
707 slot.record = record;
708 slot.prev = NIL;
709 slot.next = NIL;
710 // New occupant, new generation: any handle still naming this
711 // slot from its previous client now fails to resolve.
712 slot.generation = slot.generation.wrapping_add(1);
713 i
714 }
715 None => {
716 self.slots.push(Slot {
717 key,
718 record,
719 prev: NIL,
720 next: NIL,
721 generation: 0,
722 });
723 (self.slots.len() - 1) as u32
724 }
725 };
726 self.link_front(i);
727 i
728 }
729
730 /// Turn a handle back into a slot, or `None` if it has gone stale.
731 fn resolve(&self, handle: ClientHandle) -> Option<usize> {
732 let i = handle.slot as usize;
733 let slot = self.slots.get(i)?;
734 (slot.generation == handle.generation).then_some(i)
735 }
736
737 fn handle_for(&self, slot: u32) -> ClientHandle {
738 ClientHandle {
739 slot,
740 generation: self.slots[slot as usize].generation,
741 }
742 }
743
744 /// Admit one request: refill the client's bucket, decide its fate, and
745 /// record it. `now` is monotonic seconds.
746 pub fn admit(&mut self, key: &K, now: f64) -> Disposition {
747 self.admit_handle(key, now).0
748 }
749
750 /// `admit`, also returning the handle that addresses this client, so the
751 /// rest of the request never has to hash the key again.
752 pub fn admit_handle(&mut self, key: &K, now: f64) -> (Disposition, ClientHandle) {
753 self.stats.requests += 1;
754
755 // Global ceiling first. It is checked before the per-client bucket so
756 // that a churned-address flood — which defeats per-client limiting by
757 // never reusing an address — still cannot make us answer without
758 // bound.
759 if self.config.global_rate_hz > 0.0 {
760 let last = self.global_last.unwrap_or(now);
761 let elapsed = (now - last).max(0.0);
762 self.global_tokens = (self.global_tokens + elapsed * self.config.global_rate_hz)
763 .min(self.config.global_burst);
764 self.global_last = Some(now);
765
766 if self.global_tokens < 1.0 {
767 self.stats.dropped_rate_limit += 1;
768 self.global_drops_since_kod += 1;
769 let period = 1u32 << self.config.leak_shift.min(16);
770 if self.global_drops_since_kod >= period {
771 self.global_drops_since_kod = 0;
772 self.stats.kiss_of_death += 1;
773 return (Disposition::KissOfDeath, ClientHandle::INVALID);
774 }
775 return (Disposition::Drop, ClientHandle::INVALID);
776 }
777 }
778
779 // One hash for the common case (a client we already know), two for a
780 // client we have never seen. The previous form hashed three times per
781 // request: `contains_key`, then `touch`'s `get_mut`, then a final
782 // `get_mut` to reach the record.
783 let slot = match self.index.get(key, &self.slots) {
784 Some(i) => {
785 self.touch(i);
786 i
787 }
788 None => {
789 if self.index.len() >= self.capacity {
790 self.evict_one();
791 }
792 let i = self.alloc_slot(key.clone(), ClientRecord::new(now, self.config.burst));
793 let ClientTable { index, slots, .. } = self;
794 index.insert(key, i, slots);
795 i
796 }
797 };
798
799 let config = self.config;
800 let rate = self.refill_per_s;
801 let handle = self.handle_for(slot);
802 let record = &mut self.slots[slot as usize].record;
803
804 // Refill: one token per 2^interval seconds since we last saw them.
805 let elapsed = (now - record.last_seen).max(0.0);
806 record.tokens = (record.tokens + elapsed * rate).min(config.burst as f64);
807 record.last_seen = now;
808 record.requests += 1;
809
810 if record.tokens >= 1.0 {
811 record.tokens -= 1.0;
812 record.responses += 1;
813 self.stats.responses += 1;
814 // Spend a global token only when a response is actually produced.
815 self.global_tokens -= 1.0;
816 return (Disposition::Respond, handle);
817 }
818
819 record.dropped += 1;
820 record.drops_since_kod += 1;
821 self.stats.dropped_rate_limit += 1;
822
823 // Deterministic leak: every 2^leak_shift-th drop gets a KoD. A
824 // deterministic rule is testable, and the client cannot tell the
825 // difference from a probabilistic one.
826 let period = 1u32 << config.leak_shift.min(16);
827 if record.drops_since_kod >= period {
828 record.drops_since_kod = 0;
829 self.stats.kiss_of_death += 1;
830 (Disposition::KissOfDeath, handle)
831 } else {
832 (Disposition::Drop, handle)
833 }
834 }
835
836 /// Decide basic vs interleaved for an admitted request.
837 ///
838 /// The client signals interleaved mode by setting its origin timestamp to
839 /// the *receive* timestamp we sent last time, rather than the transmit
840 /// timestamp. That is unforgeable in the useful sense: only a client that
841 /// actually saw our last response knows it.
842 pub fn response_mode(&mut self, key: &K, request_origin: NtpTimestamp) -> ResponseMode {
843 match self.index.get(key, &self.slots) {
844 Some(i) => {
845 let handle = self.handle_for(i);
846 self.response_mode_at(handle, request_origin)
847 }
848 None => ResponseMode::Basic,
849 }
850 }
851
852 /// `response_mode` addressed by handle — no hashing.
853 pub fn response_mode_at(
854 &mut self,
855 handle: ClientHandle,
856 request_origin: NtpTimestamp,
857 ) -> ResponseMode {
858 // One lookup, not two. The original read the record with `get` and then
859 // re-found the same entry with `get_mut` to store `interleaved_now` —
860 // a second hash of the same key on the per-request path, where hashing
861 // was measured at ~38% of all instructions.
862 let Some(i) = self.resolve(handle) else {
863 return ResponseMode::Basic;
864 };
865 let record = &mut self.slots[i].record;
866 let (sent_receive, prev_transmit) = (record.last_receive_sent, record.last_transmit);
867 // "Unset" is zero for both. Note the zero-origin rejection below is
868 // INDEPENDENT of this test and was already there: a client echoing an
869 // all-zero origin cannot match an unset record even if this guard were
870 // removed, which is what makes the sentinel safe here rather than
871 // merely convenient.
872 if sent_receive.is_zero() || prev_transmit.is_zero() {
873 return ResponseMode::Basic;
874 }
875 // The client names a specific earlier response by echoing the receive
876 // timestamp we reported for it. We keep one slot, so only the most
877 // recent qualifies; anything older falls back to basic rather than
878 // answering with a transmit timestamp from the wrong exchange.
879 let interleaved = request_origin == sent_receive && !request_origin.is_zero();
880 record.interleaved_now = interleaved;
881 if interleaved {
882 self.stats.interleaved_responses += 1;
883 ResponseMode::Interleaved { prev_transmit }
884 } else {
885 ResponseMode::Basic
886 }
887 }
888
889 /// Record what we received and what we told the client, after answering.
890 pub fn note_response(&mut self, key: &K, receive: NtpTimestamp, receive_sent: NtpTimestamp) {
891 if let Some(i) = self.index.get(key, &self.slots) {
892 let handle = self.handle_for(i);
893 self.note_response_at(handle, receive, receive_sent);
894 }
895 }
896
897 /// `note_response` addressed by handle — no hashing.
898 pub fn note_response_at(
899 &mut self,
900 handle: ClientHandle,
901 receive: NtpTimestamp,
902 receive_sent: NtpTimestamp,
903 ) {
904 if let Some(i) = self.resolve(handle) {
905 let record = &mut self.slots[i].record;
906 record.last_receive = receive;
907 record.last_receive_sent = receive_sent;
908 }
909 }
910
911 /// Record the true transmit timestamp of the response just sent. Called
912 /// after `send`, which is the whole point of interleaved mode — this is a
913 /// timestamp the basic exchange cannot report because the packet has not
914 /// left yet when its own transmit field is written.
915 pub fn note_transmit(&mut self, key: &K, transmit: NtpTimestamp) {
916 if let Some(i) = self.index.get(key, &self.slots) {
917 self.slots[i as usize].record.last_transmit = transmit;
918 }
919 }
920
921 /// `note_transmit` addressed by handle — no hashing.
922 ///
923 /// This is the one called after `send`, so a handle taken before the write
924 /// is used after it. The generation check is what makes that safe: if the
925 /// client was evicted in between, the update is dropped rather than landing
926 /// on whoever inherited the slot.
927 pub fn note_transmit_at(&mut self, handle: ClientHandle, transmit: NtpTimestamp) {
928 if let Some(i) = self.resolve(handle) {
929 self.slots[i].record.last_transmit = transmit;
930 }
931 }
932
933 pub fn note_refused(&mut self) {
934 self.stats.refused += 1;
935 }
936}
937
938#[cfg(test)]
939mod tests {
940 use super::*;
941
942 fn table() -> ClientTable<u32> {
943 ClientTable::new(
944 4,
945 RateLimitConfig {
946 interval_log2: 3, // one per 8 s
947 burst: 2,
948 leak_shift: 2, // KoD every 4th drop
949 // Global ceiling off: these tests are about per-client policy.
950 global_rate_hz: 0.0,
951 global_burst: 0.0,
952 },
953 )
954 }
955
956 #[test]
957 fn address_churn_cannot_defeat_the_limiter() {
958 // Every request arrives from an address never seen before, so the
959 // per-client bucket is always full and the table churns. Without a
960 // global ceiling this answers 100% of the flood — which is what
961 // TIMECORP S12b measured before the ceiling existed.
962 let mut t = ClientTable::<u32>::new(
963 1_024,
964 RateLimitConfig {
965 interval_log2: 3,
966 burst: 8,
967 leak_shift: 4,
968 global_rate_hz: 100.0,
969 global_burst: 100.0,
970 },
971 );
972 let mut answered = 0u64;
973 // 20_000 requests from 20_000 distinct addresses inside one second.
974 for client in 0..20_000u32 {
975 if t.admit(&client, client as f64 * 5e-5) == Disposition::Respond {
976 answered += 1;
977 }
978 }
979 // Ceiling is 100/s with a 100 burst, over ~1 s: ~200 at the very most.
980 assert!(
981 answered <= 250,
982 "address churn produced {answered} answers against a 100/s ceiling"
983 );
984 assert!(answered > 0, "the ceiling must not block everything");
985 }
986
987 #[test]
988 fn the_global_ceiling_refills_over_time() {
989 let mut t = ClientTable::<u32>::new(
990 16,
991 RateLimitConfig {
992 interval_log2: -10, // per-client effectively unlimited
993 burst: 1_000_000,
994 leak_shift: 8,
995 global_rate_hz: 10.0,
996 global_burst: 10.0,
997 },
998 );
999 let mut answered = 0;
1000 for i in 0..100u32 {
1001 if t.admit(&(i % 4), 0.0) == Disposition::Respond {
1002 answered += 1;
1003 }
1004 }
1005 assert!(answered <= 11, "burst exceeded: {answered}");
1006 // Ten seconds later the bucket has refilled.
1007 assert_eq!(t.admit(&0, 10.0), Disposition::Respond);
1008 }
1009
1010 #[test]
1011 fn burst_is_allowed_then_the_limiter_bites() {
1012 let mut t = table();
1013 assert_eq!(t.admit(&1, 0.0), Disposition::Respond);
1014 assert_eq!(t.admit(&1, 0.0), Disposition::Respond);
1015 // Burst of 2 spent; further immediate requests are not answered.
1016 assert!(matches!(
1017 t.admit(&1, 0.0),
1018 Disposition::Drop | Disposition::KissOfDeath
1019 ));
1020 assert_eq!(t.stats.responses, 2);
1021 }
1022
1023 #[test]
1024 fn tokens_refill_over_time() {
1025 let mut t = table();
1026 let _ = t.admit(&1, 0.0);
1027 let _ = t.admit(&1, 0.0);
1028 assert_ne!(t.admit(&1, 0.0), Disposition::Respond);
1029 // 8 s later exactly one token is back.
1030 assert_eq!(t.admit(&1, 8.0), Disposition::Respond);
1031 assert_ne!(t.admit(&1, 8.0), Disposition::Respond);
1032 }
1033
1034 #[test]
1035 fn kiss_of_death_leaks_at_the_configured_rate_not_every_drop() {
1036 let mut t = table();
1037 let _ = t.admit(&1, 0.0);
1038 let _ = t.admit(&1, 0.0);
1039 // 12 further requests in the same instant: all over the limit.
1040 let mut kods = 0;
1041 for _ in 0..12 {
1042 if t.admit(&1, 0.0) == Disposition::KissOfDeath {
1043 kods += 1;
1044 }
1045 }
1046 // leak_shift 2 => one KoD per 4 drops => 3 of 12.
1047 assert_eq!(kods, 3, "KoD leak rate wrong");
1048 // The point of leaking: we must answer far less than we are asked, or
1049 // the limiter is itself an amplifier.
1050 assert!(
1051 (kods as u64) < t.stats.dropped_rate_limit,
1052 "KoD count must stay below the drop count"
1053 );
1054 }
1055
1056 #[test]
1057 fn one_client_cannot_starve_another() {
1058 let mut t = table();
1059 for _ in 0..50 {
1060 let _ = t.admit(&1, 0.0);
1061 }
1062 // A quiet client still gets its full burst.
1063 assert_eq!(t.admit(&2, 0.0), Disposition::Respond);
1064 assert_eq!(t.admit(&2, 0.0), Disposition::Respond);
1065 }
1066
1067 #[test]
1068 fn table_is_bounded_and_evicts_the_stalest() {
1069 let mut t = table(); // capacity 4
1070 for client in 0..4u32 {
1071 let _ = t.admit(&client, client as f64);
1072 }
1073 assert_eq!(t.len(), 4);
1074 // A fifth client evicts client 0, the least recently seen.
1075 let _ = t.admit(&99, 10.0);
1076 assert_eq!(t.len(), 4, "table exceeded its bound");
1077 assert!(t.get(&0).is_none(), "stalest client was not evicted");
1078 assert!(t.get(&99).is_some());
1079 assert_eq!(t.stats.evicted, 1);
1080 }
1081
1082 #[test]
1083 fn interleaved_requires_the_client_to_echo_our_receive_timestamp() {
1084 let mut t = table();
1085 let _ = t.admit(&1, 0.0);
1086 let rx1 = NtpTimestamp::from_unix(1_756_224_000, 0);
1087
1088 // First exchange: nothing to interleave with yet.
1089 assert_eq!(
1090 t.response_mode(&1, NtpTimestamp(0x1111)),
1091 ResponseMode::Basic
1092 );
1093 t.note_response(&1, rx1, rx1);
1094 let tx1 = NtpTimestamp::from_unix(1_756_224_000, 500);
1095 t.note_transmit(&1, tx1);
1096
1097 // Second exchange, client echoes our receive timestamp: interleaved,
1098 // and it gets the true transmit of the response it named.
1099 let _ = t.admit(&1, 8.0);
1100 match t.response_mode(&1, rx1) {
1101 ResponseMode::Interleaved { prev_transmit } => {
1102 assert_eq!(prev_transmit, tx1);
1103 }
1104 other => panic!("expected interleaved, got {other:?}"),
1105 }
1106 assert_eq!(t.stats.interleaved_responses, 1);
1107 }
1108
1109 #[test]
1110 fn interleaved_flag_tracks_use_not_capability() {
1111 let mut t = table();
1112 let rx1 = NtpTimestamp::from_unix(1_756_224_000, 0);
1113 let _ = t.admit(&1, 0.0);
1114 t.note_response(&1, rx1, rx1);
1115 t.note_transmit(&1, rx1);
1116
1117 // Capable now, but this client has never asked.
1118 let _ = t.admit(&1, 8.0);
1119 let _ = t.response_mode(&1, NtpTimestamp::ZERO);
1120 assert!(
1121 !t.get(&1).expect("record").interleaved_now,
1122 "a client that never asked must not be reported as using interleaved"
1123 );
1124
1125 // Now it asks.
1126 let _ = t.admit(&1, 16.0);
1127 let _ = t.response_mode(&1, rx1);
1128 assert!(t.get(&1).expect("record").interleaved_now);
1129
1130 // And stops asking again.
1131 let _ = t.admit(&1, 24.0);
1132 let _ = t.response_mode(&1, NtpTimestamp::ZERO);
1133 assert!(!t.get(&1).expect("record").interleaved_now);
1134 }
1135
1136 #[test]
1137 fn a_client_echoing_the_wrong_value_gets_basic_mode() {
1138 let mut t = table();
1139 let _ = t.admit(&1, 0.0);
1140 let rx1 = NtpTimestamp::from_unix(1_756_224_000, 0);
1141 t.note_response(&1, rx1, rx1);
1142 t.note_transmit(&1, NtpTimestamp::from_unix(1_756_224_000, 500));
1143
1144 let _ = t.admit(&1, 8.0);
1145 // Some other value — an off-path guess — must not unlock interleaved.
1146 assert_eq!(
1147 t.response_mode(&1, NtpTimestamp(0xDEAD_BEEF)),
1148 ResponseMode::Basic
1149 );
1150 // Nor may a zero origin.
1151 assert_eq!(t.response_mode(&1, NtpTimestamp::ZERO), ResponseMode::Basic);
1152 }
1153
1154 #[test]
1155 fn eviction_downgrades_to_basic_rather_than_lying() {
1156 let mut t = table(); // capacity 4
1157 let rx = NtpTimestamp::from_unix(1_756_224_000, 0);
1158 let _ = t.admit(&1, 0.0);
1159 t.note_response(&1, rx, rx);
1160 t.note_transmit(&1, rx);
1161
1162 // Push client 1 out.
1163 for c in 10..15u32 {
1164 let _ = t.admit(&c, 100.0 + c as f64);
1165 }
1166 assert!(t.get(&1).is_none());
1167 // Its next request must be answered in basic mode, not with another
1168 // client's timestamps.
1169 let _ = t.admit(&1, 200.0);
1170 assert_eq!(t.response_mode(&1, rx), ResponseMode::Basic);
1171 }
1172
1173 #[test]
1174 fn heavy_client_churn_stays_tractable() {
1175 // A public server sees far more addresses than its table holds, so
1176 // eviction runs on nearly every packet. A scan-based eviction is
1177 // O(capacity) each time and this test does not finish; the indexed one
1178 // is O(log n). The assertion is a *count*, not a duration: every
1179 // admission must do a bounded amount of index work, which shows up as
1180 // the table never exceeding capacity while churning far past it.
1181 let capacity = 4_096;
1182 let mut t = ClientTable::<u32>::new(capacity, RateLimitConfig::default());
1183 let churn = 200_000u32;
1184 for client in 0..churn {
1185 let _ = t.admit(&client, client as f64 * 0.001);
1186 }
1187 assert_eq!(t.len(), capacity, "table must sit exactly at capacity");
1188 assert_eq!(
1189 t.stats.evicted,
1190 (churn as u64) - capacity as u64,
1191 "every client past capacity must have cost exactly one eviction"
1192 );
1193 // True LRU: the survivors are the most recent `capacity` clients.
1194 assert!(t.get(&(churn - 1)).is_some(), "newest client was evicted");
1195 assert!(t.get(&0).is_none(), "oldest client survived");
1196 }
1197
1198 #[test]
1199 fn eviction_is_true_lru_not_insertion_order() {
1200 let mut t = ClientTable::<u32>::new(3, RateLimitConfig::default());
1201 let _ = t.admit(&1, 0.0);
1202 let _ = t.admit(&2, 1.0);
1203 let _ = t.admit(&3, 2.0);
1204 // Touch client 1 so it is no longer the stalest.
1205 let _ = t.admit(&1, 3.0);
1206 // Inserting a fourth must evict client 2, not client 1.
1207 let _ = t.admit(&4, 4.0);
1208 assert!(t.get(&1).is_some(), "recently used client was evicted");
1209 assert!(t.get(&2).is_none(), "stalest client should have gone");
1210 assert!(t.get(&3).is_some());
1211 assert!(t.get(&4).is_some());
1212 }
1213
1214 #[test]
1215 fn mru_report_is_ordered_and_bounded() {
1216 let mut t = ClientTable::<u32>::new(16, RateLimitConfig::default());
1217 for c in 0..10u32 {
1218 let _ = t.admit(&c, c as f64);
1219 }
1220 let mru = t.most_recent(3);
1221 assert_eq!(mru.len(), 3);
1222 assert_eq!(mru[0].0, 9, "most recent first");
1223 assert_eq!(mru[2].0, 7);
1224 }
1225}