dig-peer-selector 0.1.3

Self-optimizing peer-selection middleware for the DIG Node — the pure decision + learning layer between dig-download and the discovery layers (dig-dht / dig-gossip / dig-nat). Maintains a dynamic peer registry, learns each peer's throughput/RTT/reliability + saturation point + a relayed penalty from the REAL measured outcome of every transfer (measured-only, non-gameable), and ranks a candidate subset per content request to minimize P99 latency without thundering-herd. No user-facing knobs; re-uses the transport-verified peer_id = SHA-256(TLS SPKI DER).
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
//! The dynamic peer registry: `peer_id -> PeerEntry`, fed by churn + DHT candidates, bounded with
//! lowest-value eviction (SPEC.md §2).
//!
//! The registry is **fed, not self-discovered** (SPEC §2.3): gossip pool churn upserts/marks entries,
//! and DHT candidates passed to `select` upsert cold entries. A reconnecting peer keeps its learned
//! history (its quality is retained across a disconnect — SPEC §2.3). The registry is bounded (a
//! resource limit, not a behavior knob) and evicts the lowest-value entries first, preferring never to
//! evict a connected peer or one with a range in flight — but the bound is a HARD limit (§2.5): once
//! every remaining entry is connected or in-flight, capacity enforcement falls back to evicting the
//! lowest-value non-connected entry regardless of its in-flight count, so a dispatch that never settles
//! cannot pin a slot forever (#179 finding 1). `in_flight` itself is also TTL-reclaimed
//! ([`Registry::reclaim_stale_in_flight`]) so a genuinely-abandoned dispatch stops being counted busy
//! at all, independent of capacity pressure.

use std::collections::HashMap;

use dig_dht::CandidateAddr;
use dig_nat::{PeerId, TraversalKind};

use crate::quality::PeerQuality;
use crate::types::{Candidate, Provenance};

/// How long an unsettled dispatch may hold `in_flight > 0` before it is presumed abandoned and
/// reclaimed (#179 finding 1). A `record_outcome` that never arrives (a peer that goes silent after
/// being dispatched to) must not pin the entry as permanently unevictable (SPEC §2.5, §5.3). Chosen
/// generously relative to any realistic single-range transfer deadline so a merely-slow-but-alive
/// transfer is never reclaimed out from under it.
pub const DISPATCH_TTL_SECS: u64 = 300;

/// One registry entry: identity, reachability, live-link class, provenance, and the learned quality
/// model (SPEC §2.1). Field names are the API contract; private bookkeeping fields are added below.
#[derive(Debug, Clone)]
pub struct PeerEntry {
    /// The registry key: `peer_id = SHA-256(TLS SPKI DER)` (SPEC §1.3, re-used from `dig-nat`).
    pub peer_id: PeerId,
    /// Dial candidates, byte-compatible with `dig-dht` provider records / the L7 `dig.getPeers` shape.
    pub addresses: Vec<CandidateAddr>,
    /// The `dig-nat` tier of the live link, if connected — observational only (SPEC §2.1, §7.3).
    pub connection_class: Option<TraversalKind>,
    /// How this peer entered the registry (strongest-evidence provenance kept — SPEC §2.2).
    pub provenance: Provenance,
    /// The learned quality model (SPEC §3) — MEASURED-only.
    pub quality: PeerQuality,
    /// Unix seconds when first registered.
    pub first_seen: u64,
    /// Unix seconds of the most recent measured outcome, if any.
    pub last_outcome_at: Option<u64>,
    /// Whether the peer currently has a live pool link (SPEC §2.3). A disconnected peer is retained
    /// (with its quality) but is a weaker eviction-protection candidate.
    pub connected: bool,
    /// Whether the peer is banned — ineligible for selection until re-added (SPEC §9.4).
    pub banned: bool,
    /// Unix seconds when `quality.in_flight` most recently rose from `0`, or `None` while it is `0`
    /// (SPEC §2.5, §5.3). Drives TTL reclamation: a dispatch whose outcome never arrives must not pin
    /// a capacity slot forever (#179 finding 1) — `Registry::reclaim_stale_in_flight` zeros `in_flight`
    /// once this timestamp ages past [`DISPATCH_TTL_SECS`].
    pub(crate) in_flight_since: Option<u64>,
}

impl PeerEntry {
    /// A fresh, cold entry for `peer_id` first seen at `now` with the given provenance (SPEC §2.3,
    /// §3.5). Its quality is cold, so its first selection is exploratory (SPEC §4.4-E).
    pub fn cold(peer_id: PeerId, provenance: Provenance, now: u64) -> Self {
        PeerEntry {
            peer_id,
            addresses: Vec::new(),
            connection_class: None,
            provenance,
            quality: PeerQuality::cold(),
            first_seen: now,
            last_outcome_at: None,
            connected: false,
            banned: false,
            in_flight_since: None,
        }
    }

    /// Whether this peer is eligible for selection: not banned (SPEC §9.4). (Disconnected peers stay
    /// eligible — they may be reachable via a fresh dial; only a ban excludes.)
    pub fn is_eligible(&self) -> bool {
        !self.banned
    }

    /// Whether this entry may be evicted under NORMAL pressure: never a connected peer or one with a
    /// range in flight (SPEC §2.5). `Registry::enforce_capacity` prefers this, but falls back to
    /// [`PeerEntry::is_force_evictable`] when nothing satisfies it, so the capacity bound always holds.
    pub fn is_evictable(&self) -> bool {
        !self.connected && self.quality.in_flight == 0
    }

    /// Whether this entry may be evicted as a LAST RESORT when no entry satisfies [`Self::is_evictable`]
    /// (#179 finding 1): still never a currently-connected peer (a live link is never silently dropped),
    /// but a disconnected peer's stuck `in_flight` no longer protects it. This is what keeps the
    /// registry's capacity bound a hard limit even when every disconnected entry has an unsettled
    /// dispatch (SPEC §2.5 "the capacity bound is always enforceable").
    pub fn is_force_evictable(&self) -> bool {
        !self.connected
    }

    /// The most-direct dial address (lowest [`dig_dht::AddressKind::rank`]), if any (SPEC §2.1). The
    /// registry sorts most-direct-first on demand and never assumes wire order.
    pub fn preferred_address(&self) -> Option<&CandidateAddr> {
        self.addresses
            .iter()
            .filter(|a| a.kind.is_dialable())
            .min_by_key(|a| a.kind.rank())
    }

    /// An eviction *value* score (higher = more valuable = evict later). Combines staleness (age of
    /// the last measured outcome) with learned quality + confidence, per SPEC §2.5. A never-measured,
    /// long-idle peer scores lowest and is shed first.
    pub(crate) fn eviction_value(&self, now: u64) -> f64 {
        // Quality contribution: measured throughput weighted by confidence + reliability.
        let tput = self.quality.throughput.value().unwrap_or(0.0);
        let conf = self.quality.confidence();
        let rel = self.quality.reliability.rate().unwrap_or(0.0);
        let quality_value = tput * conf * (0.5 + 0.5 * rel);
        // Staleness penalty: older last-outcome => lower value. A peer never measured is maximally
        // stale (uses first_seen as the reference so a fresh cold peer isn't instantly evicted).
        let reference = self.last_outcome_at.unwrap_or(self.first_seen);
        let age = now.saturating_sub(reference) as f64;
        // Decay value by age (a soft, monotone penalty; a day-old idle peer is heavily discounted).
        let recency = 1.0 / (1.0 + age / 3600.0);
        quality_value * recency + recency // + recency so ties break toward the fresher peer
    }
}

/// Outcome of feeding a churn/candidate event into the registry (for the engine + tests to assert on).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FeedResult {
    /// A brand-new (cold) entry was created.
    Inserted,
    /// An existing entry was updated (quality preserved).
    Updated,
    /// An existing entry was marked disconnected (its quality retained).
    MarkedDisconnected,
    /// The event referred to a peer that was not present (a no-op removal).
    Absent,
}

/// The peer registry (SPEC §2). Owns the `peer_id -> PeerEntry` map and enforces the capacity bound.
#[derive(Debug, Default)]
pub struct Registry {
    entries: HashMap<PeerId, PeerEntry>,
    capacity: usize,
    /// `peer_id`s removed since the last [`Registry::drain_evicted`] call — by capacity eviction
    /// (§2.5) or explicit [`Registry::remove`] (#179 finding 2). The engine drains this after every
    /// registry-mutating call and prunes its own side maps (`last_selected`, `dispatched`) for the
    /// same keys, so a peer that leaves the registry leaves no residue elsewhere.
    evicted: Vec<PeerId>,
}

impl Registry {
    /// A registry bounded at `capacity` entries (a resource limit, SPEC §2.5).
    pub fn new(capacity: usize) -> Self {
        Registry {
            entries: HashMap::new(),
            capacity: capacity.max(1),
            evicted: Vec::new(),
        }
    }

    /// The number of entries currently held.
    pub fn len(&self) -> usize {
        self.entries.len()
    }

    /// Drain the set of `peer_id`s removed from the registry since the last drain (#179 finding 2):
    /// by capacity eviction (§2.5) or explicit [`Registry::remove`]. The caller (the engine) MUST
    /// prune any per-peer side state it keeps outside the registry for every id returned here, so
    /// nothing outlives the peer's registry entry. Returns an empty `Vec` when nothing left.
    pub fn drain_evicted(&mut self) -> Vec<PeerId> {
        std::mem::take(&mut self.evicted)
    }

    /// Whether the registry is empty.
    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    /// A read-only view of an entry.
    pub fn get(&self, peer: &PeerId) -> Option<&PeerEntry> {
        self.entries.get(peer)
    }

    /// A mutable view of an entry (engine-internal).
    pub(crate) fn get_mut(&mut self, peer: &PeerId) -> Option<&mut PeerEntry> {
        self.entries.get_mut(peer)
    }

    /// Iterate all entries (read-only).
    pub fn iter(&self) -> impl Iterator<Item = &PeerEntry> {
        self.entries.values()
    }

    /// Upsert a peer from a **candidate** (DHT/manual feed, SPEC §2.3). A new peer is created cold
    /// (its first selection is exploratory); an existing peer keeps its learned quality but refreshes
    /// its addresses/class and, if the new provenance is stronger evidence, its provenance. Returns
    /// whether an entry was created or updated. Enforces the capacity bound after an insert.
    pub fn upsert_candidate(
        &mut self,
        candidate: &Candidate,
        provenance: Provenance,
        now: u64,
    ) -> FeedResult {
        let result = match self.entries.get_mut(&candidate.peer_id) {
            Some(existing) => {
                if !candidate.addresses.is_empty() {
                    existing.addresses = candidate.addresses.clone();
                }
                if candidate.class.is_some() {
                    existing.connection_class = candidate.class;
                }
                if provenance.evidence() > existing.provenance.evidence() {
                    existing.provenance = provenance;
                }
                FeedResult::Updated
            }
            None => {
                let mut entry = PeerEntry::cold(candidate.peer_id, provenance, now);
                entry.addresses = candidate.addresses.clone();
                entry.connection_class = candidate.class;
                self.seed_class_prior(&mut entry);
                self.entries.insert(candidate.peer_id, entry);
                FeedResult::Inserted
            }
        };
        if matches!(result, FeedResult::Inserted) {
            self.enforce_capacity(now);
        }
        result
    }

    /// Mark a peer **connected** (a live pool link from `PeerAdded`), upserting a cold entry if new
    /// and preserving any existing learned quality (SPEC §2.3). Provenance `Gossip`.
    pub fn mark_connected(&mut self, peer: PeerId, provenance: Provenance, now: u64) -> FeedResult {
        match self.entries.get_mut(&peer) {
            Some(existing) => {
                existing.connected = true;
                existing.banned = false;
                if provenance.evidence() > existing.provenance.evidence() {
                    existing.provenance = provenance;
                }
                FeedResult::Updated
            }
            None => {
                let mut entry = PeerEntry::cold(peer, provenance, now);
                entry.connected = true;
                self.entries.insert(peer, entry);
                self.enforce_capacity(now);
                FeedResult::Inserted
            }
        }
    }

    /// Mark a peer **disconnected** (`PeerRemoved`) — its entry AND learned quality are RETAINED so a
    /// reconnect resumes from history (SPEC §2.3). A `banned` flag makes it ineligible until re-added
    /// (SPEC §9.4). Returns `Absent` if the peer was unknown.
    pub fn mark_disconnected(&mut self, peer: &PeerId, banned: bool) -> FeedResult {
        match self.entries.get_mut(peer) {
            Some(existing) => {
                existing.connected = false;
                if banned {
                    existing.banned = true;
                }
                FeedResult::MarkedDisconnected
            }
            None => FeedResult::Absent,
        }
    }

    /// Attach / update a live peer's connection class from `dig-nat` (SPEC §5.4, §7.3). Upserts a cold
    /// entry (provenance `Nat`) if the peer is not yet known, seeding its class prior.
    pub fn set_connection_class(
        &mut self,
        peer: PeerId,
        class: TraversalKind,
        now: u64,
    ) -> FeedResult {
        match self.entries.get_mut(&peer) {
            Some(existing) => {
                existing.connection_class = Some(class);
                FeedResult::Updated
            }
            None => {
                let mut entry = PeerEntry::cold(peer, Provenance::Nat, now);
                entry.connection_class = Some(class);
                self.seed_class_prior(&mut entry);
                self.entries.insert(peer, entry);
                self.enforce_capacity(now);
                FeedResult::Inserted
            }
        }
    }

    /// Explicitly remove a peer (rare; churn usually drives this — SPEC §5.4). A peer with a range in
    /// flight is NOT removed (removing it would corrupt in-flight accounting). Records the removal for
    /// [`Registry::drain_evicted`] (#179 finding 2) so the engine prunes its side maps.
    pub fn remove(&mut self, peer: &PeerId) -> FeedResult {
        match self.entries.get(peer) {
            Some(e) if e.quality.in_flight > 0 => FeedResult::Updated, // keep — busy
            Some(_) => {
                self.entries.remove(peer);
                self.evicted.push(*peer);
                FeedResult::MarkedDisconnected
            }
            None => FeedResult::Absent,
        }
    }

    /// Record that `count` ranges were dispatched to `peer` (in-flight bump, SPEC §5.3). Stamps
    /// `in_flight_since` when `in_flight` rises from `0` so a later sweep can tell how long the entry
    /// has been pinned (#179 finding 1).
    pub(crate) fn add_in_flight(&mut self, peer: &PeerId, count: u32, now: u64) {
        if let Some(e) = self.entries.get_mut(peer) {
            let was_idle = e.quality.in_flight == 0;
            e.quality.in_flight = e.quality.in_flight.saturating_add(count);
            if was_idle && e.quality.in_flight > 0 {
                e.in_flight_since = Some(now);
            }
        }
    }

    /// Record that a range dispatched to `peer` settled (in-flight decrement, SPEC §5.3). Clears
    /// `in_flight_since` once `in_flight` returns to `0`.
    pub(crate) fn release_in_flight(&mut self, peer: &PeerId, count: u32) {
        if let Some(e) = self.entries.get_mut(peer) {
            e.quality.in_flight = e.quality.in_flight.saturating_sub(count);
            if e.quality.in_flight == 0 {
                e.in_flight_since = None;
            }
        }
    }

    /// Reclaim `in_flight` on every entry whose dispatch has aged past [`DISPATCH_TTL_SECS`] without
    /// settling (#179 finding 1). A `record_outcome` that never arrives — a peer dispatched to and then
    /// gone silent — must not pin the entry as busy forever: this makes the accounting self-healing
    /// (independent of capacity pressure) rather than relying solely on force-eviction at capacity.
    /// Called opportunistically by the engine on each `select`/`rebalance` (SPEC §5.3).
    pub fn reclaim_stale_in_flight(&mut self, now: u64) {
        for e in self.entries.values_mut() {
            if let Some(since) = e.in_flight_since {
                if now.saturating_sub(since) >= DISPATCH_TTL_SECS {
                    e.quality.in_flight = 0;
                    e.in_flight_since = None;
                }
            }
        }
    }

    /// Seed the connection-class **prior** on a fresh cold entry (SPEC §3.3). A `Relayed` peer starts
    /// no better than a direct peer (it is not *preferred* before measured); the prior is subordinate
    /// to measured outcomes. We seed only reliability-neutral priors here — the magnitude of any
    /// relayed throughput handicap is learned by the scorer (SPEC §4.2), not baked as a constant.
    fn seed_class_prior(&self, entry: &mut PeerEntry) {
        // Cold peers get a neutral reliability prior so exploration treats them as uncertain, not
        // failed. No throughput prior is seeded (leaving throughput unmeasured => exploratory).
        entry.quality.reliability.seed_prior(0.5);
    }

    /// Enforce the capacity bound: while over capacity, evict the lowest eviction-value evictable
    /// entry (never a connected/in-flight peer — SPEC §2.5). Eviction discards learned quality; a
    /// re-learned peer starts cold again.
    ///
    /// First reclaims any TTL-expired `in_flight` (#179 finding 1) so genuinely-abandoned dispatches
    /// stop protecting their entry before eviction is even considered. If, after that, nothing
    /// satisfies the normal [`PeerEntry::is_evictable`] (every remaining entry is connected or still
    /// has a *live* in-flight dispatch), falls back to [`PeerEntry::is_force_evictable`] — shedding the
    /// lowest-value non-connected entry regardless of in-flight — so the capacity bound is a HARD limit
    /// that always holds (SPEC §2.5 "the capacity bound is always enforceable"), never merely a target
    /// an attacker can pin the registry above by feeding unique cold peer_ids that go silent.
    fn enforce_capacity(&mut self, now: u64) {
        self.reclaim_stale_in_flight(now);
        while self.entries.len() > self.capacity {
            let victim = self
                .entries
                .values()
                .filter(|e| e.is_evictable())
                .min_by(|a, b| {
                    a.eviction_value(now)
                        .partial_cmp(&b.eviction_value(now))
                        .unwrap_or(std::cmp::Ordering::Equal)
                })
                .map(|e| e.peer_id)
                .or_else(|| {
                    // Nothing is cleanly evictable — fall back to the lowest-value non-connected entry
                    // even with a live in-flight count, so the bound is never bypassed (#179 finding 1).
                    self.entries
                        .values()
                        .filter(|e| e.is_force_evictable())
                        .min_by(|a, b| {
                            a.eviction_value(now)
                                .partial_cmp(&b.eviction_value(now))
                                .unwrap_or(std::cmp::Ordering::Equal)
                        })
                        .map(|e| e.peer_id)
                });
            match victim {
                Some(id) => {
                    self.entries.remove(&id);
                    self.evicted.push(id);
                }
                // Everything left is a currently-connected live link — cannot shed further; the
                // capacity bound is only exceeded by the count of genuinely-connected peers, which is
                // bounded by real network topology, not by attacker-fed candidate volume.
                None => break,
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use dig_dht::AddressKind;

    fn pid(b: u8) -> PeerId {
        PeerId::from_bytes([b; 32])
    }
    fn cand(b: u8) -> Candidate {
        Candidate::new(pid(b), vec![CandidateAddr::direct("10.0.0.1", 9444)])
    }

    #[test]
    fn upsert_candidate_inserts_cold_then_updates_preserving_quality() {
        let mut r = Registry::new(100);
        assert_eq!(
            r.upsert_candidate(&cand(1), Provenance::Dht, 100),
            FeedResult::Inserted
        );
        assert!(r.get(&pid(1)).unwrap().quality.is_cold());
        // Simulate learning.
        r.get_mut(&pid(1))
            .unwrap()
            .quality
            .observe_throughput(500.0);
        r.get_mut(&pid(1)).unwrap().quality.bump_samples();
        // A re-upsert keeps the learned quality.
        assert_eq!(
            r.upsert_candidate(&cand(1), Provenance::Dht, 200),
            FeedResult::Updated
        );
        assert!(!r.get(&pid(1)).unwrap().quality.is_cold());
    }

    #[test]
    fn disconnect_retains_entry_and_quality() {
        let mut r = Registry::new(100);
        r.mark_connected(pid(1), Provenance::Gossip, 100);
        r.get_mut(&pid(1))
            .unwrap()
            .quality
            .observe_throughput(700.0);
        r.get_mut(&pid(1)).unwrap().quality.bump_samples();
        assert_eq!(
            r.mark_disconnected(&pid(1), false),
            FeedResult::MarkedDisconnected
        );
        let e = r.get(&pid(1)).unwrap();
        assert!(!e.connected);
        assert!(!e.quality.is_cold(), "quality retained across disconnect");
        assert!(e.is_eligible(), "a plain disconnect stays eligible");
    }

    #[test]
    fn banned_peer_is_ineligible_until_reconnect() {
        let mut r = Registry::new(100);
        r.mark_connected(pid(1), Provenance::Gossip, 100);
        r.mark_disconnected(&pid(1), true);
        assert!(!r.get(&pid(1)).unwrap().is_eligible());
        // Re-adding (reconnect) clears the ban.
        r.mark_connected(pid(1), Provenance::Gossip, 200);
        assert!(r.get(&pid(1)).unwrap().is_eligible());
    }

    #[test]
    fn stronger_provenance_wins_weaker_ignored() {
        let mut r = Registry::new(100);
        r.upsert_candidate(&cand(1), Provenance::Pex, 100);
        r.upsert_candidate(&cand(1), Provenance::Dht, 100);
        assert_eq!(r.get(&pid(1)).unwrap().provenance, Provenance::Dht);
        // A weaker provenance does not downgrade.
        r.upsert_candidate(&cand(1), Provenance::Pex, 100);
        assert_eq!(r.get(&pid(1)).unwrap().provenance, Provenance::Dht);
    }

    #[test]
    fn eviction_sheds_lowest_value_never_connected_or_in_flight() {
        let mut r = Registry::new(2);
        // Two connected, high-value peers.
        r.mark_connected(pid(1), Provenance::Gossip, 100);
        r.mark_connected(pid(2), Provenance::Gossip, 100);
        // A third cold, disconnected candidate arrives → over capacity → the lowest-value evictable
        // is shed. The two connected peers are protected, so the new cold one is evicted immediately.
        r.upsert_candidate(&cand(3), Provenance::Dht, 100);
        assert!(r.get(&pid(1)).is_some());
        assert!(r.get(&pid(2)).is_some());
        assert!(
            r.get(&pid(3)).is_none(),
            "connected peers protected; cold candidate evicted"
        );
    }

    #[test]
    fn eviction_prefers_shedding_stale_unmeasured_over_fresh_measured() {
        let mut r = Registry::new(1);
        // A measured, recently-active peer.
        r.upsert_candidate(&cand(1), Provenance::Dht, 100);
        let e = r.get_mut(&pid(1)).unwrap();
        e.quality.observe_throughput(900.0);
        e.quality.bump_samples();
        e.last_outcome_at = Some(1000);
        // A stale cold peer arrives much later → over capacity → the stale cold one should go, not the
        // valuable measured one.
        r.upsert_candidate(&cand(2), Provenance::Dht, 5000);
        assert!(
            r.get(&pid(1)).is_some(),
            "the valuable measured peer survives"
        );
        assert!(r.get(&pid(2)).is_none(), "the stale cold peer is evicted");
    }

    #[test]
    fn in_flight_peer_is_not_removed() {
        let mut r = Registry::new(100);
        r.upsert_candidate(&cand(1), Provenance::Dht, 100);
        r.add_in_flight(&pid(1), 2, 100);
        assert_eq!(r.remove(&pid(1)), FeedResult::Updated); // kept — busy
        assert!(r.get(&pid(1)).is_some());
        r.release_in_flight(&pid(1), 2);
        assert_eq!(r.remove(&pid(1)), FeedResult::MarkedDisconnected);
        assert!(r.get(&pid(1)).is_none());
    }

    #[test]
    fn set_connection_class_upserts_and_attaches() {
        let mut r = Registry::new(100);
        assert_eq!(
            r.set_connection_class(pid(1), TraversalKind::Relayed, 100),
            FeedResult::Inserted
        );
        assert_eq!(
            r.get(&pid(1)).unwrap().connection_class,
            Some(TraversalKind::Relayed)
        );
    }

    /// #179 HIGH finding 1: a dispatched-but-never-settled peer must not permanently pin a capacity
    /// slot. Feed more unique cold peers than capacity, bump `in_flight` on each (as `select` does)
    /// and never release it (the peer went silent) — the registry MUST stay bounded at `capacity`
    /// because `enforce_capacity` MUST fall back to evicting a disconnected pinned entry when nothing
    /// is cleanly evictable (SPEC §2.5 "the capacity bound is always enforceable").
    #[test]
    fn capacity_bound_holds_even_when_every_entry_is_pinned_by_stale_in_flight() {
        let capacity = 4;
        let mut r = Registry::new(capacity);
        // Feed 3x capacity unique disconnected peers, each dispatched (in_flight bumped) and never
        // settled — simulating an attacker feeding unique cold peer_ids that go silent post-dispatch.
        for b in 0..(capacity as u8 * 3) {
            r.upsert_candidate(&cand(b), Provenance::Dht, 100);
            r.add_in_flight(&pid(b), 3, 100); // mirrors select_over's headroom-sized bump
        }
        assert!(
            r.len() <= capacity,
            "registry must stay bounded at {capacity} even when every entry has stuck in_flight; got {}",
            r.len()
        );
    }

    /// #179 HIGH finding 1: a dispatch that never settles must be reclaimable by TTL/epoch so the
    /// entry becomes cleanly evictable again (not merely force-evicted at capacity, but genuinely
    /// un-pinned) — SPEC §2.5's "capacity bound is always enforceable" via a real accounting fix, not
    /// only a last-resort eviction override.
    #[test]
    fn stale_in_flight_is_reclaimed_after_ttl_making_the_entry_evictable_again() {
        let mut r = Registry::new(100);
        r.upsert_candidate(&cand(1), Provenance::Dht, 100);
        r.add_in_flight(&pid(1), 5, 100);
        assert!(
            !r.get(&pid(1)).unwrap().is_evictable(),
            "pinned while in flight"
        );
        // Long after the dispatch, with no matching release, a TTL sweep reclaims the stale in_flight.
        r.reclaim_stale_in_flight(100 + DISPATCH_TTL_SECS + 1);
        assert_eq!(
            r.get(&pid(1)).unwrap().quality.in_flight,
            0,
            "a dispatch that never settled must be reclaimed after its TTL"
        );
        assert!(
            r.get(&pid(1)).unwrap().is_evictable(),
            "reclaiming stale in_flight must make the entry evictable again"
        );
    }

    /// #179 HIGH finding 2: the registry must surface which `peer_id`s left so the engine can prune
    /// its own side maps (`last_selected`, `dispatched`). Eviction via `enforce_capacity` (triggered by
    /// an over-capacity `upsert_candidate`) must be drainable.
    #[test]
    fn drain_evicted_reports_peers_shed_by_capacity_enforcement() {
        let mut r = Registry::new(1);
        r.upsert_candidate(&cand(1), Provenance::Dht, 100);
        assert!(r.drain_evicted().is_empty(), "no eviction yet");
        // A second candidate over capacity evicts the first (lowest value, both cold/disconnected).
        r.upsert_candidate(&cand(2), Provenance::Dht, 200);
        let evicted = r.drain_evicted();
        assert_eq!(evicted, vec![pid(1)], "capacity eviction must be reported");
        // Draining again yields nothing until another eviction happens.
        assert!(r.drain_evicted().is_empty());
    }

    /// #179 HIGH finding 2: an explicit `remove` that actually deletes the entry must also be reported
    /// via `drain_evicted` so the engine prunes its side maps uniformly for every removal path.
    #[test]
    fn drain_evicted_reports_peers_shed_by_explicit_remove() {
        let mut r = Registry::new(100);
        r.upsert_candidate(&cand(1), Provenance::Dht, 100);
        r.remove(&pid(1));
        assert_eq!(r.drain_evicted(), vec![pid(1)]);
    }

    #[test]
    fn preferred_address_is_most_direct() {
        let mut r = Registry::new(100);
        let c = Candidate::new(
            pid(1),
            vec![
                CandidateAddr {
                    host: "r".into(),
                    port: 1,
                    kind: AddressKind::Reflexive,
                },
                CandidateAddr::direct("d", 2),
            ],
        );
        r.upsert_candidate(&c, Provenance::Dht, 100);
        assert_eq!(
            r.get(&pid(1)).unwrap().preferred_address().unwrap().kind,
            AddressKind::Direct
        );
    }
}