Skip to main content

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}