diff --git a/crates/skippy-cache/src/lib.rs b/crates/skippy-cache/src/lib.rs index 724658aaab..3e04549da4 100644 --- a/crates/skippy-cache/src/lib.rs +++ b/crates/skippy-cache/src/lib.rs @@ -4,6 +4,7 @@ pub mod identity; pub mod l3; pub mod manager; pub mod payload; +pub mod policy; pub mod radix; pub mod resident; pub mod source; diff --git a/crates/skippy-cache/src/policy/accounting.rs b/crates/skippy-cache/src/policy/accounting.rs new file mode 100644 index 0000000000..0c79d0f59c --- /dev/null +++ b/crates/skippy-cache/src/policy/accounting.rs @@ -0,0 +1,121 @@ +//! Shared-segment accounting: fractional credit so physical bytes are never +//! double-counted across entries (#1650 first slice). + +use std::collections::{BTreeMap, BTreeSet}; + +use crate::policy::EntryKey; + +/// Opaque shared-segment identity. +pub type SegmentId = u64; + +/// Ledger of shared physical segments and their referencing entries. +/// +/// A segment referenced by N entries contributes `size / N` bytes to each +/// entry's effective footprint; releasing an entry drops its reference, and a +/// segment with no references disappears entirely. +#[derive(Debug, Default, Clone)] +pub struct SharedSegmentLedger { + segments: BTreeMap, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct SegmentRecord { + pub size: u64, + pub references: BTreeSet, +} + +impl SharedSegmentLedger { + /// Register a segment of `size` bytes referenced by `entry`. Adding the + /// same reference twice is a no-op. + pub fn add(&mut self, segment: SegmentId, size: u64, entry: EntryKey) { + let record = self.segments.entry(segment).or_insert(SegmentRecord { + size, + references: BTreeSet::new(), + }); + record.references.insert(entry); + } + + /// Release `entry`'s reference to `segment`; drop the segment when the + /// last reference goes. + pub fn release(&mut self, segments: &[SegmentId], entry: EntryKey) { + for segment in segments { + if let Some(record) = self.segments.get_mut(segment) { + record.references.remove(&entry); + if record.references.is_empty() { + self.segments.remove(segment); + } + } + } + } + + /// Fractional bytes charged to `entry` for its shared segments. + pub fn fractional_bytes(&self, entry: EntryKey, segments: &[SegmentId]) -> f64 { + segments + .iter() + .filter_map(|s| self.segments.get(s)) + .filter(|r| r.references.contains(&entry)) + .map(|r| r.size as f64 / r.references.len() as f64) + .sum() + } + + /// Read access for marginal-release computation. + pub fn segment_record(&self, segment: SegmentId) -> Option<&SegmentRecord> { + self.segments.get(&segment) + } + + /// Physical bytes that would actually be released if `entry` were + /// removed: its exclusive bytes are caller-side, so this covers only + /// segments where this entry holds the last reference — the marginal + /// physical release, not the fractional credit. + pub fn marginal_physical_bytes(&self, entry: EntryKey, segments: &[SegmentId]) -> u64 { + segments + .iter() + .filter_map(|s| self.segments.get(s)) + .filter(|r| r.references.len() == 1 && r.references.contains(&entry)) + .map(|r| r.size) + .sum() + } + + /// Total physical bytes held in the ledger. + pub fn total_bytes(&self) -> u64 { + self.segments.values().map(|r| r.size).sum() + } + + pub fn segment_count(&self) -> usize { + self.segments.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn fractional_credit_splits_bytes_without_double_counting() { + let mut ledger = SharedSegmentLedger::default(); + ledger.add(1, 300, 10); + ledger.add(1, 300, 11); + ledger.add(1, 300, 12); + assert_eq!(ledger.total_bytes(), 300); + for entry in [10u64, 11, 12] { + assert_eq!(ledger.fractional_bytes(entry, &[1]), 100.0); + } + } + + #[test] + fn releasing_last_reference_drops_segment() { + let mut ledger = SharedSegmentLedger::default(); + ledger.add(7, 128, 1); + ledger.release(&[7], 1); + assert_eq!(ledger.total_bytes(), 0); + assert_eq!(ledger.segment_count(), 0); + } + + #[test] + fn duplicate_reference_is_idempotent() { + let mut ledger = SharedSegmentLedger::default(); + ledger.add(7, 128, 1); + ledger.add(7, 128, 1); + assert_eq!(ledger.fractional_bytes(1, &[7]), 128.0); + } +} diff --git a/crates/skippy-cache/src/policy/admission.rs b/crates/skippy-cache/src/policy/admission.rs new file mode 100644 index 0000000000..a30f5326a5 --- /dev/null +++ b/crates/skippy-cache/src/policy/admission.rs @@ -0,0 +1,295 @@ +//! Admission, probation, promotion, and eviction selection (#1650 first +//! slice). Pure decision logic over `BenefitPolicy` state. + +use serde::Serialize; + +use crate::policy::{ + BenefitPolicy, CostSample, EntryKey, EvictionVerdict, GhostStats, PolicyEntry, SegmentId, +}; + +/// What the policy decided to do with a candidate or admitted entry. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub enum AdmissionDecisionKind { + AdmitProbation, + AdmitPersist, + Promote, + Reject, +} + +impl AdmissionDecision { + /// A rejected offer whose measured cost was invalid. Constructed without + /// touching policy state. + pub fn rejected_invalid_cost() -> Self { + Self::rejected("invalid-cost-sample") + } + + /// A structurally rejected offer. Constructed without touching policy + /// state. + pub fn rejected(reason: &str) -> Self { + Self { + kind: AdmissionDecisionKind::Reject, + verdict: AdmissionVerdict::Reject, + reasons: vec![reason.into()], + probation_cap_repair: crate::policy::CapRepair::satisfied(), + } + } +} + +/// The verdict plus opaque reasons for logging. +#[derive(Debug, Clone, PartialEq)] +pub struct AdmissionDecision { + pub kind: AdmissionDecisionKind, + pub verdict: AdmissionVerdict, + pub reasons: Vec, + /// Cap-repair plan after this admission (pins never selected; a + /// `Deferred` variant reports the shortfall). The policy does not + /// remove victims itself: committed removal stays with + /// `BenefitPolicy::remove`. + pub probation_cap_repair: crate::policy::CapRepair, +} + +/// Lifecycle state of a policy entry. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PolicyEntryState { + /// Resident in memory only; not persisted to disk. + Probation, + /// Persist-eligible: survived the hit threshold. + Admitted, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AdmissionVerdict { + Admit, + Reject, +} + +pub(crate) fn consider( + policy: &mut BenefitPolicy, + key: EntryKey, + exclusive_bytes: u64, + shared: Vec<(SegmentId, u64)>, + cost: CostSample, +) -> AdmissionDecision { + let reasons: Vec; + let kind; + + let mut seen = std::collections::BTreeSet::new(); + let duplicate_segment = shared.iter().any(|(s, _)| !seen.insert(*s)); + // Prevalidate every known-size conflict across the whole list before any + // mutation (ghost removal or ledger reference) so a rejected offer + // leaves policy state untouched. + let size_conflict = shared.iter().any(|(segment, size)| { + policy + .segments + .segment_record(*segment) + .is_some_and(|record| !record.references.contains(&key) && record.size != *size) + }); + if !cost.is_valid() { + // Invalid measured costs (NaN/infinite/negative) are rejected before + // any mutation: they must never enter entry state. + kind = AdmissionDecisionKind::Reject; + reasons = vec!["invalid-cost-sample".into()]; + } else if duplicate_segment { + kind = AdmissionDecisionKind::Reject; + reasons = vec!["duplicate-segment-reference".into()]; + } else if size_conflict { + kind = AdmissionDecisionKind::Reject; + reasons = vec!["segment-size-conflict".into()]; + } else if cost.net_benefit() <= 0.0 { + kind = AdmissionDecisionKind::Reject; + reasons = vec!["no-net-benefit".into()]; + } else if exclusive_bytes == 0 && shared.is_empty() { + // Zero-footprint entries are free to keep. + kind = AdmissionDecisionKind::AdmitProbation; + reasons = vec!["zero-exclusive-bytes".into()]; + } else { + kind = AdmissionDecisionKind::AdmitProbation; + reasons = vec!["probation-new-entry".into()]; + } + + if matches!(kind, AdmissionDecisionKind::Reject) { + return AdmissionDecision { + kind, + verdict: AdmissionVerdict::Reject, + reasons, + probation_cap_repair: crate::policy::CapRepair::satisfied(), + }; + } + // Carry ghost history in: a recurring entry re-enters with its past + // reuse signal, and an entry whose history already clears the hit + // threshold admits straight to the admitted class (equivalent value + // signal). + let had_ghost = policy.ghosts.contains_key(&key); + let mut ghost = policy.ghosts.remove(&key).unwrap_or(GhostStats { + hits: 0, + reuse_weight: 0.0, + observation_weight: 0.0, + last_observation: 0, + }); + if had_ghost { + // This offer *is* a recurrence: count it as a reuse observation, the + // same way record_hit would, so eviction cannot erase the value + // signal the recurrence just demonstrated. + ghost.hits += 1; + ghost.reuse_weight = ghost.reuse_weight * policy.config.decay.factor + 1.0; + ghost.observation_weight = ghost.observation_weight * policy.config.decay.factor + 1.0; + } + let ghost_promoted = ghost.hits >= policy.config.persistence_hit_threshold as u64; + let state = if ghost_promoted { + PolicyEntryState::Admitted + } else { + PolicyEntryState::Probation + }; + // The decision kind must match the resulting entry state so a + // store-facing caller persists exactly what the policy admitted. + let kind = if ghost_promoted { + AdmissionDecisionKind::AdmitPersist + } else { + kind + }; + let segment_ids: Vec = shared.iter().map(|(s, _)| *s).collect(); + for (segment, size) in &shared { + policy.segments.add(*segment, *size, key); + } + policy.entries.insert( + key, + PolicyEntry { + state, + hits: ghost.hits, + misses: 0, + reuse_weight: ghost.reuse_weight, + observation_weight: ghost.observation_weight, + last_observation: policy.clock, + last_cost: Some(cost), + exclusive_bytes, + segments: segment_ids, + }, + ); + AdmissionDecision { + kind, + verdict: AdmissionVerdict::Admit, + reasons, + probation_cap_repair: crate::policy::CapRepair::satisfied(), + } +} + +pub(crate) fn record_hit( + policy: &mut BenefitPolicy, + key: EntryKey, + cost: CostSample, +) -> Option { + if !cost.is_valid() { + // Invalid measured costs never mutate entry state. + return None; + } + let entry = policy.entries.get_mut(&key)?; + entry.hits += 1; + entry.last_observation = policy.clock; + entry.last_cost = Some(cost); + entry.reuse_weight = entry.reuse_weight * policy.config.decay.factor + 1.0; + entry.observation_weight = entry.observation_weight * policy.config.decay.factor + 1.0; + + if entry.state == PolicyEntryState::Probation + && entry.hits >= policy.config.persistence_hit_threshold as u64 + { + entry.state = PolicyEntryState::Admitted; + return Some(AdmissionDecision { + kind: AdmissionDecisionKind::Promote, + verdict: AdmissionVerdict::Admit, + reasons: vec!["probation-second-hit".into()], + probation_cap_repair: crate::policy::CapRepair::satisfied(), + }); + } + None +} + +pub(crate) fn choose_victims( + policy: &mut BenefitPolicy, + bytes_to_free: u64, + pinned: &[EntryKey], +) -> Vec<(EntryKey, EvictionVerdict)> { + let grace = policy.config.grace_observations; + let clock = policy.clock; + let in_grace = |e: &PolicyEntry| clock.saturating_sub(e.last_observation) < grace; + + // NaN-free scores only (score::compute rejects invalid costs/config), so + // a total ordering is safe: finite values ordered normally, keys + // tie-break. We still avoid partial_cmp().unwrap() defensively. + let mut candidates: Vec<(f64, EntryKey)> = policy + .entries + .iter() + .filter(|(k, _)| !pinned.contains(k)) + .filter_map(|(k, e)| { + let score = super::score::compute(&policy.config, *k, e, &policy.segments)?; + Some((score.value, *k)) + }) + .collect(); + candidates.sort_by(|a, b| a.0.total_cmp(&b.0).then_with(|| a.1.cmp(&b.1))); + + // Marginal physical release for `key` given `selected` victims already + // chosen: exclusive bytes plus segments whose remaining references are + // all inside the selected victim set (or the entry itself). + let marginal_release = + |key: EntryKey, selected: &std::collections::BTreeSet| -> u64 { + let entry = policy.entries.get(&key).expect("candidate present"); + let mut release = entry.exclusive_bytes; + for segment in &entry.segments { + if let Some(record) = policy.segments.segment_record(*segment) { + let others_kept = record + .references + .iter() + .any(|r| *r != key && !selected.contains(r)); + if !others_kept { + release += record.size; + } + } + } + release + }; + + // Eviction order: no-hit probationers first (they have not earned + // bytes), oldest observation first — LRU within the probation class so a + // recently offered candidate is not the automatic first victim. Then + // ascending benefit score. Grace waives under a hard capacity request: + // the byte budget always wins. + let mut order: Vec<(u64, EntryKey)> = policy + .entries + .iter() + .filter(|(k, e)| { + !pinned.contains(k) && e.state == PolicyEntryState::Probation && e.hits == 0 + }) + .map(|(k, e)| (e.last_observation, *k)) + .collect(); + order.sort(); // (oldest observation, key): deterministic + let mut ordered_keys: Vec = order.into_iter().map(|(_, k)| k).collect(); + ordered_keys.extend(candidates.into_iter().map(|(_, key)| key)); + let order = ordered_keys; + + let mut freed: u64 = 0; + let mut victims = Vec::new(); + let mut selected: std::collections::BTreeSet = Default::default(); + // Pass 1: grace honored. Pass 2 (only if the hard budget is still unmet): + // grace waived — the hard byte budget always wins under pressure. + for waive_grace in [false, true] { + for key in &order { + if freed >= bytes_to_free { + break; + } + if selected.contains(key) || !policy.entries.contains_key(key) { + continue; // dedup: probation-first and scored paths overlap + } + let evictable = waive_grace || policy.entries.get(key).is_some_and(|e| !in_grace(e)); + if !evictable { + continue; + } + let release = marginal_release(*key, &selected); + selected.insert(*key); + victims.push((*key, EvictionVerdict::Evict)); + freed += release; + } + if freed >= bytes_to_free { + break; + } + } + victims +} diff --git a/crates/skippy-cache/src/policy/decay.rs b/crates/skippy-cache/src/policy/decay.rs new file mode 100644 index 0000000000..e862af8a0c --- /dev/null +++ b/crates/skippy-cache/src/policy/decay.rs @@ -0,0 +1,25 @@ +//! Exponential decay of reuse statistics (#1650 first slice). + +/// Decay tunables. `factor` is the per-observation retention weight applied to +/// past history: 0.9 keeps 90% of each entry's accumulated ratio weight per +/// new observation. Higher `pressure_decay_sensitivity` (via +/// `BenefitPolicy::observe_pressure`) pulls retention toward 0 faster under +/// churn. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct DecayConfig { + /// Per-observation retention factor in `(0, 1)`. + pub factor: f64, +} + +impl DecayConfig { + /// The retention factor must be finite and strictly inside `(0, 1)`. + pub fn is_valid(&self) -> bool { + self.factor.is_finite() && self.factor > 0.0 && self.factor < 1.0 + } +} + +impl Default for DecayConfig { + fn default() -> Self { + Self { factor: 0.9 } + } +} diff --git a/crates/skippy-cache/src/policy/lru_baseline.rs b/crates/skippy-cache/src/policy/lru_baseline.rs new file mode 100644 index 0000000000..4cbaeab97e --- /dev/null +++ b/crates/skippy-cache/src/policy/lru_baseline.rs @@ -0,0 +1,60 @@ +//! Matched same-capacity LRU baseline for policy comparison (#1650 first +//! slice). Capacity is in exclusive bytes so the comparison is apples-to- +//! apples against `BenefitPolicy` under a hard byte budget. + +use crate::policy::traces::TraceAccess; + +pub struct LruCache { + capacity_bytes: u64, + used_bytes: u64, + /// Most-recent first. + order: Vec, + sizes: std::collections::HashMap, + pub hits: u64, + pub misses: u64, + /// Saved cold-prefill cost (higher is better) over the trace. + pub saved_cost: f64, + /// Bytes written into the cache (lower is better). + pub bytes_written: u64, +} + +impl LruCache { + pub fn new(capacity_bytes: u64) -> Self { + Self { + capacity_bytes, + used_bytes: 0, + order: Vec::new(), + sizes: std::collections::HashMap::new(), + hits: 0, + misses: 0, + saved_cost: 0.0, + bytes_written: 0, + } + } + + pub fn access(&mut self, access: &TraceAccess) -> bool { + if self.order.contains(&access.entry) { + self.hits += 1; + self.saved_cost += (access.cold_prefill_cost - access.restore_cost).max(0.0); + self.order.retain(|e| *e != access.entry); + self.order.insert(0, access.entry); + return true; + } + self.misses += 1; + while self.used_bytes + access.exclusive_bytes > self.capacity_bytes { + let Some(victim) = self.order.pop() else { + break; + }; + if let Some(size) = self.sizes.remove(&victim) { + self.used_bytes = self.used_bytes.saturating_sub(size); + } + } + if access.exclusive_bytes <= self.capacity_bytes { + self.order.insert(0, access.entry); + self.sizes.insert(access.entry, access.exclusive_bytes); + self.used_bytes += access.exclusive_bytes; + self.bytes_written += access.exclusive_bytes; + } + false + } +} diff --git a/crates/skippy-cache/src/policy/mod.rs b/crates/skippy-cache/src/policy/mod.rs new file mode 100644 index 0000000000..0ce74a9bee --- /dev/null +++ b/crates/skippy-cache/src/policy/mod.rs @@ -0,0 +1,615 @@ +//! Benefit-per-exclusive-byte admission, probation, and eviction policy +//! (issue #1650, first slice). +//! +//! This module is deliberately a *pure policy*: it consumes observed events +//! (candidate offers, hits, misses, cost samples) and produces decisions +//! (admit / probation / persist / evict) with opaque reasons. It performs no +//! I/O, holds no locks on the restore path, and never sees prompt content — +//! entries are addressed by an opaque `EntryKey` the caller assigns. +//! +//! The score is the one the issue prescribes: +//! +//! ```text +//! reuse_probability * max(cold_prefill_cost - restore_cost, 0) +//! ----------------------------------------------------------- +//! exclusive_physical_bytes +//! ``` +//! +//! Shared segments are credited fractionally: a physical byte referenced by +//! N entries counts as `bytes / N` against each of them, so total accounted +//! bytes never double-count a segment. +//! +//! Determinism: every ordering falls back to `(score, entry_key)` so two runs +//! over the same trace make identical decisions. + +mod accounting; +mod admission; +mod decay; +#[cfg(test)] +mod lru_baseline; +mod score; +#[cfg(test)] +mod tests; +#[cfg(test)] +mod traces; + +pub use accounting::{SegmentId, SharedSegmentLedger}; +pub use admission::{AdmissionDecision, AdmissionDecisionKind, AdmissionVerdict, PolicyEntryState}; +pub use decay::DecayConfig; +pub use score::{BenefitScore, ScoreInputs}; + +use std::collections::BTreeMap; + +use serde::Serialize; + +/// Opaque, caller-assigned entry identity. Ordered so tie-breaks are +/// deterministic; content-free so policy logs leak nothing about prompts. +pub type EntryKey = u64; + +/// Opaque shared-segment identity. +pub type SegmentRef = SegmentId; + +/// Observed costs, in caller-defined units (the policy only compares them). +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct CostSample { + /// Cold prefill cost for the entry's token range (e.g. ms or tokens). + pub cold_prefill_cost: f64, + /// Measured `queue + restore + suffix-prefill` cost for the same range. + pub restore_cost: f64, +} + +impl CostSample { + /// Net benefit of a restore hit over recomputing cold. Never negative. + pub fn net_benefit(&self) -> f64 { + (self.cold_prefill_cost - self.restore_cost).max(0.0) + } + + /// A usable sample must be finite and nonnegative; measured costs that + /// arrive NaN/infinite (or negative) are rejected so scores and orderings + /// stay total and panic-free. + pub fn is_valid(&self) -> bool { + self.cold_prefill_cost.is_finite() + && self.restore_cost.is_finite() + && self.cold_prefill_cost >= 0.0 + && self.restore_cost >= 0.0 + } +} + +/// Policy decision log line: what was decided, and why, without content. +#[derive(Debug, Clone, PartialEq, Serialize)] +pub struct DecisionReason { + pub entry: EntryKey, + pub decision: AdmissionDecisionKind, + /// Machine-readable reason tokens, e.g. `probation-second-hit`. + pub reasons: Vec, + pub score: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub enum EvictionVerdict { + Keep, + Evict, +} + +/// Tunables. Defaults follow the issue's guidance; every field is `Copy` and +/// plain so config files can carry it verbatim later. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct PolicyConfig { + /// Bytes probation entries may collectively charge before the policy + /// must start dropping the least valuable probationers. + pub probation_byte_budget: u64, + /// Hits required to leave probation and become persist-eligible. + /// The issue names second-hit admission: `2`. + pub persistence_hit_threshold: u32, + /// Minimum reuse probability the estimator may report (floor so a single + /// hit still admits under pressure, and division stays sane). + pub min_reuse_probability: f64, + pub decay: DecayConfig, + /// Observations of grace after admission during which an entry cannot be + /// chosen as an eviction victim: probation must get a fair chance to land + /// its second hit before pressure can reclaim its bytes. + pub grace_observations: u64, + /// Maximum retained ghost records. One-shot keys must not create + /// permanent metadata. + pub ghost_capacity: usize, + /// Ghosts older than this many observations are expired so stale + /// popularity cannot revive indefinitely. + pub ghost_max_age_observations: u64, +} + +impl Default for PolicyConfig { + fn default() -> Self { + Self { + probation_byte_budget: 256 << 20, + persistence_hit_threshold: 2, + min_reuse_probability: 0.01, + decay: DecayConfig::default(), + grace_observations: 32, + ghost_capacity: 4096, + ghost_max_age_observations: 1024, + } + } +} + +impl PolicyConfig { + /// Config bounds: NaN/empty decay or an out-of-range reuse floor would + /// poison every score. `probation_byte_budget` may be 0 (probation off). + pub fn is_valid(&self) -> bool { + self.min_reuse_probability.is_finite() + && (0.0..=1.0).contains(&self.min_reuse_probability) + && self.decay.is_valid() + && self.persistence_hit_threshold >= 1 + } +} + +/// Per-entry policy state and statistics. +#[derive(Debug, Clone, PartialEq)] +pub struct PolicyEntry { + pub state: PolicyEntryState, + pub hits: u64, + pub misses: u64, + /// Decayed reuse-estimator numerator/denominator inputs. + pub reuse_weight: f64, + pub observation_weight: f64, + pub last_cost: Option, + /// Exclusive (non-shared) physical bytes charged to this entry. + pub exclusive_bytes: u64, + /// Segments this entry references; fractional credit lives in the ledger. + pub segments: Vec, + /// Clock value at this entry's last admission or hit; drives grace. + pub last_observation: u64, +} + +impl PolicyEntry { + /// Estimated reuse probability under the configured decay window: a + /// smoothed hit ratio in `[0, 1]`. + pub fn reuse_probability(&self) -> f64 { + if self.observation_weight <= 0.0 { + return 0.0; + } + (self.reuse_weight / self.observation_weight).clamp(0.0, 1.0) + } +} + +/// Cap-repair selection result. `Deferred` means pins/holds (or candidate +/// exhaustion) make the hard probation cap temporarily unsatisfiable: the +/// listed victims should still be committed, and the shortfall reported — +/// a pin is never selected to close it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CapRepair { + /// The class is at or under cap (possibly after committing `victims`). + Satisfied { victims: Vec }, + /// Committing `victims` still leaves the class `shortfall_bytes` over + /// cap; repair is deferred until pins release. + Deferred { + victims: Vec, + shortfall_bytes: u64, + }, +} + +impl CapRepair { + fn satisfied() -> Self { + CapRepair::Satisfied { + victims: Vec::new(), + } + } + + fn satisfied_with(victims: Vec) -> Self { + CapRepair::Satisfied { victims } + } + + fn deferred(victims: Vec, shortfall_bytes: u64) -> Self { + CapRepair::Deferred { + victims, + shortfall_bytes, + } + } + + /// Victims to commit regardless of variant. + pub fn victims(&self) -> &[EntryKey] { + match self { + CapRepair::Satisfied { victims } | CapRepair::Deferred { victims, .. } => victims, + } + } + + /// Over-cap bytes that cannot be repaired while pins are held. + pub fn shortfall_bytes(&self) -> u64 { + match self { + CapRepair::Satisfied { .. } => 0, + CapRepair::Deferred { + shortfall_bytes, .. + } => *shortfall_bytes, + } + } + + /// True when no further repair is possible right now. + pub fn is_deferred(&self) -> bool { + matches!(self, CapRepair::Deferred { .. }) + } +} + +/// Result of a committed removal. +#[derive(Debug, Clone, PartialEq)] +pub struct RemovalOutcome { + pub entry: PolicyEntry, + /// Cap-repair plan after this removal (shares may have risen). Pins are + /// never selected; a `Deferred` result reports the shortfall. + pub probation_cap_repair: CapRepair, +} + +/// The policy engine. Owns per-entry statistics and the shared-segment ledger; +/// the caller drives it from cache events. +pub struct BenefitPolicy { + pub(crate) config: PolicyConfig, + pub(crate) entries: BTreeMap, + pub(crate) segments: SharedSegmentLedger, + /// Monotonic observation counter driving the probation grace window. + pub(crate) clock: u64, + /// Reuse statistics that outlive eviction ("ghosts"): an entry that + /// recurs after eviction carries its history back in, so the second-hit + /// value signal survives cache pressure. Bounded by count and age. + pub(crate) ghosts: BTreeMap, +} + +impl BenefitPolicy { + /// Insert a ghost, evicting the oldest ghost when the count bound is + /// exceeded. `ghost_capacity = 0` disables ghost retention entirely. + fn insert_ghost(&mut self, key: EntryKey, stats: GhostStats) { + if self.config.ghost_capacity == 0 { + return; + } + while self.ghosts.len() >= self.config.ghost_capacity { + let oldest = self + .ghosts + .iter() + .min_by_key(|(k, g)| (g.last_observation, **k)) + .map(|(k, _)| *k); + match oldest { + Some(k) => { + self.ghosts.remove(&k); + } + None => break, + } + } + self.ghosts.insert(key, stats); + } + + /// Drop ghosts older than the configured age bound. Called from the + /// observation-driven entry points so expiry is deterministic over a + /// trace without a background timer. + fn expire_ghosts(&mut self) { + let horizon = self + .clock + .saturating_sub(self.config.ghost_max_age_observations); + self.ghosts.retain(|_, g| g.last_observation >= horizon); + } +} + +/// Surviving statistics for an evicted entry. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct GhostStats { + pub hits: u64, + pub reuse_weight: f64, + pub observation_weight: f64, + /// Clock value when the ghost was created; drives age expiry. + pub last_observation: u64, +} + +impl BenefitPolicy { + /// Panics on invalid config so misconfiguration fails at startup + /// rather than producing NaN scores later. + pub fn new(config: PolicyConfig) -> Self { + assert!(config.is_valid(), "invalid PolicyConfig: {:?}", config); + Self { + config, + entries: BTreeMap::new(), + segments: SharedSegmentLedger::default(), + clock: 0, + ghosts: BTreeMap::new(), + } + } + + pub fn config(&self) -> &PolicyConfig { + &self.config + } + + pub fn ghost(&self, key: EntryKey) -> Option<&GhostStats> { + self.ghosts.get(&key) + } + + pub fn ghost_count(&self) -> usize { + self.ghosts.len() + } + + /// Current observation clock (test-only access). + #[cfg(test)] + pub fn clock_debug(&self) -> u64 { + self.clock + } + + pub fn entry(&self, key: EntryKey) -> Option<&PolicyEntry> { + self.entries.get(&key) + } + + pub fn len(&self) -> usize { + self.entries.len() + } + + pub fn is_empty(&self) -> bool { + self.entries.is_empty() + } + + /// Offer a new candidate for admission. `exclusive_bytes` are bytes only + /// this entry would reference; `shared` lists `(segment, size)` segments + /// it would join (size is ignored if the segment is already registered). + /// `pinned` is the caller's active pin/hold set: pinned entries are + /// never selected for probation cap repair (#1650). + pub fn consider_admission( + &mut self, + key: EntryKey, + exclusive_bytes: u64, + shared: Vec<(SegmentId, u64)>, + cost: CostSample, + pinned: &[EntryKey], + ) -> AdmissionDecision { + // Full structural prevalidation before any policy mutation: an + // invalid cost, duplicate segment IDs, a size conflict, or an + // already-resident key must not advance the clock (aging grace) or + // expire ghosts. + if !cost.is_valid() { + return AdmissionDecision::rejected_invalid_cost(); + } + if self.entries.contains_key(&key) { + return AdmissionDecision::rejected("already-resident-key"); + } + let mut seen = std::collections::BTreeSet::new(); + if shared.iter().any(|(s, _)| !seen.insert(*s)) { + return AdmissionDecision::rejected("duplicate-segment-reference"); + } + if shared.iter().any(|(segment, size)| { + self.segments + .segment_record(*segment) + .is_some_and(|r| !r.references.contains(&key) && r.size != *size) + }) { + return AdmissionDecision::rejected("segment-size-conflict"); + } + self.clock += 1; + self.expire_ghosts(); + let mut decision = admission::consider(self, key, exclusive_bytes, shared, cost); + if decision.verdict == crate::policy::AdmissionVerdict::Admit { + // Hard probation cap is part of admission: the decision carries + // the cap-repair plan (pins excluded). Selection only — committed + // removal stays with `remove` so victims become ghosts. + decision.probation_cap_repair = self.select_probation_cap_victims(pinned); + } + decision + } + + /// Record a restore hit on an admitted entry; may promote out of probation. + pub fn record_hit(&mut self, key: EntryKey, cost: CostSample) -> Option { + // Validate before any mutation (including the clock). + if !cost.is_valid() { + return None; + } + self.clock += 1; + admission::record_hit(self, key, cost) + } + + /// Record a miss/cold recompute for an admitted entry (decays reuse). + /// A miss is a real observation that advances the clock, but unlike a + /// hit it is a negative value signal: it decays reuse and does NOT + /// refresh `last_observation`, so a miss-only stream ages the grace + /// window and the entry becomes evictable — grace cannot hold a + /// never-reused entry indefinitely. + pub fn record_miss(&mut self, key: EntryKey) { + self.clock += 1; + if let Some(entry) = self.entries.get_mut(&key) { + entry.misses += 1; + entry.observation_weight = entry.observation_weight * self.config.decay.factor + 1.0; + } + } + + /// Observe demand pressure (0 = idle, 1 = saturated). Higher pressure + /// decays reuse history faster than the observation base, so stale + /// popularity cannot pin bytes forever. + pub fn observe_pressure(&mut self, pressure: f64) { + // Non-finite pressure never poisons history: NaN is ignored, + // +inf saturates to full pressure, -inf to none. + if pressure.is_nan() { + return; + } + let pressure = if pressure == f64::INFINITY { + 1.0 + } else if pressure == f64::NEG_INFINITY { + 0.0 + } else { + pressure.clamp(0.0, 1.0) + }; + let base = self.config.decay.factor; + let reuse_factor = base * (1.0 - pressure); + for entry in self.entries.values_mut() { + entry.reuse_weight *= reuse_factor; + entry.observation_weight *= base + (1.0 - base) * pressure; + } + for ghost in self.ghosts.values_mut() { + ghost.reuse_weight *= reuse_factor; + ghost.observation_weight *= base + (1.0 - base) * pressure; + } + self.expire_ghosts(); + } + + /// Score an entry under the current statistics. Returns `None` for + /// entries with no cost observation yet. + pub fn score(&self, key: EntryKey) -> Option { + let entry = self.entries.get(&key)?; + score::compute(&self.config, key, entry, &self.segments) + } + + /// Pick eviction victims until `bytes_to_free` exclusive-and-fractional + /// bytes are released. Lowest score first, deterministic `(score, key)` + /// tie-break. Pinned entries are never chosen while an unpinned + /// candidate remains. + pub fn choose_victims( + &mut self, + bytes_to_free: u64, + pinned: &[EntryKey], + ) -> Vec<(EntryKey, EvictionVerdict)> { + admission::choose_victims(self, bytes_to_free, pinned) + } + + /// Bytes charged to the probation class: exclusive bytes plus the + /// fractional shared-segment credit. Shares are summed exactly (f64) + /// across all probation entries before a single rounding, so a small + /// segment shared by many references still charges its physical bytes + /// in aggregate — per-entry truncation cannot zero it out. + pub fn probation_bytes(&self) -> u64 { + let mut exact = 0.0f64; + for (k, e) in &self.entries { + if e.state == PolicyEntryState::Probation { + exact += e.exclusive_bytes as f64 + self.segments.fractional_bytes(*k, &e.segments); + } + } + // Deterministic round-half-up; the class charge is a hard bound, so + // we always round up any fractional residue. + exact.ceil() as u64 + } + + /// Select the no-hit probationers that must be evicted to bring the + /// probation class back under its byte cap, oldest observation first + /// (grace waived: the hard cap always wins). Selection only — this does + /// not mutate policy state; the caller commits each removal via + /// `remove`, which also records the ghost. Keys are deterministic + /// `(last_observation, key)` order. + /// + /// Pinned entries are never selected (#1650: preserve active + /// pins/holds). If pins make the cap temporarily unsatisfiable, the + /// result reports the shortfall instead of selecting a pin. + pub fn select_probation_cap_victims(&self, pinned: &[EntryKey]) -> CapRepair { + let cap = self.config.probation_byte_budget; + if self.probation_bytes() <= cap { + return CapRepair::satisfied(); + } + let mut probationers: Vec<(u64, EntryKey)> = self + .entries + .iter() + .filter(|(k, e)| { + e.state == PolicyEntryState::Probation && e.hits == 0 && !pinned.contains(k) + }) + .map(|(k, e)| (e.last_observation, *k)) + .collect(); + probationers.sort(); + // Fallback: probationers with hits still count against the class + // charge (e.g. their share rose when an admitted co-reference was + // removed), so cap repair must be able to select them too — after + // the zero-hit class, in deterministic age/key order. + let mut fallback: Vec<(u64, EntryKey)> = self + .entries + .iter() + .filter(|(k, e)| { + e.state == PolicyEntryState::Probation && e.hits > 0 && !pinned.contains(k) + }) + .map(|(k, e)| (e.last_observation, *k)) + .collect(); + fallback.sort(); + probationers.extend(fallback); + // Simulate each removal against a scratch ledger: removing a shared + // reference raises the survivors' fractional shares, so the remaining + // class charge must be recomputed, not decremented by stale shares. + let mut scratch = self.segments.clone(); + let mut removed: std::collections::BTreeSet = Default::default(); + let mut victims = Vec::new(); + for (_, key) in probationers { + let Some(entry) = self.entries.get(&key) else { + continue; + }; + scratch.release(&entry.segments, key); + removed.insert(key); + let charge: f64 = self + .entries + .iter() + .filter(|(k, e)| e.state == PolicyEntryState::Probation && !removed.contains(*k)) + .map(|(k, e)| e.exclusive_bytes as f64 + scratch.fractional_bytes(*k, &e.segments)) + .sum(); + victims.push(key); + if charge.ceil() as u64 <= cap { + break; + } + } + // After simulating every unpinned candidate, the class may still be + // over cap because the remainder is pinned (or candidateless). Report + // the shortfall explicitly instead of ever selecting a pin. + let final_charge: u64 = self + .entries + .iter() + .filter(|(k, e)| e.state == PolicyEntryState::Probation && !removed.contains(*k)) + .map(|(k, e)| e.exclusive_bytes as f64 + scratch.fractional_bytes(*k, &e.segments)) + .sum::() + .ceil() as u64; + if final_charge <= cap { + CapRepair::satisfied_with(victims) + } else { + CapRepair::deferred(victims, final_charge - cap) + } + } + + /// Test helper: run `f` with a temporarily different probation budget. + #[cfg(test)] + pub fn with_probation_budget(&mut self, budget: u64, f: impl FnOnce(&Self) -> R) -> R { + let saved = std::mem::replace( + // Safety of construction: same struct, one field changed. + &mut self.config.probation_byte_budget, + budget, + ); + let result = f(self); + self.config.probation_byte_budget = saved; + result + } + + /// Test-only convenience: cap enforcement with an empty pin set for the + /// comparison harness/traces, which model unpinned workloads. Store + /// callers holding pins must call `select_probation_cap_victims` with + /// their pin set and commit via `remove`. + #[cfg(test)] + pub fn enforce_probation_cap(&mut self) -> Vec { + let victims = self.select_probation_cap_victims(&[]).victims().to_vec(); + for key in victims.iter().copied() { + let _ = self.remove(key, &[]); + } + victims + } + + /// Store-facing removal: releases the entry's fractional segment credit + /// and stashes its reuse statistics as a ghost so a recurrence is + /// recognized as a value signal. Removing an admitted co-reference can + /// raise survivors' shares over the probation cap, so the removal + /// response carries the cap-repair plan; commit its victims through + /// further `remove` calls. `pinned` is the caller's active pin/hold + /// set: pins are never selected (#1650). + pub fn remove(&mut self, key: EntryKey, pinned: &[EntryKey]) -> Option { + let entry = self.entries.remove(&key)?; + self.segments.release(&entry.segments, key); + self.insert_ghost( + key, + GhostStats { + hits: entry.hits, + reuse_weight: entry.reuse_weight, + observation_weight: entry.observation_weight, + last_observation: self.clock, + }, + ); + let cap_repair = self.select_probation_cap_victims(pinned); + Some(RemovalOutcome { + entry, + probation_cap_repair: cap_repair, + }) + } + + /// Remove without returning cap-repair victims (test-only harness use + /// where the caller re-selects the cap itself). External callers must + /// use `remove` so the hard probation cap cannot be bypassed. + #[cfg(test)] + pub fn remove_without_cap_repair(&mut self, key: EntryKey) -> Option { + let outcome = self.remove(key, &[])?; + Some(outcome.entry) + } +} diff --git a/crates/skippy-cache/src/policy/score.rs b/crates/skippy-cache/src/policy/score.rs new file mode 100644 index 0000000000..b0c8dbc724 --- /dev/null +++ b/crates/skippy-cache/src/policy/score.rs @@ -0,0 +1,50 @@ +//! The benefit-per-exclusive-byte score (#1650 first slice). + +use crate::policy::{CostSample, EntryKey, PolicyConfig, PolicyEntry, SharedSegmentLedger}; + +/// Raw inputs and the resulting score, for tests and opaque logging. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct ScoreInputs { + pub reuse_probability: f64, + pub net_benefit: f64, + pub exclusive_bytes: f64, +} + +/// A computed score with its inputs. Higher is better; ordering by +/// `(value, entry_key)` is the canonical deterministic order. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct BenefitScore { + pub value: f64, + pub inputs: ScoreInputs, +} + +pub(crate) fn compute( + config: &PolicyConfig, + key: EntryKey, + entry: &PolicyEntry, + ledger: &SharedSegmentLedger, +) -> Option { + let cost: CostSample = entry.last_cost?; + if !cost.is_valid() { + return None; + } + let reuse = entry.reuse_probability().max(config.min_reuse_probability); + let shared = ledger.fractional_bytes(key, &entry.segments); + let exclusive = entry.exclusive_bytes as f64 + shared; + if exclusive <= 0.0 { + return None; + } + let net = cost.net_benefit(); + let value = reuse * net / exclusive; + if !value.is_finite() { + return None; + } + Some(BenefitScore { + value, + inputs: ScoreInputs { + reuse_probability: reuse, + net_benefit: net, + exclusive_bytes: exclusive, + }, + }) +} diff --git a/crates/skippy-cache/src/policy/tests.rs b/crates/skippy-cache/src/policy/tests.rs new file mode 100644 index 0000000000..6a6b03df49 --- /dev/null +++ b/crates/skippy-cache/src/policy/tests.rs @@ -0,0 +1,1156 @@ +//! Unit and comparison tests for the benefit policy (#1650 first slice). + +use super::*; +use crate::policy::admission::AdmissionDecisionKind; +use crate::policy::lru_baseline::LruCache; +use crate::policy::traces; + +fn cost(cold: f64, restore: f64) -> CostSample { + CostSample { + cold_prefill_cost: cold, + restore_cost: restore, + } +} + +#[test] +fn rejects_entries_with_no_net_benefit() { + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + let decision = policy.consider_admission(1, 1 << 20, vec![], cost(100.0, 150.0), &[]); + assert_eq!(decision.kind, AdmissionDecisionKind::Reject); + assert!(policy.is_empty()); +} + +#[test] +fn new_entries_start_in_probation_and_promote_on_second_hit() { + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + policy.consider_admission(1, 1 << 20, vec![], cost(400.0, 120.0), &[]); + assert_eq!(policy.entry(1).unwrap().state, PolicyEntryState::Probation); + + assert!(policy.record_hit(1, cost(400.0, 120.0)).is_none()); // first hit + let promoted = policy.record_hit(1, cost(400.0, 120.0)).unwrap(); // second + assert_eq!(promoted.kind, AdmissionDecisionKind::Promote); + assert_eq!(policy.entry(1).unwrap().state, PolicyEntryState::Admitted); +} + +#[test] +fn score_divides_by_exclusive_bytes() { + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + policy.consider_admission(1, 2 << 20, vec![], cost(300.0, 100.0), &[]); + policy.record_hit(1, cost(300.0, 100.0)); + policy.record_hit(1, cost(300.0, 100.0)); + let score = policy.score(1).unwrap(); + // reuse=1.0, net=200.0, exclusive=2MiB + assert!((score.inputs.net_benefit - 200.0).abs() < 1e-9); + assert!((score.inputs.exclusive_bytes - (2.0 * 1024.0 * 1024.0)).abs() < 1e-9); + assert!((score.value - 200.0 / (2.0 * 1024.0 * 1024.0)).abs() < 1e-9); +} + +#[test] +fn shared_segments_get_fractional_credit_in_score() { + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + let seg = (7u64, 3 << 20); + policy.consider_admission(1, 0, vec![seg], cost(400.0, 100.0), &[]); + policy.consider_admission(2, 0, vec![seg], cost(400.0, 100.0), &[]); + policy.consider_admission(3, 0, vec![seg], cost(400.0, 100.0), &[]); + // Each entry charges 1MiB of the shared 3MiB segment. + assert_eq!(policy.segments.total_bytes(), 3 << 20); + for key in [1u64, 2, 3] { + let score = policy.score(key).unwrap(); + assert!((score.inputs.exclusive_bytes - (1.0 * 1024.0 * 1024.0)).abs() < 1e-9); + } +} + +#[test] +fn eviction_picks_lowest_score_deterministically() { + // Grace off: force the score order. + let mut policy = BenefitPolicy::new(PolicyConfig { + grace_observations: 0, + ..PolicyConfig::default() + }); + // Same value, different footprint: smaller footprint -> higher score. + policy.consider_admission(1, 8 << 20, vec![], cost(400.0, 100.0), &[]); + policy.consider_admission(2, 1 << 20, vec![], cost(400.0, 100.0), &[]); + for key in [1u64, 2] { + policy.record_hit(key, cost(400.0, 100.0)); + policy.record_hit(key, cost(400.0, 100.0)); + } + let victims = policy.choose_victims(8 << 20, &[]); + assert_eq!(victims.first().map(|(k, _)| *k), Some(1)); +} + +#[test] +fn pinned_entries_are_not_victims() { + // Grace off: force the pinned check. + let mut policy = BenefitPolicy::new(PolicyConfig { + grace_observations: 0, + ..PolicyConfig::default() + }); + policy.consider_admission(1, 1 << 20, vec![], cost(400.0, 100.0), &[]); + policy.consider_admission(2, 1 << 20, vec![], cost(400.0, 100.0), &[]); + let victims = policy.choose_victims(1 << 20, &[2]); + assert_eq!(victims.iter().map(|(k, _)| *k).collect::>(), vec![1]); +} + +#[test] +fn decay_shrinks_stale_reuse_probability() { + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + policy.consider_admission(1, 1 << 20, vec![], cost(400.0, 100.0), &[]); + for _ in 0..10 { + policy.record_hit(1, cost(400.0, 100.0)); + } + let hot = policy.entry(1).unwrap().reuse_probability(); + for _ in 0..50 { + policy.observe_pressure(1.0); + } + let stale = policy.entry(1).unwrap().reuse_probability(); + assert!(stale < hot); +} + +#[test] +fn policy_decisions_are_deterministic_across_runs() { + let trace = traces::zipf_hotset_trace(42, 500); + let run = || { + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + let mut evictions = Vec::new(); + for access in &trace { + if policy.entry(access.entry).is_some() { + policy.record_hit( + access.entry, + cost(access.cold_prefill_cost, access.restore_cost), + ); + } else { + policy.consider_admission( + access.entry, + access.exclusive_bytes, + vec![], + cost(access.cold_prefill_cost, access.restore_cost), + &[], + ); + } + for (key, verdict) in policy.choose_victims(1 << 30, &[]) { + if verdict == EvictionVerdict::Evict { + evictions.push(key); + policy.remove(key, &[]); + } + } + } + evictions + }; + assert_eq!(run(), run()); +} + +/// Policy-vs-LRU comparison metrics over a trace at matched capacity. +struct Comparison { + policy_saved_cost: f64, + lru_saved_cost: f64, + policy_bytes_written: u64, + lru_bytes_written: u64, +} + +fn compare(trace: &[traces::TraceAccess], capacity_bytes: u64) -> Comparison { + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + let mut lru = LruCache::new(capacity_bytes); + let mut policy_saved = 0.0f64; + let mut policy_written = 0u64; + + for access in trace { + if policy.entry(access.entry).is_some() { + let decision = policy.record_hit( + access.entry, + cost(access.cold_prefill_cost, access.restore_cost), + ); + policy_saved += (access.cold_prefill_cost - access.restore_cost).max(0.0); + // Promotion is the persistence event for a probational entry: + // the bytes written are the resident entry's retained size at + // admission, not the current access's (re-sampled) size. + if let Some(AdmissionDecision { + kind: AdmissionDecisionKind::Promote, + .. + }) = &decision + { + let persisted = policy + .entry(access.entry) + .map(|e| e.exclusive_bytes) + .expect("promoted entry resident"); + policy_written += persisted; + } + } else { + // Re-offer a rejected/evicted entry each time it recurs; + // probation keeps new candidates in the accounting set until + // pressure forces a choice. + let decision = policy.consider_admission( + access.entry, + access.exclusive_bytes, + vec![], + cost(access.cold_prefill_cost, access.restore_cost), + &[], + ); + // Count bytes only when the decision actually persists. + // AdmitProbation is memory-only accounting; the entry becomes + // written when a later hit promotes it to Admitted (persist). + if decision.kind == AdmissionDecisionKind::AdmitPersist { + policy_written += access.exclusive_bytes; + } + // Hard probation cap is enforced as part of admission: commit the + // selected victims. + for key in decision.probation_cap_repair.victims().iter().copied() { + policy.remove(key, &[]); + } + } + let mut used: u64 = policy.entries.values().map(|e| e.exclusive_bytes).sum(); + while used > capacity_bytes { + let victims = policy.choose_victims(used - capacity_bytes, &[]); + if victims.is_empty() { + break; + } + for (key, verdict) in victims { + if verdict == EvictionVerdict::Evict { + policy.remove(key, &[]); + } + } + used = policy.entries.values().map(|e| e.exclusive_bytes).sum(); + } + lru.access(access); + } + Comparison { + policy_saved_cost: policy_saved, + lru_saved_cost: lru.saved_cost, + policy_bytes_written: policy_written, + lru_bytes_written: lru.bytes_written, + } +} + +/// The acceptance direction the issue requires: on a trace where one-shot +/// large entries pollute an LRU, the benefit policy must STRICTLY save +/// more cold prefill cost or STRICTLY write fewer persisted bytes, with +/// the non-winning dimension held inside an explicit regression +/// tolerance. +#[test] +fn beats_lru_on_one_shot_pollution_trace() { + let trace = traces::one_shot_trace(7, 2_000); + let capacity = 64 << 20; + let comparison = compare(&trace, capacity); + const TOLERANCE: f64 = 0.05; + let wins_saved = comparison.policy_saved_cost > comparison.lru_saved_cost; + let wins_written = comparison.policy_bytes_written < comparison.lru_bytes_written; + // Strict improvement in at least one dimension (#1650 acceptance). + assert!( + wins_saved || wins_written, + "one-shot pollution @ {capacity}: saved policy={:.1} lru={:.1}, written policy={} lru={} — no strict improvement", + comparison.policy_saved_cost, + comparison.lru_saved_cost, + comparison.policy_bytes_written, + comparison.lru_bytes_written + ); + // The non-winning dimension must stay inside an explicit regression + // tolerance rather than degrading unboundedly. + if !wins_saved { + assert!( + comparison.policy_saved_cost >= comparison.lru_saved_cost * (1.0 - TOLERANCE), + "policy saved {} vs lru {} — saved cost regressed beyond {:.0}%", + comparison.policy_saved_cost, + comparison.lru_saved_cost, + TOLERANCE * 100.0 + ); + } + if !wins_written { + assert!( + (comparison.policy_bytes_written as f64) + <= comparison.lru_bytes_written as f64 * (1.0 + TOLERANCE), + "policy wrote {} vs lru {} — writes regressed beyond {:.0}%", + comparison.policy_bytes_written, + comparison.lru_bytes_written, + TOLERANCE * 100.0 + ); + } +} + +#[test] +fn no_regression_on_turn_growth_trace() { + let trace = traces::turn_growth_trace(11, 4, 12); + let capacity = 64 << 20; + let comparison = compare(&trace, capacity); + assert!( + comparison.policy_saved_cost >= comparison.lru_saved_cost * 0.95, + "policy saved {} vs lru {}", + comparison.policy_saved_cost, + comparison.lru_saved_cost + ); +} + +#[test] +fn no_regression_on_zipf_hotset_trace() { + let trace = traces::zipf_hotset_trace(3, 4_000); + let capacity = 64 << 20; + let comparison = compare(&trace, capacity); + assert!( + comparison.policy_saved_cost >= comparison.lru_saved_cost * 0.95, + "policy saved {} vs lru {}", + comparison.policy_saved_cost, + comparison.lru_saved_cost + ); +} + +#[test] +fn no_regression_on_mixed_size_trace() { + let trace = traces::mixed_size_trace(5, 3_000); + let capacity = 256 << 20; + let comparison = compare(&trace, capacity); + assert!( + comparison.policy_saved_cost >= comparison.lru_saved_cost * 0.95, + "policy saved {} vs lru {}", + comparison.policy_saved_cost, + comparison.lru_saved_cost + ); +} + +/// The mixed-size trace must actually exercise mixed resident size +/// classes and large-entry eviction at a capacity smaller than the +/// 256 MiB class. +#[test] +fn mixed_size_trace_has_real_mixed_residency_and_large_eviction() { + let trace = traces::mixed_size_trace(5, 300); + // Distinct keys per class, each consistently sized. + let mut sizes = std::collections::BTreeMap::new(); + let mut classes = std::collections::BTreeSet::new(); + for a in &trace { + let prev = sizes.insert(a.entry, a.exclusive_bytes); + assert!( + prev.is_none_or(|s| s == a.exclusive_bytes), + "key {} changed size class", + a.entry + ); + classes.insert(a.exclusive_bytes); + } + assert_eq!(classes.len(), 3, "all three size classes present"); + // Run the trace at a capacity that cannot hold a 256 MiB entry + // alongside the hot 4 MiB set: large entries must be admitted + // (probation) and evicted, never violating capacity. + let capacity = 96 << 20; + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + let mut resident_classes = std::collections::BTreeSet::new(); + for access in &trace { + if policy.entry(access.entry).is_some() { + policy.record_hit( + access.entry, + cost(access.cold_prefill_cost, access.restore_cost), + ); + } else { + let decision = policy.consider_admission( + access.entry, + access.exclusive_bytes, + vec![], + cost(access.cold_prefill_cost, access.restore_cost), + &[], + ); + for key in decision.probation_cap_repair.victims().iter().copied() { + policy.remove(key, &[]); + } + } + let mut used: u64 = policy.entries.values().map(|e| e.exclusive_bytes).sum(); + while used > capacity { + let victims = policy.choose_victims(used - capacity, &[]); + if victims.is_empty() { + break; + } + for (key, verdict) in victims { + if verdict == EvictionVerdict::Evict { + policy.remove(key, &[]); + } + } + used = policy.entries.values().map(|e| e.exclusive_bytes).sum(); + } + assert!(used <= capacity, "used {} over capacity after access", used); + for e in policy.entries.values() { + resident_classes.insert(e.exclusive_bytes); + } + } + assert!( + resident_classes.len() >= 2, + "mixed resident sizes never observed: {:?}", + resident_classes + ); +} + +#[test] +fn invalid_costs_and_config_are_rejected_not_panicked() { + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + let nan = f64::NAN; + policy.consider_admission( + 1, + 1 << 20, + vec![], + CostSample { + cold_prefill_cost: nan, + restore_cost: 10.0, + }, + &[], + ); + assert!(policy.score(1).is_none()); // NaN never enters the ordering + assert!( + !PolicyConfig { + min_reuse_probability: f64::NAN, + ..PolicyConfig::default() + } + .is_valid() + ); + assert!( + !PolicyConfig { + persistence_hit_threshold: 0, + ..PolicyConfig::default() + } + .is_valid() + ); + assert!( + !PolicyConfig { + decay: DecayConfig { factor: 0.0 }, + ..PolicyConfig::default() + } + .is_valid() + ); + assert!( + !PolicyConfig { + decay: DecayConfig { factor: 1.5 }, + ..PolicyConfig::default() + } + .is_valid() + ); +} + +#[test] +fn choose_victims_never_returns_a_duplicate_key() { + let mut policy = BenefitPolicy::new(PolicyConfig { + grace_observations: 0, + ..PolicyConfig::default() + }); + for key in 1..=8u64 { + policy.consider_admission(key, 1 << 20, vec![], cost(400.0, 100.0), &[]); + } + let victims = policy.choose_victims(u64::MAX, &[]); + let keys: Vec = victims.iter().map(|(k, _)| *k).collect(); + let mut sorted = keys.clone(); + sorted.sort(); + sorted.dedup(); + assert_eq!(keys.len(), sorted.len(), "duplicate victim keys"); + assert_eq!(keys.len(), 8); +} + +#[test] +fn victim_selection_counts_marginal_physical_bytes_of_shared_segments() { + // Three entries share one segment; only evicting the last reference + // physically frees it. Victim selection must keep choosing until the + // requested *physical* bytes are covered. + let mut policy = BenefitPolicy::new(PolicyConfig { + grace_observations: 0, + ..PolicyConfig::default() + }); + let seg = (1u64, 3 << 20); + policy.consider_admission(1, 0, vec![seg], cost(400.0, 100.0), &[]); + policy.consider_admission(2, 0, vec![seg], cost(400.0, 100.0), &[]); + policy.consider_admission(3, 0, vec![seg], cost(400.0, 100.0), &[]); + for key in [1u64, 2, 3] { + policy.record_hit(key, cost(400.0, 100.0)); + policy.record_hit(key, cost(400.0, 100.0)); + } + // Freeing 3 MiB must select all three references, not one (each alone + // releases nothing physical). + let victims = policy.choose_victims(3 << 20, &[]); + assert_eq!(victims.len(), 3); + // Freeing 1 MiB: marginal release of two of the three is 0, the third is + // 3 MiB; the loop must not stop before the target is met. + let victims = policy.choose_victims(1 << 20, &[]); + assert_eq!(victims.len(), 3); +} + +#[test] +fn probation_byte_cap_is_enforced_over_grace() { + // Grace must never hold the policy over the hard probation byte cap. + // Cap fits two small entries only. + let mut policy = BenefitPolicy::new(PolicyConfig { + probation_byte_budget: 2 << 20, + ..PolicyConfig::default() + }); + for key in 1..=10u64 { + policy.consider_admission(key, 1 << 20, vec![], cost(400.0, 100.0), &[]); + } + assert!(policy.probation_bytes() > 2 << 20); + for key in policy.enforce_probation_cap() { + policy.remove(key, &[]); + } + assert!( + policy.probation_bytes() <= 2 << 20, + "probation bytes {} over cap {}", + policy.probation_bytes(), + 2 << 20 + ); +} + +#[test] +fn recurrence_after_eviction_is_a_value_signal() { + // An entry evicted before its second hit must not restart from zero + // history: its recurrence carries ghost statistics and counts as a + // reuse observation (issue: "second-hit or equivalent value signal"). + let mut policy = BenefitPolicy::new(PolicyConfig { + grace_observations: 0, + ..PolicyConfig::default() + }); + policy.consider_admission(1, 1 << 20, vec![], cost(400.0, 100.0), &[]); + assert!(policy.record_hit(1, cost(400.0, 100.0)).is_none()); // first hit + policy.remove(1, &[]); // evicted before the second hit + policy.consider_admission(1, 1 << 20, vec![], cost(400.0, 100.0), &[]); + assert_eq!( + policy.entry(1).unwrap().state, + PolicyEntryState::Admitted, + "recurrence with ghost history must re-admit as a value signal" + ); +} + +#[test] +fn invalid_cost_samples_never_mutate_entry_state() { + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + let nan = f64::NAN; + // Admission with an invalid sample must not insert. + let decision = policy.consider_admission( + 1, + 1 << 20, + vec![], + CostSample { + cold_prefill_cost: nan, + restore_cost: 10.0, + }, + &[], + ); + assert_eq!(decision.verdict, AdmissionVerdict::Reject); + assert!(policy.is_empty()); + // A valid admission followed by an invalid hit must not poison state. + policy.consider_admission(1, 1 << 20, vec![], cost(400.0, 100.0), &[]); + let before = policy.entry(1).unwrap().clone(); + assert!( + policy + .record_hit( + 1, + CostSample { + cold_prefill_cost: 400.0, + restore_cost: nan + } + ) + .is_none() + ); + assert_eq!(policy.entry(1).unwrap().last_cost, before.last_cost); + assert_eq!(policy.entry(1).unwrap().hits, before.hits); + // NaN pressure is ignored entirely. + policy.observe_pressure(nan); + assert_eq!(policy.entry(1).unwrap().reuse_weight, before.reuse_weight); +} + +#[test] +fn probation_cap_counts_shared_charge_and_is_enforced_by_admission() { + // Probation entries backed entirely by shared segments charge fractional + // bytes and must not grow without bound. + let mut policy = BenefitPolicy::new(PolicyConfig { + probation_byte_budget: 1 << 20, // smaller than one shared segment + grace_observations: 0, + ..PolicyConfig::default() + }); + let mut victims = Vec::new(); + for key in 1..=6u64 { + let decision = + policy.consider_admission(key, 0, vec![(1u64, 3 << 20)], cost(400.0, 100.0), &[]); + for v in decision.probation_cap_repair.victims().iter().copied() { + policy.remove(v, &[]); + victims.push(v); + } + } + assert!( + policy.probation_bytes() <= 1 << 20, + "probation bytes {} over cap", + policy.probation_bytes() + ); + assert!(!victims.is_empty(), "shared-only probation must be capped"); +} + +#[test] +fn ghosts_are_bounded_by_count_and_age() { + // Count bound: one-shot keys must not create permanent metadata. + let mut policy = BenefitPolicy::new(PolicyConfig { + ghost_capacity: 8, + ghost_max_age_observations: 100, + grace_observations: 0, + ..PolicyConfig::default() + }); + for key in 0..64u64 { + policy.consider_admission(key, 1 << 20, vec![], cost(400.0, 100.0), &[]); + for (_, verdict) in policy.choose_victims(u64::MAX, &[]) { + if verdict == EvictionVerdict::Evict { + policy.remove(key, &[]); + break; + } + } + } + assert!( + policy.ghost_count() <= 8, + "ghost count {}", + policy.ghost_count() + ); + + // Age bound: stale popularity cannot revive indefinitely. + let mut policy = BenefitPolicy::new(PolicyConfig { + ghost_capacity: 1024, + ghost_max_age_observations: 10, + grace_observations: 0, + ..PolicyConfig::default() + }); + policy.consider_admission(99, 1 << 20, vec![], cost(400.0, 100.0), &[]); + policy.remove(99, &[]); + assert!(policy.ghost(99).is_some()); + for i in 0..50u64 { + // Unique keys: an already-resident key is rejected before the clock + // advances, so age must be driven by fresh observations. + policy.consider_admission(1000 + i, 1 << 20, vec![], cost(400.0, 100.0), &[]); + } + assert!( + policy.ghost(99).is_none(), + "old ghost must expire via age bound" + ); +} + +#[test] +fn probation_cap_selection_is_not_committed_removal() { + // Selection must leave state untouched so callers can commit physically; + // committed removal through `remove` records the ghost. + let mut policy = BenefitPolicy::new(PolicyConfig { + probation_byte_budget: 1 << 20, + grace_observations: 0, + ..PolicyConfig::default() + }); + for key in 1..=4u64 { + policy.consider_admission(key, 1 << 20, vec![], cost(400.0, 100.0), &[]); + } + let victims = policy.select_probation_cap_victims(&[]).victims().to_vec(); + assert!(!victims.is_empty()); + assert_eq!(policy.len(), 4, "selection must not remove entries"); + let decision = policy.consider_admission(5, 1 << 20, vec![], cost(400.0, 100.0), &[]); + assert!(!decision.probation_cap_repair.victims().is_empty()); + for key in decision.probation_cap_repair.victims() { + policy.remove(*key, &[]); + } + // Committed removals become ghosts (bounded). + assert!(policy.ghost_count() > 0); +} + +#[test] +fn invalid_observation_leaves_clock_and_ghosts_untouched() { + let nan = f64::NAN; + let mut policy = BenefitPolicy::new(PolicyConfig { + grace_observations: 2, + ghost_max_age_observations: 1, + ..PolicyConfig::default() + }); + // Seed one ghost. + policy.consider_admission(9, 1 << 20, vec![], cost(400.0, 100.0), &[]); + policy.remove(9, &[]); + assert!(policy.ghost(9).is_some()); + + // Invalid admission: clock must not advance, ghost must survive + // (age bound is 1, so a real observation would have expired it). + let before_clock = policy.clock_debug(); + let decision = policy.consider_admission( + 1, + 1 << 20, + vec![], + CostSample { + cold_prefill_cost: nan, + restore_cost: 10.0, + }, + &[], + ); + assert_eq!(decision.verdict, AdmissionVerdict::Reject); + assert_eq!(policy.clock_debug(), before_clock); + assert!(policy.ghost(9).is_some()); + + // Invalid hit: same invariants. + policy.consider_admission(2, 1 << 20, vec![], cost(400.0, 100.0), &[]); + let before_clock = policy.clock_debug(); + assert!( + policy + .record_hit( + 2, + CostSample { + cold_prefill_cost: nan, + restore_cost: 10.0 + } + ) + .is_none() + ); + assert_eq!(policy.clock_debug(), before_clock); +} + +#[test] +fn many_reference_small_segment_still_charges_probation_bytes() { + // A 4-byte segment referenced by 9 probation entries: per-entry + // truncation would charge 0; the class total must still be 4. + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + for key in 1..=9u64 { + policy.consider_admission(key, 0, vec![(1u64, 4)], cost(400.0, 100.0), &[]); + } + assert!( + policy.probation_bytes() >= 4, + "charged {}", + policy.probation_bytes() + ); + // And with a 1-byte cap, admission must select shared-only victims. + let mut policy = BenefitPolicy::new(PolicyConfig { + probation_byte_budget: 0, + ..PolicyConfig::default() + }); + let decision = policy.consider_admission(1, 0, vec![(1u64, 4)], cost(400.0, 100.0), &[]); + assert!(!decision.probation_cap_repair.victims().is_empty()); +} + +#[test] +fn ghost_promoted_recurrence_returns_admit_persist() { + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + policy.consider_admission(1, 1 << 20, vec![], cost(400.0, 100.0), &[]); + assert!(policy.record_hit(1, cost(400.0, 100.0)).is_none()); // hits = 1 + policy.remove(1, &[]); + let decision = policy.consider_admission(1, 1 << 20, vec![], cost(400.0, 100.0), &[]); + assert_eq!(decision.kind, AdmissionDecisionKind::AdmitPersist); + assert_eq!(policy.entry(1).unwrap().state, PolicyEntryState::Admitted); + // Non-promoted recurrence stays AdmitProbation/Probation. + policy.consider_admission(2, 1 << 20, vec![], cost(400.0, 100.0), &[]); + policy.remove(2, &[]); + let decision = policy.consider_admission(2, 1 << 20, vec![], cost(400.0, 100.0), &[]); + assert_eq!(decision.kind, AdmissionDecisionKind::AdmitProbation); + assert_eq!(policy.entry(2).unwrap().state, PolicyEntryState::Probation); +} + +#[test] +fn zero_ghost_capacity_is_a_real_zero_bound() { + let mut policy = BenefitPolicy::new(PolicyConfig { + ghost_capacity: 0, + ..PolicyConfig::default() + }); + assert!( + PolicyConfig { + ghost_capacity: 0, + ..PolicyConfig::default() + } + .is_valid() + ); + policy.consider_admission(1, 1 << 20, vec![], cost(400.0, 100.0), &[]); + policy.remove(1, &[]); + assert_eq!(policy.ghost_count(), 0, "zero capacity must retain nothing"); +} + +#[test] +fn cap_victim_selection_covers_recomputed_shared_shares() { + // Removing shared references raises survivors' shares: the selected set + // must actually bring the class under cap once committed. + let mut policy = BenefitPolicy::new(PolicyConfig { + probation_byte_budget: 9 << 20, + grace_observations: 0, + ..PolicyConfig::default() + }); + // Three no-hit probationers, each 1 MiB exclusive + a shared 9 MiB + // segment (3 MiB share each → class charge 3*(1+3) = 12 MiB > 9 MiB). + for key in 1..=3u64 { + let decision = + policy.consider_admission(key, 1 << 20, vec![(7u64, 9 << 20)], cost(400.0, 100.0), &[]); + for v in decision.probation_cap_repair.victims().iter().copied() { + policy.remove(v, &[]); + } + } + assert!( + policy.probation_bytes() <= 9 << 20, + "committed class charge {} over cap", + policy.probation_bytes() + ); + // Removal must have happened through the decision path. + assert!(policy.len() < 3); +} + +#[test] +fn duplicate_segment_references_are_rejected() { + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + let decision = policy.consider_admission( + 1, + 0, + vec![(42u64, 100), (42u64, 100)], + cost(400.0, 100.0), + &[], + ); + assert_eq!(decision.verdict, AdmissionVerdict::Reject); + assert!( + decision + .reasons + .contains(&"duplicate-segment-reference".to_string()) + ); + assert!(policy.is_empty()); + // Conflicting size for a known segment is also rejected. + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + policy.consider_admission(1, 0, vec![(42u64, 100)], cost(400.0, 100.0), &[]); + let decision = policy.consider_admission(2, 0, vec![(42u64, 200)], cost(400.0, 100.0), &[]); + assert_eq!(decision.verdict, AdmissionVerdict::Reject); + assert!( + decision + .reasons + .contains(&"segment-size-conflict".to_string()) + ); + // Same size for a known segment is fine. + let decision = policy.consider_admission(3, 0, vec![(42u64, 100)], cost(400.0, 100.0), &[]); + assert_eq!(decision.verdict, AdmissionVerdict::Admit); + // One 100-byte segment: class charge is exactly 100. + assert_eq!(policy.probation_bytes(), 100); +} + +#[test] +fn cap_victim_selection_reproduces_reported_counterexample() { + // Exact shape of the reported probe: before=10,590,618 over + // cap=9,437,184 (9 MiB); stale-share subtraction selected [1,2] and + // left the committed class at 10,354,688. Reproduce it with concrete + // numbers: 10 MiB cap basis scaled to 9 MiB via three probationers — + // 1 MiB exclusive each (3 MiB) plus one shared 8 MiB segment + // (8/3 MiB per share -> class ~3+8=11 MiB before, over cap). + let mut policy = BenefitPolicy::new(PolicyConfig { + probation_byte_budget: 9 << 20, + grace_observations: 0, + ..PolicyConfig::default() + }); + // Admission applies cap victims incrementally, so build the over-cap + // state with cap victims disabled (huge probation budget), then swap + // in the real cap and select. + for key in 1..=3u64 { + policy.consider_admission(key, 1 << 20, vec![(7u64, 8 << 20)], cost(400.0, 100.0), &[]); + } + let before = policy.probation_bytes(); + assert!(before > 9 << 20, "before {}", before); + // Now select against the real cap by constructing the over-cap state + // through the public API: reset the budget by direct selection. + let victims = policy + .with_probation_budget(9 << 20, |p| p.select_probation_cap_victims(&[])) + .victims() + .to_vec(); + assert!(!victims.is_empty()); + for key in &victims { + policy.remove(*key, &[]); + } + let after = policy.probation_bytes(); + assert!( + after <= 9 << 20, + "committed class {} still over cap after victims {:?}", + after, + victims + ); +} + +#[test] +fn conflict_on_a_later_segment_leaves_state_untouched() { + // Admit segment 1 at size 100, then offer a new key whose later segment + // conflicts: [(2,100),(1,200)]. The reject must leave the ledger without + // segment 2, no entry for the rejected key, and any matching ghost intact. + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + policy.consider_admission(1, 0, vec![(1u64, 100)], cost(400.0, 100.0), &[]); + + // Seed a ghost for the key that will be rejected so ghost survival is + // observable. + policy.consider_admission(9, 1 << 20, vec![], cost(400.0, 100.0), &[]); + policy.remove(9, &[]); + assert!(policy.ghost(9).is_some()); + + let decision = policy.consider_admission( + 9, + 1 << 20, + vec![(2u64, 100), (1u64, 200)], + cost(400.0, 100.0), + &[], + ); + assert_eq!(decision.verdict, AdmissionVerdict::Reject); + assert!( + decision + .reasons + .contains(&"segment-size-conflict".to_string()) + ); + // No entry for the rejected key; the ghost survived the reject. + assert!(policy.entry(9).is_none()); + assert!( + policy.ghost(9).is_some(), + "rejected re-offer must not lose the ghost" + ); + // Segment 2 was never registered. + assert!(policy.segments.segment_record(2).is_none()); + // Segment 1 still has exactly one reference (the original entry). + let record = policy.segments.segment_record(1).expect("segment 1 intact"); + assert_eq!(record.references, std::collections::BTreeSet::from([1u64])); + assert_eq!(record.size, 100); + // Class charge unchanged: the original entry's shared 100 bytes only. + assert_eq!(policy.probation_bytes(), 100); +} + +#[test] +fn structural_rejects_leave_clock_and_zero_horizon_ghosts_untouched() { + // ghost_max_age_observations = 0: any real observation would expire the + // ghost. Duplicate-ID and size-conflict rejects must not. + let mk = || { + BenefitPolicy::new(PolicyConfig { + ghost_max_age_observations: 0, + grace_observations: 4, + ..PolicyConfig::default() + }) + }; + + // Duplicate segment IDs. + let mut policy = mk(); + policy.consider_admission(9, 1 << 20, vec![], cost(400.0, 100.0), &[]); + policy.remove(9, &[]); + assert!(policy.ghost(9).is_some()); + let clock = policy.clock_debug(); + let decision = policy.consider_admission( + 1, + 0, + vec![(42u64, 100), (42u64, 100)], + cost(400.0, 100.0), + &[], + ); + assert_eq!(decision.verdict, AdmissionVerdict::Reject); + assert_eq!( + policy.clock_debug(), + clock, + "duplicate reject must not advance clock" + ); + assert!( + policy.ghost(9).is_some(), + "duplicate reject must not expire ghosts" + ); + + // Size conflict on a later segment. + let mut policy = mk(); + policy.consider_admission(1, 0, vec![(1u64, 100)], cost(400.0, 100.0), &[]); + policy.consider_admission(9, 1 << 20, vec![], cost(400.0, 100.0), &[]); + policy.remove(9, &[]); + assert!(policy.ghost(9).is_some()); + let clock = policy.clock_debug(); + let decision = policy.consider_admission( + 9, + 1 << 20, + vec![(2u64, 100), (1u64, 200)], + cost(400.0, 100.0), + &[], + ); + assert_eq!(decision.verdict, AdmissionVerdict::Reject); + assert_eq!( + policy.clock_debug(), + clock, + "conflict reject must not advance clock" + ); + assert!( + policy.ghost(9).is_some(), + "conflict reject must not expire ghosts" + ); +} + +#[test] +fn already_resident_key_rejection_leaves_no_stale_references() { + let mut policy = BenefitPolicy::new(PolicyConfig::default()); + policy.consider_admission(1, 0, vec![(1u64, 100)], cost(400.0, 100.0), &[]); + // Re-admitting a resident key is rejected and must not swap the entry's + // segment list or leak old ledger references. + let decision = policy.consider_admission(1, 0, vec![(2u64, 50)], cost(400.0, 100.0), &[]); + assert_eq!(decision.verdict, AdmissionVerdict::Reject); + assert!( + decision + .reasons + .contains(&"already-resident-key".to_string()) + ); + let record = policy.segments.segment_record(1).expect("segment 1 intact"); + assert_eq!(record.references, std::collections::BTreeSet::from([1u64])); + assert!( + policy.segments.segment_record(2).is_none(), + "segment 2 must not be registered" + ); + assert_eq!(policy.probation_bytes(), 100); +} + +#[test] +fn admitted_coreference_removal_repairs_the_probation_cap() { + // Ghost-promoted admitted A on shared segment S; probation P shares S + // (half-share fits the cap); one hit on P (still probation at + // threshold 2); remove A -> P's charge rises to all of S and exceeds + // the cap. The removal response must carry P as a cap victim. + let mut policy = BenefitPolicy::new(PolicyConfig { + probation_byte_budget: 1 << 20, // 1 MiB cap; S is 1.5 MiB + grace_observations: 0, + ..PolicyConfig::default() + }); + // Build A as ghost-promoted: admit, one hit, evict (ghost), recur. + policy.consider_admission(1, 0, vec![(7u64, (3 << 20) / 2)], cost(400.0, 100.0), &[]); + policy.record_hit(1, cost(400.0, 100.0)); + policy.remove_without_cap_repair(1); + let decision = + policy.consider_admission(1, 0, vec![(7u64, (3 << 20) / 2)], cost(400.0, 100.0), &[]); + assert_eq!(decision.kind, AdmissionDecisionKind::AdmitPersist); + for v in decision.probation_cap_repair.victims().iter().copied() { + policy.remove_without_cap_repair(v); + } + assert_eq!(policy.entry(1).unwrap().state, PolicyEntryState::Admitted); + + // P shares S: half-share = 0.75 MiB fits the 1 MiB cap. + let decision = + policy.consider_admission(2, 0, vec![(7u64, (3 << 20) / 2)], cost(400.0, 100.0), &[]); + assert_eq!(decision.verdict, AdmissionVerdict::Admit); + for v in decision.probation_cap_repair.victims().iter().copied() { + policy.remove_without_cap_repair(v); + } + // One hit on P: still probation at threshold 2. + policy.record_hit(2, cost(400.0, 100.0)); + assert_eq!(policy.entry(2).unwrap().state, PolicyEntryState::Probation); + assert!( + policy.probation_bytes() <= 1 << 20, + "half-share state over cap: {}", + policy.probation_bytes() + ); + + // Remove the admitted co-reference: P's charge rises to all of S. + let outcome = policy.remove(1, &[]).expect("A removed"); + assert!( + !outcome.probation_cap_repair.victims().is_empty(), + "removal must carry cap victims for the risen share" + ); + for v in outcome.probation_cap_repair.victims() { + policy.remove_without_cap_repair(*v); + } + assert!( + policy.probation_bytes() <= 1 << 20, + "class {} still over cap after repair", + policy.probation_bytes() + ); + assert!(policy.entry(2).is_none(), "P must be the repair victim"); +} + +#[test] +fn cap_repair_prefers_unpinned_over_older_pinned_probationer() { + // Older pinned probationer (key 1) must never be selected while the + // younger unpinned probationer (key 2) can repair the cap (#1650). + let mut policy = BenefitPolicy::new(PolicyConfig { + probation_byte_budget: 1 << 20, + ..PolicyConfig::default() + }); + for key in 1..=2u64 { + let decision = policy.consider_admission(key, 1 << 20, vec![], cost(400.0, 100.0), &[1]); + // Do not commit yet: build the over-cap state first. + assert!( + !decision.probation_cap_repair.victims().contains(&1), + "admission never selects a pin" + ); + } + let repair = policy.select_probation_cap_victims(&[1]); + assert!( + !repair.victims().contains(&1), + "pinned key selected as victim: {:?}", + repair.victims() + ); + assert!(repair.victims().contains(&2)); + assert!(!repair.is_deferred()); + for v in repair.victims().iter().copied() { + policy.remove(v, &[1]); + } + assert!(policy.probation_bytes() <= 1 << 20); + assert!(policy.entries.contains_key(&1), "pin must survive repair"); +} + +#[test] +fn all_pinned_cap_is_deferred_with_shortfall_never_a_pin() { + let mut policy = BenefitPolicy::new(PolicyConfig { + probation_byte_budget: 1 << 20, + ..PolicyConfig::default() + }); + let pinned = [1u64, 2]; + for (n, key) in pinned.iter().enumerate() { + let decision = + policy.consider_admission(*key, 1 << 20, vec![], cost(400.0, 100.0), &pinned); + let repair = &decision.probation_cap_repair; + assert!( + repair.victims().is_empty(), + "all candidates pinned: no victim may be selected" + ); + // First admission is under cap; the second pushes the all-pinned + // class over cap with no selectable candidate. + if n == 1 { + assert!( + repair.is_deferred(), + "unsatisfiable cap must be Deferred, got {:?}", + repair + ); + assert_eq!(repair.shortfall_bytes(), 1 << 20); + } + assert!( + !repair.victims().contains(&1) && !repair.victims().contains(&2), + "pins never selected even when cap is unsatisfiable" + ); + } + // The removal path honors pins identically: a caller-forced removal of + // the pinned key 1 leaves key 2 at exactly the cap — satisfied, no pin + // selected. + let outcome = policy.remove(1, &pinned).expect("entry 1 removed"); + assert!(!outcome.probation_cap_repair.is_deferred()); + assert!(outcome.probation_cap_repair.victims().is_empty()); + assert!(policy.entries.contains_key(&2)); + // Once the pin releases, repair becomes satisfiable again (key 2 is + // selectable again). + let repair = policy.select_probation_cap_victims(&[]); + assert!(!repair.is_deferred()); +} + +/// Grace must expire under repeated misses: a miss is a real observation +/// that advances the clock but is a negative value signal (no recency +/// refresh), so a miss-only stream ages an entry out of grace and makes +/// it evictable instead of holding it indefinitely. +#[test] +fn grace_expires_under_repeated_misses() { + let grace = 8u64; + let mut policy = BenefitPolicy::new(PolicyConfig { + grace_observations: grace, + ..PolicyConfig::default() + }); + policy.consider_admission(1, 1 << 20, vec![], cost(400.0, 100.0), &[]); + policy.consider_admission(2, 1 << 20, vec![], cost(400.0, 100.0), &[]); + // Miss key 1 seven times (each miss advances the clock): the clock at + // admission was 2, so key 1's age becomes 8 (out of grace) while key + // 2's age becomes 7 (still in grace) — exactly one observation on + // either side of the grace boundary, not an ordering artifact. + let clock_at_admission = policy.clock_debug(); + for _ in 0..grace - 1 { + policy.record_miss(1); + } + assert!( + policy.clock_debug() == clock_at_admission + grace - 1, + "misses must advance the clock" + ); + let e1 = policy.entry(1).unwrap(); + let e2 = policy.entry(2).unwrap(); + assert_eq!( + policy.clock_debug().saturating_sub(e1.last_observation), + grace, + "key 1 exactly out of grace" + ); + assert_eq!( + policy.clock_debug().saturating_sub(e2.last_observation), + grace - 1, + "key 2 exactly one observation inside grace" + ); + // Out-of-grace entry 1 is now evictable in pass 1 (grace honored); + // requesting only its bytes must never touch in-grace entry 2. + let victims = policy.choose_victims(1 << 20, &[]); + let evictable: Vec = victims.iter().map(|(k, _)| *k).collect(); + assert!( + evictable.contains(&1), + "miss-only entry must age out of grace, victims {:?}", + evictable + ); + assert!( + !evictable.contains(&2), + "in-grace entry must not be evicted first, victims {:?}", + evictable + ); + // A miss-only stream cannot hold grace forever — already asserted above: + // entry 1's age is exactly `grace` (evictable) after only misses. +} diff --git a/crates/skippy-cache/src/policy/traces.rs b/crates/skippy-cache/src/policy/traces.rs new file mode 100644 index 0000000000..e4d8ca0c0b --- /dev/null +++ b/crates/skippy-cache/src/policy/traces.rs @@ -0,0 +1,137 @@ +//! Deterministic synthetic traces for policy comparison (#1650 first slice). +//! Seeded xorshift so every run replays identically. + +pub struct Rng(u64); + +impl Rng { + pub fn new(seed: u64) -> Self { + Self(seed.max(1)) + } + pub fn next_u64(&mut self) -> u64 { + let mut x = self.0; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + self.0 = x; + x + } + pub fn next_f64(&mut self) -> f64 { + (self.next_u64() >> 11) as f64 / (1u64 << 53) as f64 + } +} + +/// One cache access in a trace: which entry, and how valuable its reuse is. +#[derive(Debug, Clone, Copy)] +pub struct TraceAccess { + pub entry: u64, + pub cold_prefill_cost: f64, + pub restore_cost: f64, + pub exclusive_bytes: u64, +} + +pub const HOT_ZIPF_ENTRIES: u64 = 64; + +/// Zipf-like hotset: entry i is accessed with probability proportional to +/// 1/(i+1), so a small hotset dominates while a long tail streams once. +pub fn zipf_hotset_trace(seed: u64, len: usize) -> Vec { + let mut rng = Rng::new(seed); + let weights: Vec = (0..HOT_ZIPF_ENTRIES) + .map(|i| 1.0 / (i as f64 + 1.0)) + .collect(); + let total: f64 = weights.iter().sum(); + let mut out = Vec::with_capacity(len); + for _ in 0..len { + let pick = rng.next_f64() * total; + let mut entry = 0; + let mut acc = 0.0; + for (i, w) in weights.iter().enumerate() { + acc += w; + if pick <= acc { + entry = i as u64; + break; + } + } + out.push(access_for(entry, &mut rng)); + } + out +} + +/// Turn growth: each "session" revisits its whole history, extending it by a +/// fresh entry — the realistic agentic chat shape. +pub fn turn_growth_trace(seed: u64, sessions: u64, turns: u64) -> Vec { + let mut rng = Rng::new(seed); + let mut out = Vec::new(); + for session in 0..sessions { + for turn in 0..turns { + for entry in 0..=turn { + out.push(access_for(session * turns + entry, &mut rng)); + } + } + } + out +} + +/// One-shot stream: every entry seen exactly once, occasionally interleaved +/// with a hot entry so the policy must not wreck the hotset. +pub fn one_shot_trace(seed: u64, len: usize) -> Vec { + let mut rng = Rng::new(seed); + let mut out = Vec::with_capacity(len); + for i in 0..len as u64 { + if i % 8 == 7 { + out.push(access_for(0, &mut rng)); // hot anchor + } + out.push(access_for(1_000_000 + i, &mut rng)); + } + out +} + +/// Mixed sizes: footprints span three orders of magnitude. +pub fn mixed_size_trace(seed: u64, len: usize) -> Vec { + let mut rng = Rng::new(seed); + let mut out = Vec::with_capacity(len); + for i in 0..len as u64 { + let class = (i / 32) % 3; + // Distinct key per size class: reusing one key would make the + // larger sizes hits on a 64 KiB resident entry, so the trace + // would never exercise mixed resident sizes or large-entry + // eviction. + let mut access = access_for(class * 32 + i % 32, &mut rng); + access.exclusive_bytes = match class { + 0 => 64 << 10, + 1 => 4 << 20, + _ => 256 << 20, + }; + out.push(access); + } + out +} + +fn access_for(entry: u64, rng: &mut Rng) -> TraceAccess { + let cold = 50.0 + rng.next_f64() * 400.0; + TraceAccess { + entry, + cold_prefill_cost: cold, + restore_cost: cold * 0.3, + exclusive_bytes: (1 + rng.next_u64() % 8) << 20, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn turn_growth_keys_do_not_collide_across_long_sessions() { + let turns = 1_001; + let trace = turn_growth_trace(7, 2, turns); + let first_session_len = (turns * (turns + 1) / 2) as usize; + let first_second_session_key = trace[first_session_len].entry; + + assert_eq!(first_second_session_key, turns); + assert!( + trace[..first_session_len] + .iter() + .all(|access| access.entry < first_second_session_key) + ); + } +}