diff --git a/docs/adr/ADR-0004-commitment-bound-quote-pricing.md b/docs/adr/ADR-0004-commitment-bound-quote-pricing.md index 04f7a954..43041cb6 100644 --- a/docs/adr/ADR-0004-commitment-bound-quote-pricing.md +++ b/docs/adr/ADR-0004-commitment-bound-quote-pricing.md @@ -501,3 +501,67 @@ of a peer missing fraction `x` of committed bytes is ~`1-(1-x)^k` dispersed clustered, compounding across the fleet's per-peer audit rate. Proposed target, pending an adversarial staging run with hash-retaining deleters: at least 99% detection within 24 hours for at least 1% missing bytes. + +## Amendment 4 (2026-07-23): pending admission is sized to the launch budget + +Amendment 3 stopped dead entries from squatting the queue, but admission +itself was still unbounded relative to the budget: `pending` shared the +4096-entry commitment-cache cap while the token bucket can launch at most +~31 audits inside one effective answerability window (150 min of +eligibility at one token per 5 minutes, plus a burst of 2). At the +2026-07-17 staging arrival rate (~113 admitted nominations per node-hour +against ~12 launches) over 99% of admitted work could never launch: the +queue was an aging backlog whose entries mostly expired, and the +`pending` / `oldest_pending_quote_age_ms` telemetry measured the backlog's +age rather than schedulable work. + +`pending` is now capped at `FIRST_AUDIT_PENDING_CAP`, the number of usable +launches inside one window under the STRICT reserve-time predicate: burst +tokens at age zero plus every refill that still leaves the quote answerable +through the maximum launch jitter and send slack (currently 31). The unit +test simulates the shipped predicate instant-by-instant and must agree with +the derivation, so retuning the budget, window, or jitter re-sizes +admission automatically. The summary line reports `pending_cap` and +`reserved` (the one reservation held outside the queue), so schedulable +occupancy is `pending + reserved`, a first-class dashboard quantity. + +Overflow follows two rules, in order. First, an arrival that would displace +a different peer grants the queue a reservation opportunity before anything +is evicted: if no reservation is outstanding and a token and in-flight slot +are free, one pending entry moves into the reservation and the arrival is +admitted with no eviction. The scheduler holds at most ONE outstanding +jitter reservation — a deliberate one-entry guard slot outside the queue, +so total schedulable occupancy is at most `cap + 1`, reported as +`pending` plus `reserved` — which means an ingress burst consumes at most +one launch opportunity per jitter window, even while a second burst token +remains. Second, if an otherwise-admissible distinct-peer arrival still +finds the queue at capacity after that opportunity, it displaces a +uniformly RANDOM incumbent, accounted as `capacity_evicted` — the steady +overload signal. +Random selection is the security property: with deterministic retention +(oldest-out or newest-kept), an ordered batch of `cap` distinct-peer paid +nominations flushes a chosen target with certainty; with a random victim, +each overflow evicts a specific target with probability `1/cap`, so +suppressing it with confidence `1−e` costs about `cap · ln(1/e)` +distinct-peer nominations (~95 paid pins for 95% at cap 31, ~140 for 99%) +and is never certain, while the target may launch in the meantime. A fresh +pin still always enters the sample with an unpredictable chance of prompt +audit; refusing at the door would instead let a sustained flood guarantee +every later pin is never admitted. Displacement stamps no suppression +state, so a displaced peer's next nomination is judged like any newcomer. +Under light load (arrivals below the drain rate) the cap is not reached +and coverage remains exhaustive. + +For comparison, the pre-cap queue's 4,096-entry bound made targeted +eviction expensive but protected work that mostly could not launch anyway: +at observed staging rates, over 99% of those entries expired unaudited, so +the practical suppression of any given pin was already the budget share, +not the eviction threshold. The honest quantity to compare across designs +is P(audited before answerability expiry), which the cap leaves at the +budget share while making the skipped remainder visible at admission time. + +Coverage semantics are otherwise unchanged from Amendments 2-3: a budgeted +probabilistic exam backstopped by the gossip-lottery path. What changes is +the honesty of the funnel — admitted now approximates launchable, and +overflow is explicit at admission time (`capacity_evicted`) instead of +surfacing hours later as mass expiry (`outside_answerability_window`). diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 98a44dc3..7b3d21fa 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -47,7 +47,8 @@ use std::pin::Pin; use crate::logging::{debug, error, info, warn}; use futures::stream::FuturesUnordered; use futures::{Future, StreamExt}; -use rand::Rng; +use rand::rngs::StdRng; +use rand::{Rng, SeedableRng}; use tokio::sync::{mpsc, Notify, RwLock, Semaphore}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; @@ -204,7 +205,8 @@ enum FirstAuditQueueOutcome { /// lost the freshness tie (it was not the newer of the two) and the /// existing entry was retained. Benign, not an attack signal. RetainedOnTie, - /// A DIFFERENT peer's entry was evicted by the bounded LRU to make room. + /// A DIFFERENT peer's entry was displaced from the capped pending queue + /// (random victim) to make room. CapacityEvicted { peer: PeerId, pin: [u8; 32] }, } @@ -227,6 +229,7 @@ fn coalesce_first_audit_event( pending: &mut LruCache, incoming: MonetizedPinEvent, incoming_is_newer: bool, + rng: &mut StdRng, ) -> FirstAuditQueueOutcome { if let Some(existing) = pending.peek(&incoming.peer) { // Strictly lower -> the incoming loses and is dropped WITHOUT touching @@ -244,9 +247,36 @@ fn coalesce_first_audit_event( let _ = pending.push(incoming.peer, incoming); return FirstAuditQueueOutcome::Coalesced; } - // No same-peer entry: a `push` at capacity evicts the LRU (a DIFFERENT peer). + // No same-peer entry. At capacity, displace a UNIFORMLY RANDOM incumbent, + // never the LRU: deterministic keep-newest would let an ordered batch of + // `cap` distinct-peer nominations flush a chosen target before any launch + // lane sees it, cutting the eviction cost from probabilistic to exact. A + // random victim caps an attacker's per-nomination eviction probability at + // `1/cap` regardless of arrival order or timing, so suppressing a specific + // target with confidence `1-e` costs ~`cap*ln(1/e)` distinct-peer paid + // nominations and is never certain (ADR-0004 Amendment 4). + if pending.len() >= pending.cap().get() { + let victim = { + let idx = rng.gen_range(0..pending.len()); + pending.iter().nth(idx).map(|(p, _)| *p) + }; + if let Some(victim_peer) = victim { + if let Some(evicted) = pending.pop(&victim_peer) { + let _ = pending.push(incoming.peer, incoming); + return FirstAuditQueueOutcome::CapacityEvicted { + peer: victim_peer, + pin: evicted.pin, + }; + } + } + } match pending.push(incoming.peer, incoming) { None => FirstAuditQueueOutcome::Queued, + // Unreachable: the random-displacement branch above guarantees + // `len < cap` here (the caller holds `&mut`, so no concurrent insert + // exists). Kept so any future logic error surfaces as an ACCOUNTED + // eviction rather than silent loss — but note this arm would be + // LRU-order, not random, so it must stay unreachable. Some((evicted_peer, evicted)) => FirstAuditQueueOutcome::CapacityEvicted { peer: evicted_peer, pin: evicted.pin, @@ -469,6 +499,38 @@ fn quote_answerable_through_nominal_jitter(quote_ts: SystemTime, now: SystemTime quote_within_audit_window(quote_ts, now) && quote_within_audit_window(quote_ts, latest_send) } +/// Open the single first-audit reservation from `pending` if none is +/// outstanding: samples the launch jitter, snapshots the shared cooldown +/// read-only, and delegates to [`FirstAuditScheduler::try_reserve`]. One +/// shared implementation for both drainer call sites — the per-wake launch +/// phase and the pre-overflow opportunity inside the ingress batch — so the +/// two cannot drift. +async fn open_first_audit_reservation( + scheduler: &mut FirstAuditScheduler, + cooldown: &RwLock>, + observability: &Arc, +) { + if scheduler.has_reservation() { + return; + } + let jitter = Duration::from_millis(rand::thread_rng().gen_range( + 0..=u64::try_from(config::FIRST_AUDIT_LAUNCH_JITTER_MAX.as_millis()).unwrap_or(u64::MAX), + )); + let inflight = observability.inflight.load(Ordering::Relaxed); + let reserved = { + let cooldown = cooldown.read().await; + scheduler.try_reserve(Instant::now(), inflight, jitter, &cooldown, observability) + }; + if reserved { + if let Some(peer) = scheduler.reserved_peer() { + debug!( + "First-audit scheduler: audit_trigger=first_monetized outcome=reserved peer={peer} pending={}", + scheduler.pending_len() + ); + } + } +} + /// A far-future `Instant` used to effectively DISABLE the promotion-timer /// select arm when no reservation is outstanding. The drainer still wakes at /// least every [`config::FIRST_AUDIT_RETRY_INTERVAL`] via its tick, so this only @@ -501,7 +563,14 @@ struct FirstAuditScheduler { /// (a real send), never at reservation, so a cancelled reservation can be /// re-nominated. first_audited: LruCache<[u8; 32], ()>, - /// Newest-per-peer pending nominations not yet launched. + /// Highest-count-per-peer pending nominations not yet launched. Capped at + /// [`FIRST_AUDIT_PENDING_CAP`] (what the token budget can launch within + /// one effective answerability window); at capacity a uniformly RANDOM + /// incumbent is displaced and counted as `capacity_evicted`, so eviction + /// of a specific entry can never be forced deterministically. One further + /// entry may be held in `reserved` outside this queue — a deliberate + /// one-entry guard slot, so total schedulable occupancy is at most + /// `cap + 1`; the summary line reports both (`pending`/`reserved`). pending: LruCache, /// Token bucket + per-peer re-audit window. limiter: FirstAuditLimiter, @@ -521,18 +590,24 @@ struct FirstAuditScheduler { /// Such an event is dropped at ingress: never queued, and hence never /// launched nor marked first-audited. self_peer: PeerId, + /// RNG for random-victim displacement (ADR-0004 Amendment 4). Owned by the + /// scheduler so tests can seed it and reproduce eviction sequences exactly; + /// production seeds from OS entropy at construction. + rng: StdRng, } impl FirstAuditScheduler { fn new(now: Instant, self_peer: PeerId) -> Self { - let cap = NonZeroUsize::new(MAX_LAST_COMMITMENT_BY_PEER).unwrap_or(NonZeroUsize::MIN); + let dedup_cap = NonZeroUsize::new(MAX_LAST_COMMITMENT_BY_PEER).unwrap_or(NonZeroUsize::MIN); + let pending_cap = NonZeroUsize::new(FIRST_AUDIT_PENDING_CAP).unwrap_or(NonZeroUsize::MIN); Self { - first_audited: LruCache::new(cap), - pending: LruCache::new(cap), + first_audited: LruCache::new(dedup_cap), + pending: LruCache::new(pending_cap), limiter: FirstAuditLimiter::new(now), reserved: None, oldest_first_lane: false, self_peer, + rng: StdRng::from_entropy(), } } @@ -570,7 +645,7 @@ impl FirstAuditScheduler { /// Drop every pending nomination that has aged past the answerability /// horizon. `try_reserve` collects expired entries only while it can scan — /// under token starvation it returns at the budget gate first — so without - /// a periodic sweep dead entries squat the bounded LRU (evicting live peers + /// a periodic sweep dead entries squat the capped pending queue (displacing /// at capacity) and inflate the `pending`/`oldest_pending_quote_age_ms` /// telemetry. Each removal is accounted as `outside_answerability_window`, /// exactly like a scan-time expiry. Returns how many entries were dropped; @@ -598,6 +673,16 @@ impl FirstAuditScheduler { self.reserved.is_some() } + /// Whether admitting `event` would displace a DIFFERENT peer's pending + /// entry: the queue is at capacity and `event.peer` has no incumbent to + /// coalesce into. The drainer checks this before enqueueing so pending + /// work gets a reservation opportunity BEFORE destructive overflow + /// (ADR-0004 Amendment 4) — a successful reservation moves one entry out + /// of `pending`, freeing the slot without any eviction. + fn would_displace(&self, event: &MonetizedPinEvent) -> bool { + self.pending.len() >= self.pending.cap().get() && self.pending.peek(&event.peer).is_none() + } + /// When the outstanding reservation becomes eligible for promotion. fn reserved_ready_at(&self) -> Option { self.reserved.as_ref().map(|r| r.ready_at) @@ -610,13 +695,20 @@ impl FirstAuditScheduler { /// Admit a monetized nomination into `pending`. Dropped at ingress if it /// targets the local peer (see [`Self::self_peer`]); dropped as a duplicate - /// if already first-audited; the window screen is bypassed for the - /// currently reserved peer (so a successor is retained across the - /// reservation, never window-dropped); otherwise window-screened. - /// Coalescing is highest-count-per-peer (newest on a tie) — a lower-count - /// successor never displaces a higher-count pending pin. An incumbent that - /// has aged past the answerability horizon is dropped (and accounted as an - /// expiry) before coalescing, so a dead pin never vetoes a live nomination. + /// if already first-audited; the window screen is bypassed for the currently + /// reserved peer (so a successor is retained across the reservation, never + /// window-dropped); otherwise window-screened. Coalescing is + /// highest-count-per-peer (newest on a tie) — a lower-count successor never + /// displaces a higher-count pending pin. An incumbent that has aged past + /// the answerability horizon is dropped (and accounted as an expiry) before + /// coalescing, so a dead pin never vetoes a live nomination. Admission for + /// a NEW peer at capacity displaces a uniformly RANDOM incumbent (see + /// [`FIRST_AUDIT_PENDING_CAP`] and [`coalesce_first_audit_event`]): + /// displacement is accounted as `capacity_evicted` and stamps no + /// suppression, so a displaced peer's next nomination is judged like any + /// newcomer. The drainer additionally offers a reservation opportunity + /// via [`Self::would_displace`] before calling this on an overflowing + /// arrival. fn enqueue(&mut self, event: MonetizedPinEvent, obs: &Arc) { if event.peer == self.self_peer { obs.self_target_skipped.fetch_add(1, Ordering::Relaxed); @@ -651,7 +743,7 @@ impl FirstAuditScheduler { } // Ordinary enqueue: the incoming arrived last, so it wins an equal-count // tie. - match coalesce_first_audit_event(&mut self.pending, event, true) { + match coalesce_first_audit_event(&mut self.pending, event, true, &mut self.rng) { FirstAuditQueueOutcome::Queued => { obs.queued.fetch_add(1, Ordering::Relaxed); } @@ -827,7 +919,7 @@ impl FirstAuditScheduler { // counted at ingress, so `queued` is not re-incremented): a capacity // eviction of a DIFFERENT peer and a suppressed reserved event are // both observable per the ADR funnel. - match coalesce_first_audit_event(&mut self.pending, event, false) { + match coalesce_first_audit_event(&mut self.pending, event, false, &mut self.rng) { FirstAuditQueueOutcome::CapacityEvicted { .. } => { obs.capacity_evicted.fetch_add(1, Ordering::Relaxed); } @@ -997,6 +1089,55 @@ const COMMITMENT_SIG_VERIFY_MIN_INTERVAL: Duration = Duration::from_secs(60); /// the cap is the third line of defence against sybil/churn flooding. const MAX_LAST_COMMITMENT_BY_PEER: usize = 4096; +/// ADR-0004 Amendment 4: admission cap for the first-audit pending queue, +/// sized to the launch budget instead of the commitment cache. +/// +/// The token bucket can launch at most one audit per +/// [`config::FIRST_AUDIT_LAUNCH_INTERVAL`] (plus the +/// [`config::FIRST_AUDIT_BUDGET_BURST`] allowance), and a pending pin stays +/// launchable for at most the effective answerability window +/// ([`GOSSIP_ANSWERABILITY_TTL`] − [`MONETIZED_AUDIT_SKEW_MARGIN`]). Any +/// occupancy beyond `window / interval + burst` is work that cannot launch +/// before it expires, so admitting it only builds an aging backlog whose +/// telemetry measures the backlog instead of schedulable work. +/// +/// At capacity, admission displaces a uniformly RANDOM incumbent (counted as +/// `capacity_evicted`), so under overload `pending` degrades to a budget-sized +/// rolling sample in which no arrival order lets an attacker deterministically +/// flush a chosen target; a fresh pin always enters the sample with an +/// unpredictable chance of near-immediate audit, whereas refusing at the door +/// would let a sustained payment flood guarantee that every later pin is never +/// admitted. See [`coalesce_first_audit_event`] for the displacement rule and +/// ADR-0004 Amendment 4 for the eviction-cost math. +/// +/// Derivation uses the STRICT reserve-time horizon: a launch at quote age `a` +/// is usable only if the quote stays answerable through `a` plus the maximum +/// launch jitter and send slack, so the last usable refill is the largest +/// `k` with `k * interval < cutoff - jitter_max - slack`. The unit test +/// simulates the shipped predicate instant-by-instant and must agree. +#[allow(clippy::cast_possible_truncation)] // bounded by TTL-secs/interval-secs + burst (~31) +const FIRST_AUDIT_PENDING_CAP: usize = { + let strict_horizon_secs = GOSSIP_ANSWERABILITY_TTL + .as_secs() + .saturating_sub(MONETIZED_AUDIT_SKEW_MARGIN.as_secs()) + .saturating_sub(config::FIRST_AUDIT_LAUNCH_JITTER_MAX.as_secs()) + .saturating_sub(FIRST_AUDIT_SEND_LATENCY_SLACK.as_secs()); + // Strictly-inside count: a refill landing exactly ON the horizon is dead + // (`age >= cutoff` rejects), hence the `- 1`. Compile-time division-by-zero + // if the launch interval is ever zeroed: a zero interval makes the budget + // (and this cap) meaningless. + let refills = + strict_horizon_secs.saturating_sub(1) / config::FIRST_AUDIT_LAUNCH_INTERVAL.as_secs(); + (refills + config::FIRST_AUDIT_BUDGET_BURST as u64) as usize +}; + +// Compile-time guardrails: the cap must be a usable `LruCache` capacity and +// strictly tighter than the commitment-cache bound it replaced. +const _: () = { + assert!(FIRST_AUDIT_PENDING_CAP >= 1); + assert!(FIRST_AUDIT_PENDING_CAP < MAX_LAST_COMMITMENT_BY_PEER); +}; + /// Cap on the sticky `ever_capable_peers` set. Bounds memory so a /// long-running bootstrap node cannot have the set grow without limit /// from peer-id churn. Sized at 4x `MAX_LAST_COMMITMENT_BY_PEER` so @@ -1661,6 +1802,22 @@ impl ReplicationEngine { event = rx.recv() => match event { Some(e) => { observability.received.fetch_add(1, Ordering::Relaxed); + // Pre-overflow reservation opportunity: before an + // arrival may displace a DIFFERENT peer, let the + // queue launch — a successful reservation frees the + // slot without eviction. One reservation can be + // outstanding at a time, so this consumes at most + // one slot per jitter window; arrivals beyond it + // fall through to random displacement (ADR-0004 + // Amendment 4). + if scheduler.would_displace(&e) { + open_first_audit_reservation( + &mut scheduler, + &gossip_audit.cooldown, + &observability, + ) + .await; + } scheduler.enqueue(e, &observability); // Drain a bounded burst so a flood cannot starve the // launch phase. @@ -1669,6 +1826,14 @@ impl ReplicationEngine { match rx.try_recv() { Ok(e) => { observability.received.fetch_add(1, Ordering::Relaxed); + if scheduler.would_displace(&e) { + open_first_audit_reservation( + &mut scheduler, + &gossip_audit.cooldown, + &observability, + ) + .await; + } scheduler.enqueue(e, &observability); drained += 1; } @@ -1776,33 +1941,12 @@ impl ReplicationEngine { // authoritative stamp is at promotion). Serializing // reservations preserves per-launch lane alternation and keeps // at most one jitter timer live. - if !scheduler.has_reservation() { - let jitter = Duration::from_millis( - rand::thread_rng().gen_range( - 0..=u64::try_from(config::FIRST_AUDIT_LAUNCH_JITTER_MAX.as_millis()) - .unwrap_or(u64::MAX), - ), - ); - let inflight = observability.inflight.load(Ordering::Relaxed); - let reserved = { - let cooldown = gossip_audit.cooldown.read().await; - scheduler.try_reserve( - Instant::now(), - inflight, - jitter, - &cooldown, - &observability, - ) - }; - if reserved { - if let Some(peer) = scheduler.reserved_peer() { - debug!( - "First-audit scheduler: audit_trigger=first_monetized outcome=reserved peer={peer} pending={}", - scheduler.pending_len() - ); - } - } - } + open_first_audit_reservation( + &mut scheduler, + &gossip_audit.cooldown, + &observability, + ) + .await; if last_summary.elapsed() >= config::FIRST_AUDIT_SUMMARY_INTERVAL { // Token-independent hygiene: collect entries that aged past @@ -1817,7 +1961,7 @@ impl ReplicationEngine { ); } info!( - "First-audit scheduler summary: audit_trigger=first_monetized ingress_dropped={} received={} queued={} coalesced={} suppressed_lower={} duplicates={} capacity_evicted={} self_target_skipped={} cooldown_deferred_attempts={} rate_deferred_attempts={} window_deduped={} launched={} passed={} timeout={} failed={} bootstrap_claims={} idle={} insufficient_keys={} outside_answerability_window={} pending={} oldest_pending_quote_age_ms={} inflight={} tokens={}", + "First-audit scheduler summary: audit_trigger=first_monetized ingress_dropped={} received={} queued={} coalesced={} suppressed_lower={} duplicates={} capacity_evicted={} self_target_skipped={} cooldown_deferred_attempts={} rate_deferred_attempts={} window_deduped={} launched={} passed={} timeout={} failed={} bootstrap_claims={} idle={} insufficient_keys={} outside_answerability_window={} pending={} pending_cap={} reserved={} oldest_pending_quote_age_ms={} inflight={} tokens={}", FIRST_AUDIT_INGRESS_DROPPED.load(Ordering::Relaxed), observability.received.load(Ordering::Relaxed), observability.queued.load(Ordering::Relaxed), @@ -1838,6 +1982,8 @@ impl ReplicationEngine { observability.insufficient_keys.load(Ordering::Relaxed), observability.outside_answerability_window.load(Ordering::Relaxed), scheduler.pending_len(), + FIRST_AUDIT_PENDING_CAP, + u8::from(scheduler.has_reservation()), scheduler.oldest_pending_quote_age_ms(SystemTime::now()), observability.inflight.load(Ordering::Relaxed), scheduler.tokens(), @@ -6285,6 +6431,7 @@ mod tests { #[test] fn first_audit_coalescing_keeps_highest_count_and_exposes_eviction() { let mut pending = LruCache::new(NonZeroUsize::new(1).unwrap()); + let mut rng = StdRng::seed_from_u64(7); let peer = test_peer(1); let base = MonetizedPinEvent { peer, @@ -6295,7 +6442,7 @@ mod tests { // First insert into an empty slot: Queued. assert_eq!( - coalesce_first_audit_event(&mut pending, base, true), + coalesce_first_audit_event(&mut pending, base, true, &mut rng), FirstAuditQueueOutcome::Queued ); @@ -6306,7 +6453,7 @@ mod tests { ..base }; assert_eq!( - coalesce_first_audit_event(&mut pending, lower, true), + coalesce_first_audit_event(&mut pending, lower, true, &mut rng), FirstAuditQueueOutcome::SuppressedLower ); assert_eq!( @@ -6322,7 +6469,7 @@ mod tests { ..base }; assert_eq!( - coalesce_first_audit_event(&mut pending, higher, true), + coalesce_first_audit_event(&mut pending, higher, true, &mut rng), FirstAuditQueueOutcome::Coalesced ); assert_eq!( @@ -6337,7 +6484,7 @@ mod tests { ..base }; assert_eq!( - coalesce_first_audit_event(&mut pending, equal_newer, true), + coalesce_first_audit_event(&mut pending, equal_newer, true, &mut rng), FirstAuditQueueOutcome::Coalesced ); assert_eq!(pending.peek(&peer).map(|e| e.pin), Some([4; 32])); @@ -6348,7 +6495,7 @@ mod tests { ..base }; assert_eq!( - coalesce_first_audit_event(&mut pending, equal_older, false), + coalesce_first_audit_event(&mut pending, equal_older, false, &mut rng), FirstAuditQueueOutcome::RetainedOnTie ); assert_eq!(pending.peek(&peer).map(|e| e.pin), Some([4; 32])); @@ -6361,7 +6508,7 @@ mod tests { ..base }; assert_eq!( - coalesce_first_audit_event(&mut pending, other_peer, true), + coalesce_first_audit_event(&mut pending, other_peer, true, &mut rng), FirstAuditQueueOutcome::CapacityEvicted { peer, pin: [4; 32] } ); assert_eq!(pending.len(), 1); @@ -6642,9 +6789,228 @@ mod tests { assert_eq!(obs.outside_answerability_window.load(Ordering::Relaxed), 0); } + /// A fresh nomination for peer `b` (distinct pin per peer). + fn live_nomination(b: u8) -> MonetizedPinEvent { + MonetizedPinEvent { + peer: test_peer(b), + pin: [b; 32], + key_count: 100, + quote_ts: SystemTime::now(), + } + } + + /// The pending admission cap must equal the number of usable launches the + /// token budget can perform inside one answerability window, counted by + /// SIMULATING the shipped reserve-time predicate instant-by-instant (burst + /// tokens at age zero, one refill per launch interval), not by repeating + /// the derivation formula. Any larger and admitted work is guaranteed to + /// expire unlaunched; any smaller and the budget idles while nominations + /// are displaced. + #[test] + fn first_audit_pending_cap_matches_strict_launch_horizon() { + let quote_ts = SystemTime::now(); + // Burst launches fire at age zero and must pass the reserve predicate. + assert!(quote_answerable_through_nominal_jitter(quote_ts, quote_ts)); + let mut usable = usize::try_from(config::FIRST_AUDIT_BUDGET_BURST).expect("burst fits"); + let mut refill = 1u32; + loop { + let age = config::FIRST_AUDIT_LAUNCH_INTERVAL * refill; + let at = quote_ts + age; + if !quote_answerable_through_nominal_jitter(quote_ts, at) { + break; + } + usable += 1; + refill += 1; + assert!(refill < 10_000, "runaway horizon simulation"); + } + assert_eq!( + FIRST_AUDIT_PENDING_CAP, usable, + "cap must match the simulated strict launch horizon" + ); + } + + /// The pending queue is sized to the launch budget while the dedup set + /// keeps the commitment-cache bound: shrinking `first_audited` would + /// forget audited pins and re-admit duplicates. + #[test] + fn first_audit_pending_cap_independent_of_dedup_cap() { + let scheduler = FirstAuditScheduler::new(Instant::now(), test_peer(99)); + assert_eq!(scheduler.pending.cap().get(), FIRST_AUDIT_PENDING_CAP); + assert_eq!( + scheduler.first_audited.cap().get(), + MAX_LAST_COMMITMENT_BY_PEER + ); + } + + /// Admission beyond the cap displaces exactly one incumbent per overflow + /// and every displacement is accounted as `capacity_evicted` — the steady + /// overload signal — never silent loss, and never an admission refusal. + #[test] + fn first_audit_admission_beyond_cap_displaces_and_counts() { + let obs = Arc::new(FirstAuditObservability::default()); + let mut scheduler = FirstAuditScheduler::new(Instant::now(), test_peer(99)); + let overflow: usize = 5; + let total = FIRST_AUDIT_PENDING_CAP + overflow; + + for i in 0..total { + let b = u8::try_from(i + 1).expect("test peer count fits u8"); + scheduler.enqueue(live_nomination(b), &obs); + } + + assert_eq!(scheduler.pending_len(), FIRST_AUDIT_PENDING_CAP); + assert_eq!( + obs.capacity_evicted.load(Ordering::Relaxed), + u64::try_from(overflow).expect("overflow fits u64"), + "each admission past the cap displaces exactly one entry" + ); + assert_eq!( + obs.queued.load(Ordering::Relaxed), + u64::try_from(total).expect("total fits u64"), + "displacement is not an admission refusal" + ); + // Every arrival was admitted (the newest always enters the sample). + let last = u8::try_from(total).expect("test peer count fits u8"); + assert!( + scheduler.pending.peek(&test_peer(last)).is_some(), + "the newest arrival always enters the sample" + ); + } + + /// The reviewer-demonstrated suppression route: a target followed by an + /// ordered batch of distinct-peer nominations, all inside one ingress + /// drain batch (`pending_cap < batch <= FIRST_AUDIT_DRAIN_BATCH`), with + /// the token bucket EMPTY so no reservation can intervene. Under the old + /// keep-newest LRU the target was evicted with certainty; under random + /// displacement its per-overflow eviction probability is `1/cap`, so + /// across repeated trials the target must survive some runs and be + /// displaced in others — never deterministically flushed, and every + /// displacement accounted. + #[test] + fn first_audit_ordered_flood_cannot_deterministically_evict_target() { + let flood: usize = 60; // pending_cap < 60 <= FIRST_AUDIT_DRAIN_BATCH + assert!(FIRST_AUDIT_PENDING_CAP < flood); + assert!(flood <= config::FIRST_AUDIT_DRAIN_BATCH); + + let trials = 100u64; + let mut survived = 0u32; + let mut evicted = 0u32; + for trial in 0..trials { + let obs = Arc::new(FirstAuditObservability::default()); + let mut scheduler = FirstAuditScheduler::new(Instant::now(), test_peer(99)); + // Deterministic per-trial seed: the run is fully reproducible (no + // flake budget at all) while still exercising 100 distinct + // eviction sequences. + scheduler.rng = StdRng::seed_from_u64(trial); + // Drain the burst tokens so the pre-overflow reservation + // opportunity cannot fire: this isolates pure retention. + scheduler.limiter.reserve_token(); + scheduler.limiter.reserve_token(); + + let target = test_peer(200); + scheduler.enqueue(live_nomination(200), &obs); + for i in 0..flood { + let b = u8::try_from(i + 1).expect("flood peer fits u8"); + scheduler.enqueue(live_nomination(b), &obs); + } + + assert_eq!(scheduler.pending_len(), FIRST_AUDIT_PENDING_CAP); + let overflows = u64::try_from(1 + flood - FIRST_AUDIT_PENDING_CAP) + .expect("overflow count fits u64"); + assert_eq!( + obs.capacity_evicted.load(Ordering::Relaxed), + overflows, + "every overflow past the cap is accounted, none silent" + ); + if scheduler.pending.peek(&target).is_some() { + survived += 1; + } else { + evicted += 1; + } + } + // With eviction probability 1/cap per overflow, P(target survives one + // trial) ~= (1 - 1/31)^30 ~= 0.37, so both outcomes appear across the + // 100 seeded trials — and the seeds make the split exactly + // reproducible rather than a (vanishingly small) flake budget. + assert!( + survived > 0, + "target must survive some ordered floods — deterministic eviction \ + would mean keep-newest retention regressed" + ); + assert!( + evicted > 0, + "target must also be displaceable — otherwise overflow is refusing \ + admissions instead of sampling" + ); + } + + /// With a token AVAILABLE, an overflowing arrival must first give pending + /// work a launch opportunity: the drainer's pre-overflow reservation pulls + /// one entry out of the queue, so admission proceeds without any eviction. + #[test] + fn first_audit_overflow_reserves_before_destructive_eviction() { + let obs = Arc::new(FirstAuditObservability::default()); + let mut scheduler = FirstAuditScheduler::new(Instant::now(), test_peer(99)); + let cooldown: HashMap = HashMap::new(); + + for i in 0..FIRST_AUDIT_PENDING_CAP { + let b = u8::try_from(i + 1).expect("test peer count fits u8"); + scheduler.enqueue(live_nomination(b), &obs); + } + let overflowing = live_nomination(250); + assert!(scheduler.would_displace(&overflowing)); + + // The drainer's pre-overflow sequence: reserve, then enqueue. + assert!(scheduler.try_reserve(Instant::now(), 0, Duration::ZERO, &cooldown, &obs)); + assert!( + !scheduler.would_displace(&overflowing), + "a successful reservation frees the slot" + ); + scheduler.enqueue(overflowing, &obs); + + assert_eq!(scheduler.pending_len(), FIRST_AUDIT_PENDING_CAP); + assert_eq!( + obs.capacity_evicted.load(Ordering::Relaxed), + 0, + "no destructive eviction when a launch opportunity existed" + ); + } + + /// Displacement stamps no suppression state: a displaced peer's next + /// nomination re-enters like any newcomer (it is NOT window-deduped or + /// treated as a duplicate), so under overload every peer keeps an + /// unpredictable chance of prompt first audit. + #[test] + fn first_audit_displaced_peer_may_be_renominated() { + let obs = Arc::new(FirstAuditObservability::default()); + let mut scheduler = FirstAuditScheduler::new(Instant::now(), test_peer(99)); + let total = FIRST_AUDIT_PENDING_CAP + 1; + + for i in 0..total { + let b = u8::try_from(i + 1).expect("test peer count fits u8"); + scheduler.enqueue(live_nomination(b), &obs); + } + assert_eq!(obs.capacity_evicted.load(Ordering::Relaxed), 1); + // Find whichever peer the random displacement removed. + let displaced = (1..=total) + .map(|i| u8::try_from(i).expect("test peer count fits u8")) + .find(|b| scheduler.pending.peek(&test_peer(*b)).is_none()) + .expect("exactly one admitted peer was displaced"); + + scheduler.enqueue(live_nomination(displaced), &obs); + + assert!( + scheduler.pending.peek(&test_peer(displaced)).is_some(), + "the displaced peer re-enters the sample" + ); + assert_eq!(scheduler.pending_len(), FIRST_AUDIT_PENDING_CAP); + assert_eq!(obs.capacity_evicted.load(Ordering::Relaxed), 2); + assert_eq!(obs.duplicates.load(Ordering::Relaxed), 0); + assert_eq!(obs.window_deduped.load(Ordering::Relaxed), 0); + } + /// The periodic sweep collects expired pending entries even when the token /// bucket is empty — the reserve path returns at the budget gate and never - /// scans — keeping the bounded LRU and the pending/oldest-age telemetry + /// scans — keeping the capped pending queue and the pending/oldest-age telemetry /// honest under fleet-wide starvation. #[test] fn first_audit_sweep_expired_collects_dead_entries_without_tokens() { @@ -6921,6 +7287,7 @@ mod tests { quote_ts: now - Duration::from_secs(secs), }, true, + &mut scheduler.rng, ); } assert_eq!( @@ -6940,6 +7307,7 @@ mod tests { quote_ts: now + Duration::from_secs(5), }, true, + &mut skewed.rng, ); assert_eq!( skewed.oldest_pending_quote_age_ms(now), @@ -6949,13 +7317,19 @@ mod tests { } /// A flood of strictly-lower-count same-peer nominations must neither - /// displace the retained higher pin NOR disturb its LRU position (each is - /// suppressed via `peek`, no `push`), and each must be counted as + /// displace the retained higher pin NOR disturb its recency position (each + /// is suppressed via `peek`, no `push`), and each must be counted as /// `suppressed_lower` — the attempted cheaper-pin self-erasure signal. + /// Recency is asserted DIRECTLY via the queue's MRU-to-LRU iteration order + /// (the documented `lru` contract that also drives the reserve scan's lane + /// ordering), deliberately independent of the overflow policy: eviction is + /// random-victim, so no assertion may require a particular peer to be + /// displaced. #[test] fn first_audit_suppressed_lower_flood_leaves_recency_and_counts() { let mut pending: LruCache = LruCache::new(NonZeroUsize::new(2).unwrap()); + let mut rng = StdRng::seed_from_u64(11); let victim = test_peer(1); let other = test_peer(2); // Victim (high count) inserted first (older), then `other` (newer/MRU). @@ -6968,6 +7342,7 @@ mod tests { quote_ts: SystemTime::now(), }, true, + &mut rng, ); let _ = coalesce_first_audit_event( &mut pending, @@ -6978,6 +7353,13 @@ mod tests { quote_ts: SystemTime::now(), }, true, + &mut rng, + ); + let order_before: Vec = pending.iter().map(|(p, _)| *p).collect(); + assert_eq!( + order_before, + vec![other, victim], + "sanity: `other` is MRU, the victim is LRU" ); // Flood the victim with cheaper nominations. @@ -6992,6 +7374,7 @@ mod tests { quote_ts: SystemTime::now(), }, true, + &mut rng, ); assert_eq!(out, FirstAuditQueueOutcome::SuppressedLower); suppressed += 1; @@ -7002,27 +7385,12 @@ mod tests { pending.peek(&victim).map(|e| (e.pin, e.key_count)), Some(([1; 32], 400)) ); - // Recency unchanged: `other` is still MRU (a new distinct peer at cap 2 - // would evict the LRU; the victim must be the LRU, so `third` evicts the - // victim, not `other`). - let third = test_peer(3); - let out = coalesce_first_audit_event( - &mut pending, - MonetizedPinEvent { - peer: third, - pin: [7; 32], - key_count: 100, - quote_ts: SystemTime::now(), - }, - true, - ); + // Recency untouched: the iteration order (which the reserve lanes + // consume) is byte-for-byte what it was before the flood. + let order_after: Vec = pending.iter().map(|(p, _)| *p).collect(); assert_eq!( - out, - FirstAuditQueueOutcome::CapacityEvicted { - peer: victim, - pin: [1; 32] - }, - "the suppressed-lower flood must not have promoted the victim above `other`" + order_after, order_before, + "the suppressed-lower flood must not have changed any recency position" ); }