diff --git a/Cargo.lock b/Cargo.lock index 0d2908fb6a..35276c0521 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7391,6 +7391,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "skippy-cache", "skippy-protocol", "skippy-runtime", "skippy-topology", diff --git a/crates/skippy-bench/Cargo.toml b/crates/skippy-bench/Cargo.toml index 8604d172aa..d845ecf5f5 100644 --- a/crates/skippy-bench/Cargo.toml +++ b/crates/skippy-bench/Cargo.toml @@ -10,6 +10,7 @@ clap.workspace = true csv = "1.4.0" dirs = "6.0.0" libc = "0.2" +skippy-cache = { path = "../skippy-cache" } skippy-protocol = { path = "../skippy-protocol" } skippy-runtime = { path = "../skippy-runtime" } skippy-topology = { path = "../skippy-topology" } diff --git a/crates/skippy-bench/src/cli.rs b/crates/skippy-bench/src/cli.rs index 8b68c33065..9ebca354ba 100644 --- a/crates/skippy-bench/src/cli.rs +++ b/crates/skippy-bench/src/cli.rs @@ -35,6 +35,8 @@ pub enum CommandKind { LocalSplitChainBinary(LocalSplitChainBinaryArgs), #[command(name = "verify-window-local")] VerifyWindowLocal(VerifyWindowLocalArgs), + #[command(name = "l2-tier")] + L2Tier(L2TierArgs), #[command(name = "chat-corpus")] ChatCorpus(ChatCorpusArgs), #[command(name = "token-lengths")] @@ -45,6 +47,35 @@ pub enum CommandKind { Run(RunArgs), } +#[derive(Parser)] +pub struct L2TierArgs { + /// Working directory for the L3 store. Must not exist: the bench + /// creates, sentinel-marks, and (unless `--keep-store`) removes a + /// directory it owns, and refuses any pre-existing path instead of + /// deleting it. + #[arg(long, default_value = "/tmp/skippy-l2-tier-bench")] + pub store_root: PathBuf, + /// Number of timed L3-cold / L2-warm matched pairs after warmup. + #[arg(long, default_value_t = 50)] + pub pairs: usize, + /// Recorded prefix length in tokens (the synthetic conversation length). + #[arg(long, default_value_t = 1_893)] + pub tokens: usize, + /// Bytes of KV payload per token — sized to mimic a real dense model's + /// per-token KV footprint at the target dtype. + #[arg(long, default_value_t = 512)] + pub kv_bytes_per_token: usize, + /// L2 budget in MiB. Defaults to four times one entry. + #[arg(long)] + pub l2_budget_mib: Option, + /// Keep the L3 store directory after the run for inspection. + #[arg(long, default_value_t = false)] + pub keep_store: bool, + /// Model identity stamped into the L3 tier and L2 keys. + #[arg(long, default_value = "bench-model")] + pub model_identity: String, +} + #[derive(Parser)] pub struct EvalArgs { #[command(subcommand)] diff --git a/crates/skippy-bench/src/l2_tier.rs b/crates/skippy-bench/src/l2_tier.rs new file mode 100644 index 0000000000..4a7b6bac04 --- /dev/null +++ b/crates/skippy-bench/src/l2_tier.rs @@ -0,0 +1,309 @@ +//! `l2-tier` benchmark: cold L3 fill versus warm L2 lookup on identical +//! packed entries (#1651). +//! +//! Builds a temporary L3 store, spills a synthetic multi-turn prompt at a +//! recorded prefix length, then measures two restore paths in-process: +//! +//! - **L3 cold fill**: `L3Tier::fill_longest` — index probe + segment +//! assembly + digest verification from disk. +//! - **L2 warm lookup**: `L2Tier::get` + `to_payload` + materialization — +//! the entry was admitted from an identical verified L3 fill, so the +//! lookup is a digest-keyed handle assembly (no re-hash: admission +//! verified the wire once; segments are immutable afterward). +//! +//! Both arms are timed through the same boundary: the moment their bytes +//! are usable (`full_state_bytes_timed`). A multi-segment L2 entry still +//! materializes its wire on read, so stopping the L2 timer at handle +//! creation would understate the real cost; the equality gate compares the +//! materialized bytes of both arms before the pair is counted. The +//! handle-only lookup time is reported separately as +//! `l2_handle_lookup_ns`. +//! +//! Admission hashing (the one-time wire BLAKE3) is measured separately +//! and reported as its own metric, never inside the timed lookup. +//! +//! The store root is owned by the run: it must not exist beforehand (the +//! bench refuses existing paths instead of deleting user data) and it is +//! marked with an ownership sentinel so cleanup never touches a directory +//! the bench did not create. +use std::path::{Path, PathBuf}; +use std::time::Instant; + +use anyhow::{Context, Result}; + +use crate::cli::L2TierArgs; +use skippy_cache::{ + ExactStatePayload, ExactStatePayloadMirror, L2Origin, L2Tier, l2_cache_key, l3_prefix_key, +}; + +/// Marker file proving the bench created the store root itself; cleanup +/// refuses to `remove_dir_all` a directory without it. +const OWNERSHIP_SENTINEL: &str = ".skippy-l2-tier-bench-owned"; + +/// Create a fresh, bench-owned store root. Existing paths are refused — +/// the bench must never delete a user-supplied directory it did not +/// create. +fn prepare_store_root(requested: &Path) -> Result { + if requested.symlink_metadata().is_ok() { + anyhow::bail!( + "refusing to use store root {}: the path already exists; the bench only runs \ + in a root it created itself", + requested.display() + ); + } + std::fs::create_dir_all(requested) + .with_context(|| format!("failed to create bench store root {}", requested.display()))?; + std::fs::write( + requested.join(OWNERSHIP_SENTINEL), + b"skippy-bench l2-tier store\n", + ) + .with_context(|| format!("failed to mark {} as bench-owned", requested.display()))?; + Ok(requested.to_path_buf()) +} + +/// Remove a bench-owned store root. Refuses paths without the ownership +/// sentinel so `remove_dir_all` can never hit arbitrary input. +fn remove_owned_store_root(root: &Path) -> Result<()> { + if !root.join(OWNERSHIP_SENTINEL).is_file() { + anyhow::bail!( + "refusing to remove store root {}: missing bench ownership sentinel", + root.display() + ); + } + std::fs::remove_dir_all(root) + .with_context(|| format!("failed to remove bench store root {}", root.display())) +} + +fn percentile(samples_ns: &mut [u128], pct: f64) -> f64 { + samples_ns.sort_unstable(); + let index = ((pct / 100.0) * (samples_ns.len() as f64 - 1.0)).round() as usize; + samples_ns[index.min(samples_ns.len() - 1)] as f64 +} + +pub fn l2_tier(args: L2TierArgs) -> Result<()> { + // Validation: reject degenerate configurations up front instead of + // dividing by zero or allocating nothing below. + if args.pairs == 0 { + anyhow::bail!("--pairs must be at least 1"); + } + if args.tokens == 0 { + anyhow::bail!("--tokens must be at least 1"); + } + if args.kv_bytes_per_token == 0 { + anyhow::bail!("--kv-bytes-per-token must be at least 1"); + } + + // Owned store root: refuse existing paths rather than deleting them, + // and mark the created directory so cleanup stays bounded to it. + let store_root = prepare_store_root(&args.store_root)?; + + let namespace = "bench-namespace"; + let state_identity = args.model_identity.clone(); + let token_ids: Vec = (0..args.tokens).map(|i| (i % 128_000) as i32).collect(); + + // Deterministic synthetic KV payload: content matters only for digests, + // size matters for timing. + let payload_len = args + .tokens + .checked_mul(args.kv_bytes_per_token) + .context("--tokens * --kv-bytes-per-token overflows")?; + let payload_bytes: Vec = (0..payload_len).map(|i| (i % 251) as u8).collect(); + let payload = ExactStatePayload::full_state(payload_bytes); + + let tier = skippy_cache::L3Tier::open( + store_root.clone(), + (payload_len as u64) * 8, + state_identity.clone(), + 64 * 1024, + ) + .context("failed to open bench L3 tier")?; + + // Spill once: this is the population path, not the measured path. + let manifest_key = tier + .spill(namespace, &token_ids, &payload, None, None) + .context("bench spill failed")?; + let _ = manifest_key; + + // Locate once to learn the recorded prefix key/digest used by both paths. + let location = tier + .locate_longest(namespace, &token_ids, 8) + .context("bench locate failed")? + .context("bench spill was not locatable")?; + let manifest = tier.store().load_manifest(&location.manifest_key)?; + let payload_digest = manifest.payload_digest.clone(); + let recorded_tokens = manifest.token_count; + + let l2_budget_bytes = args + .l2_budget_mib + .map(|mib| { + mib.checked_mul(1024 * 1024) + .context("--l2-budget-mib overflows") + }) + .transpose()? + .unwrap_or(payload_len as u64 * 4); + let l2 = L2Tier::new(l2_budget_bytes); + + let cache_key = l2_cache_key(&args.model_identity, &state_identity, namespace, &token_ids); + + // Warmup: one L3 fill, then admit it into L2 from the verified wire. + // The wire check inside `admit` is the one-time admission hash; it is + // timed separately below. + let warm_fill = tier + .fill_longest(namespace, &token_ids, 8) + .context("bench warmup L3 fill failed")? + .context("bench warmup L3 fill missed")?; + let (warm_wire, _) = warm_fill.payload.full_state_bytes_timed().context("wire")?; + let admission_started = Instant::now(); + l2.admit( + cache_key.clone(), + warm_fill.token_count, + payload_digest.clone(), + &warm_wire, + ExactStatePayloadMirror::from_manifest(&manifest) + .map_err(|refusal| anyhow::anyhow!(refusal.reason()))?, + L2Origin::FromL3, + ) + .map_err(|refusal| anyhow::anyhow!("bench warmup L2 admit refused: {}", refusal.reason()))?; + let admission_hash_ns = admission_started.elapsed().as_nanos(); + + let mut l3_samples: Vec = Vec::with_capacity(args.pairs); + let mut l2_samples: Vec = Vec::with_capacity(args.pairs); + let mut l2_handle_samples: Vec = Vec::with_capacity(args.pairs); + + for pair in 0..args.pairs { + // Cold-ish L3 fill: the OS page cache will help after warmup, which + // matches the production comparison — both paths run on the same + // machine state, the delta is the tier delta. + let start = Instant::now(); + let fill = tier + .fill_longest(namespace, &token_ids, 8) + .context("bench L3 fill failed")? + .context("bench L3 fill missed")?; + let l3_ns = start.elapsed().as_nanos(); + + // Timed through the same boundary as the L3 arm, starting before + // the lookup: the L3 timer covers index probe + assembly + + // verification, so the L2 timer covers lookup + handle assembly + + // materialization — both arms measure "nothing to usable bytes". + // The handle-only lookup time (this same `get`, inner timer) is + // reported separately as `l2_handle_lookup_ns`. + let start = Instant::now(); + let handle_start = Instant::now(); + let hit = l2.get(&cache_key); + let l2_handle_ns = handle_start.elapsed().as_nanos(); + let l2_payload = hit.as_ref().map(|hit| hit.to_payload()); + let (l2_bytes, _) = l2_payload + .as_ref() + .context("bench L2 payload missing")? + .full_state_bytes_timed() + .context("bench L2 bytes")?; + let l2_ns = start.elapsed().as_nanos(); + + // Correctness gate, outside the timer: L2 must return + // byte-identical state to the L3 fill, or the speedup is + // meaningless. + let hit = hit.context("bench L2 lookup missed")?; + anyhow::ensure!(hit.token_count == fill.token_count); + anyhow::ensure!(hit.payload_digest == payload_digest); + let (l3_bytes, _) = fill.payload.full_state_bytes_timed().context("l3 bytes")?; + anyhow::ensure!( + l3_bytes.as_ref() == l2_bytes.as_ref(), + "pair {pair}: L2 payload diverged from L3 fill" + ); + + l3_samples.push(l3_ns); + l2_samples.push(l2_ns); + l2_handle_samples.push(l2_handle_ns); + } + + let stats = l2.stats(); + let mut l3_sorted = l3_samples.clone(); + let mut l2_sorted = l2_samples.clone(); + let mut l2_handle_sorted = l2_handle_samples.clone(); + let summary = serde_json::json!({ + "bench": "l2-tier", + "pairs": args.pairs, + "tokens": recorded_tokens, + "payload_bytes": payload_len, + "l2_budget_bytes": l2_budget_bytes, + "model_identity": args.model_identity, + "l3_fill_ns": { + "p50": percentile(&mut l3_sorted, 50.0), + "p99": percentile(&mut l3_sorted, 99.0), + }, + "l2_lookup_to_usable_bytes_ns": { + "p50": percentile(&mut l2_sorted, 50.0), + "p99": percentile(&mut l2_sorted, 99.0), + }, + "l2_handle_lookup_ns": { + "p50": percentile(&mut l2_handle_sorted, 50.0), + "p99": percentile(&mut l2_handle_sorted, 99.0), + }, + "l2_admission_hash_ns_one_time": admission_hash_ns, + "speedup_p50": percentile(&mut l3_sorted, 50.0) / percentile(&mut l2_sorted, 50.0).max(1.0), + "l2_stats": { + "hits": stats.hits, + "misses": stats.misses, + "evictions": stats.evictions, + "bytes": stats.bytes, + "segments": stats.segments, + "shared_bytes_admitted": stats.shared_bytes_admitted, + }, + "l3_prefix_key": l3_prefix_key(namespace, &token_ids), + }); + println!("{summary}"); + + if !args.keep_store { + remove_owned_store_root(&store_root)?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn store_root_refuses_existing_paths() { + let dir = + std::env::temp_dir().join(format!("skippy-l2-bench-refuse-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("create pre-existing dir"); + std::fs::write(dir.join("precious.txt"), b"user data").expect("seed user data"); + + let err = prepare_store_root(&dir).expect_err("existing path must be refused"); + assert!( + err.to_string().contains("refusing to use store root"), + "unexpected error: {err}" + ); + assert!( + dir.join("precious.txt").is_file(), + "pre-existing contents must survive the refusal" + ); + + // A root the bench created is removable; a lookalike without the + // sentinel is not. (Creation itself stays refused for any + // pre-existing path, owned or not.) + let _ = std::fs::remove_dir_all(&dir); + + let owned = + std::env::temp_dir().join(format!("skippy-l2-bench-owned-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&owned); + prepare_store_root(&owned).expect("fresh root is created and owned"); + assert!(owned.join(OWNERSHIP_SENTINEL).is_file()); + remove_owned_store_root(&owned).expect("owned root is removable"); + assert!(!owned.exists()); + + // Unmarked directory: cleanup must refuse. + let unowned = + std::env::temp_dir().join(format!("skippy-l2-bench-unowned-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&unowned); + std::fs::create_dir_all(&unowned).expect("create unowned dir"); + std::fs::write(unowned.join("keep.txt"), b"user data").expect("seed"); + let err = + remove_owned_store_root(&unowned).expect_err("unowned root removal must be refused"); + assert!(err.to_string().contains("sentinel"), "unexpected: {err}"); + assert!(unowned.join("keep.txt").is_file(), "contents survive"); + let _ = std::fs::remove_dir_all(&unowned); + } +} diff --git a/crates/skippy-bench/src/main.rs b/crates/skippy-bench/src/main.rs index 1f0de777b0..39743c8bd7 100644 --- a/crates/skippy-bench/src/main.rs +++ b/crates/skippy-bench/src/main.rs @@ -3,6 +3,7 @@ mod cli; mod direct_return_listener; mod distributed; mod evals; +mod l2_tier; mod local_single; mod local_split; mod model_identity; @@ -54,6 +55,7 @@ fn main() -> Result<()> { CommandKind::LocalSplitCompare(args) => local_split_compare(args), CommandKind::LocalSplitChainBinary(args) => local_split_chain_binary(args), CommandKind::VerifyWindowLocal(args) => verify_window_local(args), + CommandKind::L2Tier(args) => l2_tier::l2_tier(args), CommandKind::ChatCorpus(args) => chat_corpus(args), CommandKind::TokenLengths(args) => token_lengths(args), CommandKind::FocusedRuntime(args) => focused_runtime(args), diff --git a/crates/skippy-cache/src/l2/mod.rs b/crates/skippy-cache/src/l2/mod.rs new file mode 100644 index 0000000000..6ce0658fc2 --- /dev/null +++ b/crates/skippy-cache/src/l2/mod.rs @@ -0,0 +1,2438 @@ +//! Host-RAM L2 tier over the packed L3 segment format (#1651). +//! +//! The radix cache (L1) holds resident payloads; the L3 tier holds the same +//! state durably on disk as content-addressed packed segments. This module +//! adds the missing middle tier: a bounded host-RAM cache of *immutable +//! packed segments*, keyed and identified exactly like the L3 entries they +//! mirror. +//! +//! Contract (mirrors `crate::tier::L3Tier`): +//! +//! - **Identity**: entries are stamped with the tier's model and exact-state +//! identities. The cache key includes the identities, so a re-identity is +//! a wholesale miss, never a silent hit. +//! - **Coordinates**: entries are keyed by the same +//! `(namespace, token path)` coordinates L3 uses — +//! [`crate::tier::l3_prefix_key`] / [`crate::tier::l3_namespace_key`] — so +//! an L2 hit is interchangeable with the L3 entry it cached. +//! - **Segment sharing**: an entry stores immutable `Arc>` segment +//! handles keyed by their content digests plus a layout that maps the +//! entry's L3 manifest segment list onto those handles. A longer prefix +//! that extends a shorter one references the same segment handles, so +//! turn growth shares prefix bytes instead of duplicating them — L2 RAM +//! tracks *distinct segment* bytes, not per-prefix assembled bytes. +//! - **Integrity**: the whole concatenated L3 wire digest (the manifest key) +//! is verified exactly once, at admission, against the payload being +//! admitted. After admission the segment bytes are immutable, so every +//! later read is a digest lookup plus handle assembly — no re-hash. An +//! admission-time mismatch refuses the insert; L2 never holds bytes it +//! did not verify. +//! - **Bounded**: the byte budget is charged with each entry's *distinct* +//! segment bytes (bytes not already held by an in-flight insert) and +//! enforced by evicting in deterministic LRU order. A payload whose +//! distinct bytes exceed the whole budget is refused. A segment shared +//! with an already-admitted entry is shared for accounting too: only the +//! first admission pays for it. +//! - **Zero-copy reads**: `get` clones handles, not bytes; the returned +//! `CacheBytes` is a block-backed view over the shared segment storages, +//! contiguous in the single-segment case. +//! +//! This first slice is a standalone store with no wiring into the request +//! path; the benchmark harness drives it directly. L2 promotion/demotion +//! policy and server integration land in a later slice. +use std::{ + collections::HashMap, + ops::Range, + sync::{ + Arc, Mutex, + atomic::{AtomicU64, Ordering}, + }, +}; + +use crate::payload::{CacheBytes, ExactStatePayloadKind}; +use crate::{HandoffManifest, segment_digest}; +#[cfg(test)] +use crate::{HandoffSegmentRef, MANIFEST_VERSION}; + +/// Where an entry came from, for telemetry and promotion policy later. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum L2Origin { + /// Admitted from an assembled L3 fill (verified wire). + FromL3, + /// Admitted from another verified wire source (tests, prefetch). + Direct, +} + +/// LRU eviction accounting for one removed entry. +/// +/// `freed_bytes` is what removal actually released: segments whose last +/// referencing entry left the tier. `retained_bytes` is shared-segment +/// bytes that stay because another entry still references them. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct L2Eviction { + pub cache_key: String, + pub freed_bytes: u64, + pub retained_bytes: u64, +} + +impl L2Eviction { + /// Bytes charged to the budget for this entry (what its removal freed). + pub fn payload_bytes(&self) -> u64 { + self.freed_bytes + } +} + +/// Read path counters. One snapshot per `stats()` call. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct L2Stats { + pub entries: u64, + /// Sum of entries' distinct segment charges — the live budget usage. + pub bytes: u64, + /// Sum of entry payload lengths including cross-entry sharing; larger + /// than `bytes` exactly when entries share prefix segments. + pub logical_bytes: u64, + /// Distinct immutable segment handles currently held. + pub segments: u64, + /// Bytes held in the segment pool (== `bytes` when the pool is live). + pub segment_bytes: u64, + /// Distinct segment bytes a single admission did not have to copy + /// because an earlier admission already held them. + pub shared_bytes_admitted: u64, + pub budget_bytes: u64, + pub hits: u64, + pub misses: u64, + pub inserts: u64, + pub evictions: u64, + /// Admissions refused because the payload digest did not match the + /// bytes (hash mismatch or malformed digest string). + pub admission_rejects: u64, + pub refused_bytes: u64, +} + +/// One assembled L2 entry: which stored segment handles make up the wire, +/// in manifest order, plus the payload split so a fill can be rebuilt. +#[derive(Debug, Clone)] +pub struct L2Layout { + pub payload_kind: ExactStatePayloadKind, + pub total_bytes: u64, + pub kv_bytes: u64, + pub recurrent_bytes: u64, + /// `(segment digest, byte range within the assembled wire)` per + /// manifest segment, in manifest order. Ranges concatenate to + /// `0..total_bytes` exactly as the L3 manifest tiles them. + pub segments: Vec<(String, Range)>, +} + +/// The L2 mirror of an assembled L3 entry: verified segment handles plus +/// the layout needed to rebuild a serving payload without disk I/O. +#[derive(Debug, Clone)] +pub enum ExactStatePayloadMirror { + FullState { layout: L2Layout }, + RecurrentOnly { layout: L2Layout }, + KvRecurrent { layout: L2Layout }, +} + +impl ExactStatePayloadMirror { + pub fn kind(&self) -> ExactStatePayloadKind { + match self { + Self::FullState { .. } => ExactStatePayloadKind::FullState, + Self::RecurrentOnly { .. } => ExactStatePayloadKind::RecurrentOnly, + Self::KvRecurrent { .. } => ExactStatePayloadKind::KvRecurrent, + } + } + + /// Total wire length of the entry (the assembled payload length). + pub fn byte_len(&self) -> u64 { + match self { + Self::FullState { layout } + | Self::RecurrentOnly { layout } + | Self::KvRecurrent { layout } => layout.total_bytes, + } + } + + /// Build a mirror from a captured L3 manifest. Callers must verify the + /// payload wire against `manifest.payload_digest` — `admit` does this — + /// before the mirror is stored. + pub fn from_manifest(manifest: &HandoffManifest) -> Result { + let kind = match manifest.payload_kind.as_str() { + "full-state" => ExactStatePayloadKind::FullState, + "recurrent-only" => ExactStatePayloadKind::RecurrentOnly, + "kv-recurrent" => ExactStatePayloadKind::KvRecurrent, + other => { + return Err(L2InsertRefusal::UnknownPayloadKind(other.to_string())); + } + }; + let mut offset = 0u64; + let segments = + manifest + .segments + .iter() + .enumerate() + .map(|(index, segment)| { + if segment.index != index as u32 { + return Err(L2InsertRefusal::MalformedManifest(format!( + "segment {} records index {} but sits at position {index}", + segment.digest, segment.index + ))); + } + if segment.offset != offset { + return Err(L2InsertRefusal::MalformedManifest(format!( + "segment {} records offset {} but tiles at {offset}", + segment.digest, segment.offset + ))); + } + let start = offset; + offset = offset.checked_add(segment.bytes).ok_or( + L2InsertRefusal::MalformedManifest("segment tiling overflows".to_string()), + )?; + Ok((segment.digest.clone(), start..offset)) + }) + .collect::, _>>()?; + if offset != manifest.total_bytes { + return Err(L2InsertRefusal::MalformedManifest(format!( + "segments tile {offset} bytes but the manifest records {}", + manifest.total_bytes + ))); + } + let layout = L2Layout { + payload_kind: kind, + total_bytes: manifest.total_bytes, + kv_bytes: manifest.kv_bytes, + recurrent_bytes: manifest.recurrent_bytes, + segments, + }; + Ok(match kind { + ExactStatePayloadKind::FullState => Self::FullState { layout }, + ExactStatePayloadKind::RecurrentOnly => Self::RecurrentOnly { layout }, + ExactStatePayloadKind::KvRecurrent => Self::KvRecurrent { layout }, + }) + } + + fn layout(&self) -> &L2Layout { + match self { + Self::FullState { layout } + | Self::RecurrentOnly { layout } + | Self::KvRecurrent { layout } => layout, + } + } + + /// Segment digests in wire order, deduplicated. + fn segment_digests(&self) -> Vec<&str> { + let mut seen = Vec::new(); + for (digest, _) in &self.layout().segments { + if !seen.contains(&digest.as_str()) { + seen.push(digest.as_str()); + } + } + seen + } +} + +/// A hit handed to the caller: the entry's layout plus `Arc` clones of the +/// segment handles the layout references, keyed by digest. The tier stores +/// this directly on `L2Hit` so payload assembly needs no tier lock. +#[derive(Debug, Clone)] +pub struct L2Hit { + pub payload: ExactStatePayloadMirror, + pub token_count: u64, + pub payload_digest: String, + /// Distinct segment handles referenced by the layout, keyed by digest. + pub(crate) segments: HashMap, +} + +impl L2Hit { + /// Rebuild a serving payload. Cheap in the common cases: the returned + /// `CacheBytes` is a block-backed view sharing the stored segment + /// storages (`Arc` clones, not byte copies); a single whole-storage + /// segment borrows it contiguously. Only a multi-segment read of + /// distinct storages materializes bytes, and only into the caller's + /// `Cow` on `as_cow`. + pub fn to_payload(&self) -> crate::payload::ExactStatePayload { + let layout = self.payload.layout(); + let wire = self.wire_view(0..layout.total_bytes); + match self.payload.kind() { + crate::payload::ExactStatePayloadKind::FullState => { + crate::payload::ExactStatePayload::FullState { bytes: wire } + } + crate::payload::ExactStatePayloadKind::RecurrentOnly => { + crate::payload::ExactStatePayload::RecurrentOnly { recurrent: wire } + } + crate::payload::ExactStatePayloadKind::KvRecurrent => { + // Split the wire at kv_bytes exactly like L3 load does: kv + // is the leading block-backed view, recurrent the tail. Both + // share the same storages; no bytes are copied here. + let kv_len = layout.kv_bytes.min(layout.total_bytes); + let kv = self.wire_view(0..kv_len); + let recurrent = self.wire_view(kv_len..layout.total_bytes); + crate::payload::ExactStatePayload::KvRecurrent { kv, recurrent } + } + } + } + + /// Block-backed `CacheBytes` over `range` of the assembled wire, in + /// wire order. Blocks outside `range` are skipped; edge blocks are + /// narrowed to the overlap. Byte-identical views share the same + /// segment storages; nothing is copied. + pub(crate) fn wire_view(&self, range: Range) -> CacheBytes { + let layout = self.payload.layout(); + let start = range.start.min(layout.total_bytes); + let end = range.end.min(layout.total_bytes); + let blocks = layout + .segments + .iter() + .filter_map(|(digest, segment_range)| { + let block_start = segment_range.start.max(start); + let block_end = segment_range.end.min(end); + if block_start >= block_end { + return None; + } + let storage = self + .segments + .get(digest) + .map(|handle| Arc::clone(&handle.bytes)) + .unwrap_or_else(|| Arc::new(Vec::new())); + let len = storage.len() as u64; + let from = (block_start - segment_range.start).min(len); + let to = (block_end - segment_range.start).min(len); + Some((digest.clone(), storage, (from as usize)..(to as usize))) + }) + .collect::>(); + CacheBytes::from_shared_blocks(end.saturating_sub(start), blocks) + } +} + +/// Presence probe result. Probing never changes LRU recency. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct L2Peek { + pub token_count: u64, + pub payload_digest: String, + /// Total wire length including segments shared with other entries. + pub payload_bytes: u64, + /// Distinct segment bytes charged to the budget for this entry. + pub distinct_bytes: u64, + pub origin: L2Origin, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum L2InsertRefusal { + EmptyPayload, + OverBudget { + payload_bytes: u64, + }, + /// Admission would push the pool past the budget with the incoming + /// entry's handles pinned: even evicting every other entry could not + /// free the excess, so the transaction was rolled back untouched. + ProtectedOvercommit { + pool_bytes: u64, + budget_bytes: u64, + }, + MalformedDigest, + /// Admission hashing found the wire's BLAKE3 different from the digest + /// the payload claims (the L3 manifest key). + DigestMismatch { + expected: String, + actual: String, + }, + /// A layout segment's claimed digest does not match the BLAKE3 of its + /// exact wire range: the pool identity would not describe the bytes it + /// is supposed to serve. + SegmentDigestMismatch { + digest: String, + expected: String, + actual: String, + }, + /// A layout claims a segment digest for two different wire ranges, or + /// claims a digest the pool already holds with different content that + /// another live entry still references. Content-addressed identity must + /// stay unambiguous. + ConflictingSegment { + digest: String, + detail: String, + }, + UnknownPayloadKind(String), + MalformedManifest(String), +} + +impl L2InsertRefusal { + pub fn reason(&self) -> String { + match self { + Self::EmptyPayload => "refusing to cache an empty exact-state payload".to_string(), + Self::OverBudget { payload_bytes } => format!( + "payload of {payload_bytes} distinct bytes exceeds the entire L2 budget; \ + caching it would evict everything else" + ), + Self::ProtectedOvercommit { + pool_bytes, + budget_bytes, + } => format!( + "admission would leave the pool at {pool_bytes} bytes against a \ + {budget_bytes}-byte budget even after evicting every unpinned entry: \ + the admission's own (shared or pinned) segments are not evictable, \ + so it was rolled back" + ), + Self::MalformedDigest => { + "payload digest is not a 64-hex-character blake3 string".to_string() + } + Self::DigestMismatch { expected, actual } => format!( + "admission digest check failed: wire hashes to {actual} but the payload \ + claims {expected}" + ), + Self::SegmentDigestMismatch { + digest, + expected, + actual, + } => format!( + "segment {digest} does not describe its wire range: range hashes to {actual} \ + but the layout claims {expected}" + ), + Self::ConflictingSegment { digest, detail } => { + format!("conflicting claims for segment {digest}: {detail}") + } + Self::UnknownPayloadKind(kind) => { + format!("manifest holds unknown payload kind {kind}") + } + Self::MalformedManifest(detail) => { + format!("malformed L3 manifest: {detail}") + } + } + } +} + +/// An immutable segment: content-addressed bytes shared by `Arc`. +#[derive(Debug, Clone)] +pub(crate) struct SegmentHandle { + pub bytes: Arc>, +} + +#[derive(Debug)] +struct L2Entry { + payload: ExactStatePayloadMirror, + token_count: u64, + payload_digest: String, + origin: L2Origin, + /// LRU clock, bumped on successful hits only (probes are side-effect + /// free). + last_used: u64, + /// Distinct segment bytes charged against the budget. Shared segments + /// already held by other entries are not charged here. + charge_bytes: u64, + /// Total wire length including shared segments (telemetry). + payload_bytes: u64, +} + +#[derive(Default)] +struct L2Inner { + map: HashMap, + /// Content-addressed pool of immutable segments. + segments: HashMap, + /// Distinct segment bytes in the pool — the real RAM footprint. + bytes: u64, + clock: u64, +} + +/// Undo log for one admission transaction. Every pool/map mutation an +/// admission performs — reserving new handles, overwriting a digest with +/// new content, releasing orphaned handles, removing the replaced entry, +/// and the entries eviction removes — is recorded here so a refused +/// admission (protected overcommit) can restore the tier exactly. Charges +/// are not journaled: every exit path recomputes them from the live map. +#[derive(Default)] +struct AdmitJournal { + reserved: Vec, + /// Previous `Arc` handles overwritten by this admission's same-digest + /// new-content installs, restored on rollback. + overwrites: Vec<(String, SegmentHandle)>, + /// Pool bytes that left with the overwritten handles. + overwritten_bytes: u64, + /// Released-orphan handles: `(digest, handle)` pairs to reinstall on + /// rollback. + released_orphans: Vec<(String, SegmentHandle)>, + removed_entries: Vec<(String, L2Entry)>, + reserved_bytes: u64, + released_bytes: u64, +} + +impl AdmitJournal { + fn rollback(self, inner: &mut L2Inner) { + for (digest, handle) in self.overwrites { + inner.segments.insert(digest, handle); + } + for digest in &self.reserved { + inner.segments.remove(digest); + } + for (digest, handle) in self.released_orphans { + inner.segments.insert(digest, handle); + } + for (key, entry) in self.removed_entries { + inner.map.insert(key, entry); + } + inner.bytes = inner + .bytes + .saturating_add(self.released_bytes) + .saturating_add(self.overwritten_bytes) + .saturating_sub(self.reserved_bytes); + } +} + +/// Counters kept outside the map lock so `stats()` never blocks hits. +#[derive(Default)] +struct L2AtomicStats { + hits: AtomicU64, + misses: AtomicU64, + inserts: AtomicU64, + evictions: AtomicU64, + admission_rejects: AtomicU64, + refused_bytes: AtomicU64, + shared_bytes_admitted: AtomicU64, +} + +/// Bounded host-RAM L2 over immutable packed L3 segments. +pub struct L2Tier { + inner: Mutex, + budget_bytes: u64, + stats: L2AtomicStats, +} + +/// A digest string must be a BLAKE3 hex digest: `blake3:`-prefixed (as L3 +/// digests are) or bare 64 hex characters. +fn is_valid_digest(digest: &str) -> bool { + let hex = digest.strip_prefix("blake3:").unwrap_or(digest); + hex.len() == 64 && hex.bytes().all(|b| b.is_ascii_hexdigit()) +} + +/// One validated layout segment: its claimed digest and the exact verified +/// slice of the wire it describes. +type ValidatedSegment<'a> = (&'a str, &'a [u8]); + +/// Fully validate a mirror's layout against the verified wire *before* any +/// pool mutation: +/// +/// - the tiling is non-empty and contiguous over `0..total_bytes`; +/// - `kv_bytes + recurrent_bytes == total_bytes`; +/// - every segment digest is well-formed and hashes its exact wire range. +/// +/// A layout that fails any check is refused (`MalformedManifest` or +/// `SegmentDigestMismatch`): the pool is content-addressed, so a handle's +/// digest must describe the bytes it serves. +fn validate_layout<'a>( + mirror: &'a ExactStatePayloadMirror, + wire: &'a [u8], +) -> Result>, L2InsertRefusal> { + let layout = mirror.layout(); + if layout.segments.is_empty() { + return Err(L2InsertRefusal::MalformedManifest( + "layout holds no segments".to_string(), + )); + } + if layout.total_bytes != wire.len() as u64 { + return Err(L2InsertRefusal::MalformedManifest(format!( + "layout claims {total} total bytes but the wire holds {len}", + total = layout.total_bytes, + len = wire.len() + ))); + } + if layout.kv_bytes.saturating_add(layout.recurrent_bytes) != layout.total_bytes { + return Err(L2InsertRefusal::MalformedManifest(format!( + "kv ({kv}) + recurrent ({rec}) bytes do not tile the {total}-byte payload", + kv = layout.kv_bytes, + rec = layout.recurrent_bytes, + total = layout.total_bytes + ))); + } + let mut expected_start = 0u64; + let mut validated = Vec::with_capacity(layout.segments.len()); + for (digest, range) in &layout.segments { + if !is_valid_digest(digest) { + return Err(L2InsertRefusal::MalformedManifest(format!( + "segment digest {digest:?} is not a blake3 hex digest" + ))); + } + if range.start != expected_start || range.end < range.start || range.end > wire.len() as u64 + { + return Err(L2InsertRefusal::MalformedManifest(format!( + "segment range {range:?} does not contiguously tile the wire at offset \ + {expected_start}" + ))); + } + expected_start = range.end; + let slice = &wire[range.start as usize..range.end as usize]; + let actual = segment_digest(slice); + if actual != *digest { + return Err(L2InsertRefusal::SegmentDigestMismatch { + digest: digest.clone(), + expected: digest.clone(), + actual, + }); + } + validated.push((digest.as_str(), slice)); + } + if expected_start != layout.total_bytes { + return Err(L2InsertRefusal::MalformedManifest(format!( + "segments tile {expected_start} bytes but the layout claims {}", + layout.total_bytes + ))); + } + Ok(validated) +} +impl L2Tier { + pub fn new(budget_bytes: u64) -> Self { + Self { + inner: Mutex::new(L2Inner::default()), + budget_bytes, + stats: L2AtomicStats::default(), + } + } + + pub fn budget_bytes(&self) -> u64 { + self.budget_bytes + } + + /// Admit an assembled entry. + /// + /// `wire` is the payload's concatenated L3 wire — the exact bytes whose + /// BLAKE3 is the manifest key. Admission verifies + /// `segment_digest(&wire) == payload_digest` and refuses the insert on + /// mismatch: L2 never holds bytes it did not verify. Every layout + /// segment's digest is verified against its exact wire range before the + /// pool is touched, so a handle can never serve bytes its digest does + /// not describe. After admission the segment bytes are immutable, so + /// reads are a digest lookup plus handle assembly — no re-hash. + /// + /// The admission is atomic with respect to the segment pool: the + /// incoming layout's segments are installed (or reserved) *before* the + /// previous entry at the same key is released and before eviction runs, + /// with those handles pinned against removal, so a replacement or a + /// sharing admission can never release a handle it is about to + /// reference. Returns the evictions the admission caused. The budget is + /// charged with the entry's *distinct* segment bytes: segments already + /// held by another entry are shared, not duplicated, and only the first + /// admission pays for them. + pub fn admit( + &self, + cache_key: String, + token_count: u64, + payload_digest: String, + wire: &[u8], + mirror: ExactStatePayloadMirror, + origin: L2Origin, + ) -> Result, L2InsertRefusal> { + if !is_valid_digest(&payload_digest) { + self.stats.admission_rejects.fetch_add(1, Ordering::Relaxed); + return Err(L2InsertRefusal::MalformedDigest); + } + // The one integrity check on the whole wire: it must hash to the + // claimed manifest-key digest. + let actual = segment_digest(wire); + if actual != payload_digest { + self.stats.admission_rejects.fetch_add(1, Ordering::Relaxed); + return Err(L2InsertRefusal::DigestMismatch { + expected: payload_digest, + actual, + }); + } + let payload_bytes = mirror.byte_len(); + if payload_bytes == 0 { + // Mirrors L3: an empty payload cannot represent state. + return Err(L2InsertRefusal::EmptyPayload); + } + if payload_bytes != wire.len() as u64 { + return Err(L2InsertRefusal::MalformedManifest(format!( + "mirror claims {payload_bytes} payload bytes but the verified wire holds {}", + wire.len() + ))); + } + // Every segment slice is hashed against its claimed digest before + // any pool mutation: pool reuse trusts content, not digest text. + let segments = validate_layout(&mirror, wire)?; + + let mut inner = self.inner.lock().expect("L2 map lock poisoned"); + // Distinct-byte charge: segments the pool already holds with + // exactly the verified content are shared (anything else conflicts + // or is new). `shared` pins every handle this admission references + // — including segments still owned by the entry being replaced — + // so release and eviction below cannot drop them out from under + // the transaction. + let mut shared: Vec = Vec::new(); + let mut new_segments: Vec<(String, SegmentHandle)> = Vec::new(); + let mut new_bytes = 0u64; + let mut shared_bytes = 0u64; + for &(digest, slice) in segments.iter() { + if new_segments.iter().any(|(d, _)| d == digest) { + // Within-admission duplicate digest: the validation pass + // already proved both slices have identical content, so + // keep the first copy. + shared_bytes += slice.len() as u64; + continue; + } + match inner.segments.get(digest) { + // Pool already holds this exact content: share it, whoever + // currently owns it. + Some(handle) if handle.bytes.as_ref() == slice => { + shared_bytes += slice.len() as u64; + shared.push(digest.to_string()); + } + // Same digest text, different bytes in the pool. + Some(_) => { + // Replacing an entry at the same key legitimately + // re-uses a digest with new content: the old owner is + // about to be released. Anything else is a conflict — + // a stale handle another live entry still references + // must never be swapped underneath it. + let replacing_same_key = inner + .map + .get(&cache_key) + .is_some_and(|entry| entry.payload.segment_digests().contains(&digest)); + if !replacing_same_key { + return Err(L2InsertRefusal::ConflictingSegment { + digest: digest.to_string(), + detail: "the pool holds different bytes under this digest \ + for another live entry" + .to_string(), + }); + } + new_bytes = new_bytes.saturating_add(slice.len() as u64); + new_segments.push(( + digest.to_string(), + SegmentHandle { + bytes: Arc::new(slice.to_vec()), + }, + )); + } + None => { + new_bytes = new_bytes.saturating_add(slice.len() as u64); + new_segments.push(( + digest.to_string(), + SegmentHandle { + bytes: Arc::new(slice.to_vec()), + }, + )); + } + } + } + if new_bytes > self.budget_bytes { + self.stats + .refused_bytes + .fetch_add(new_bytes, Ordering::Relaxed); + return Err(L2InsertRefusal::OverBudget { + payload_bytes: new_bytes, + }); + } + // Admission is all-or-nothing. Every mutation from here is + // journaled (`AdmitJournal`, module-level); if eviction cannot + // bring the final pool footprint under budget (the incoming + // entry's own handles are pinned, so an admission sharing bytes + // with its victim can exceed what eviction frees), the journal is + // rolled back and the admission is refused without touching the + // tier. + let mut journal = AdmitJournal { + reserved: Vec::new(), + overwrites: Vec::new(), + overwritten_bytes: 0, + released_orphans: Vec::new(), + removed_entries: Vec::new(), + reserved_bytes: 0, + released_bytes: 0, + }; + let protected_set: Vec = new_segments + .iter() + .map(|(digest, _)| digest.clone()) + .chain(shared.iter().cloned()) + .collect(); + // Reserve the new handles in the pool before releasing anything, + // so a digest re-used with new content is unambiguous from here on + // and the incoming entry's bytes cannot be dropped mid-transaction. + for (digest, handle) in &new_segments { + let handle_len = handle.bytes.len() as u64; + inner.bytes = inner.bytes.saturating_add(handle_len); + journal.reserved_bytes = journal.reserved_bytes.saturating_add(handle_len); + if let Some(previous) = inner.segments.insert(digest.clone(), handle.clone()) { + // Same digest text re-used with new content: only possible + // when replacing the same key, which still holds the old + // handle. The previous bytes leave the pool now (net + // reserved delta is `new − old`); journal them so rollback + // restores the original count. + let previous_len = previous.bytes.len() as u64; + inner.bytes = inner.bytes.saturating_sub(previous_len); + journal.overwritten_bytes = journal.overwritten_bytes.saturating_add(previous_len); + journal.overwrites.push((digest.clone(), previous)); + } else { + journal.reserved.push(digest.clone()); + } + } + // One entry per cache key: a re-admit at the same coordinates is a + // replacement (fresher state for the same prefix), not a duplicate. + // The old entry's segments survive release where the incoming + // layout shares them (`protected_set`), so identical-wire re-admits + // never delete their own handles. Released orphan handles are + // journaled so rollback reinstates them, and segments that stay + // are transfer-charged to their surviving owners before the new + // entry lands. + if let Some(existing) = inner.map.remove(&cache_key) { + let digests = existing.payload.segment_digests(); + Self::recompute_all_charges(&mut inner); + for digest in digests { + if protected_set.iter().any(|p| p == digest) { + continue; + } + let still_referenced = inner + .map + .values() + .any(|other| other.payload.segment_digests().contains(&digest)); + if still_referenced { + continue; + } + if let Some(handle) = inner.segments.remove(digest) { + let released = handle.bytes.len() as u64; + inner.bytes = inner.bytes.saturating_sub(released); + journal.released_bytes = journal.released_bytes.saturating_add(released); + journal.released_orphans.push((digest.to_string(), handle)); + } + } + journal.removed_entries.push((cache_key.clone(), existing)); + } + // Evict to make room: the reservation already counts toward + // `inner.bytes`, so the pool (including this admission's distinct + // bytes) must fit the whole budget. Shared handles are pinned and + // can keep a victim from freeing — those retained bytes transfer + // to this entry's charge below. + let evictions = self.evict_to_limit( + &mut inner, + self.budget_bytes, + &cache_key, + &protected_set, + &mut journal, + ); + // The hard byte cap: if eviction could not free the excess even by + // evicting every non-pinned entry, the admission is refused and + // rolled back — the tier never sits over budget after `admit`. + if inner.bytes > self.budget_bytes { + let pool_bytes = inner.bytes; + let rolled_back_evictions = evictions.len() as u64; + journal.rollback(&mut inner); + // Charges were recomputed during eviction; restore them to + // match the rolled-back state. + Self::recompute_all_charges(&mut inner); + self.stats + .evictions + .fetch_sub(rolled_back_evictions, Ordering::Relaxed); + let over = pool_bytes.saturating_sub(self.budget_bytes); + self.stats.refused_bytes.fetch_add(over, Ordering::Relaxed); + return Err(L2InsertRefusal::ProtectedOvercommit { + pool_bytes, + budget_bytes: self.budget_bytes, + }); + } + // The entry lands with a zero charge; the deterministic recompute + // below assigns it exactly the pooled segments it owns (segments + // whose lowest-key live reference it is) and refreshes every other + // entry's charge, so the sum of charges always equals the pool. + inner.clock = inner.clock.wrapping_add(1); + let last_used = inner.clock; + self.stats + .shared_bytes_admitted + .fetch_add(shared_bytes, Ordering::Relaxed); + inner.map.insert( + cache_key.clone(), + L2Entry { + payload: mirror, + token_count, + payload_digest, + origin, + last_used, + charge_bytes: 0, + payload_bytes, + }, + ); + // Assign every pooled segment exactly once to its lowest-key live + // reference — the inserted entry included. + Self::recompute_all_charges(&mut inner); + self.stats.inserts.fetch_add(1, Ordering::Relaxed); + Ok(evictions) + } + + /// A verified hit records recency and returns the entry's layout with + /// `Arc` clones of its segment handles (no byte copies). Digests are + /// not re-hashed: admission verified the wire, and segments are + /// immutable afterward. + /// + /// If a segment handle the entry references is missing from the pool, + /// the entry is corrupt: the hit is downgraded to a miss, the entry and + /// its surviving segments are removed, LRU recency never moves, and the + /// miss counter is incremented. + pub fn get(&self, cache_key: &str) -> Option { + let mut inner = self.inner.lock().expect("L2 map lock poisoned"); + let digests: Vec = match inner.map.get(cache_key) { + Some(entry) => entry + .payload + .segment_digests() + .into_iter() + .map(str::to_string) + .collect(), + // Absent key: a miss, never an LRU touch. + None => { + self.stats.misses.fetch_add(1, Ordering::Relaxed); + return None; + } + }; + let missing = digests + .iter() + .any(|digest| !inner.segments.contains_key(digest)); + if missing { + // Corrupt entry: a segment handle vanished without an entry + // removal. Serve a miss, never partial bytes; drop the entry + // and its surviving handles; recency stays untouched. + let removed = inner.map.remove(cache_key); + if let Some(entry) = removed { + Self::recompute_all_charges(&mut inner); + self.release_entry_segments(&mut inner, &entry, &[]); + } + self.stats.misses.fetch_add(1, Ordering::Relaxed); + return None; + } + let now = { + inner.clock = inner.clock.wrapping_add(1); + inner.clock + }; + let entry = inner.map.get_mut(cache_key).expect("entry checked above"); + entry.last_used = now; + let payload = entry.payload.clone(); + let token_count = entry.token_count; + let payload_digest = entry.payload_digest.clone(); + self.stats.hits.fetch_add(1, Ordering::Relaxed); + let mut segments = HashMap::with_capacity(digests.len()); + for digest in &digests { + let handle = inner + .segments + .get(digest) + .expect("all digests checked above"); + segments.insert(digest.clone(), handle.clone()); + } + Some(L2Hit { + payload, + token_count, + payload_digest, + segments, + }) + } + + /// Presence probe: side-effect free. It does not touch LRU recency — + /// prefix probing must not make entries hot — and returns only + /// metadata. Recency is updated by `get` after a successful verified + /// hit. + pub fn peek(&self, cache_key: &str) -> Option { + let inner = self.inner.lock().expect("L2 map lock poisoned"); + let entry = inner.map.get(cache_key)?; + Some(L2Peek { + token_count: entry.token_count, + payload_digest: entry.payload_digest.clone(), + payload_bytes: entry.payload_bytes, + distinct_bytes: entry.charge_bytes, + origin: entry.origin, + }) + } + + pub fn remove(&self, cache_key: &str) -> Option { + let mut inner = self.inner.lock().expect("L2 map poisoned"); + let removed = inner.map.remove(cache_key)?; + // Retained bytes come from the pool's actual references: segments + // of this entry that other entries still reference after the + // removal stay in the pool. Charged bytes are never transferred + // between entries — a survivor's charge already excludes shared + // segments — so this cannot drift the survivors' accounting below + // the physical pool they own. + let mut retained = 0u64; + for digest in removed.payload.segment_digests() { + let referenced_elsewhere = inner + .map + .values() + .any(|other| other.payload.segment_digests().contains(&digest)); + if referenced_elsewhere { + retained = retained.saturating_add( + inner + .segments + .get(digest) + .map(|handle| handle.bytes.len() as u64) + .unwrap_or(0), + ); + } + } + // Segments that stay are now physically owned by the survivors: + // recompute charges from the live map (the removed entry is + // already out of it). + Self::recompute_all_charges(&mut inner); + let before = inner.bytes; + self.release_entry_segments(&mut inner, &removed, &[]); + let freed = before.saturating_sub(inner.bytes); + Some(L2Eviction { + cache_key: cache_key.to_string(), + freed_bytes: freed, + retained_bytes: retained, + }) + } + + /// Drop everything; returns the distinct bytes released. + pub fn clear(&self) -> u64 { + let mut inner = self.inner.lock().expect("L2 map lock poisoned"); + let bytes = inner.bytes; + inner.map.clear(); + inner.segments.clear(); + inner.bytes = 0; + bytes + } + + pub fn len(&self) -> usize { + self.inner.lock().expect("L2 map lock poisoned").map.len() + } + + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// Test-only: drop every pool handle without touching the entry map, + /// to force the missing-handle corruption path in `get`. + #[cfg(test)] + fn clear_pool_for_test(&self) { + let mut inner = self.inner.lock().expect("L2 map lock poisoned"); + inner.segments.clear(); + inner.bytes = 0; + } + + /// Point-in-time snapshot combining atomics with the locked totals. + pub fn stats(&self) -> L2Stats { + let inner = self.inner.lock().expect("L2 map lock poisoned"); + let logical_bytes = inner + .map + .values() + .map(|entry| entry.payload_bytes) + .sum::(); + L2Stats { + entries: inner.map.len() as u64, + bytes: inner.bytes, + logical_bytes, + segments: inner.segments.len() as u64, + segment_bytes: inner.bytes, + shared_bytes_admitted: self.stats.shared_bytes_admitted.load(Ordering::Relaxed), + budget_bytes: self.budget_bytes, + hits: self.stats.hits.load(Ordering::Relaxed), + misses: self.stats.misses.load(Ordering::Relaxed), + inserts: self.stats.inserts.load(Ordering::Relaxed), + evictions: self.stats.evictions.load(Ordering::Relaxed), + admission_rejects: self.stats.admission_rejects.load(Ordering::Relaxed), + refused_bytes: self.stats.refused_bytes.load(Ordering::Relaxed), + } + } + + /// Recompute every entry's charge from the live map: each pooled + /// segment is assigned exactly once, to its lowest-cache-key live + /// reference, and each entry's `charge_bytes` is the sum of the + /// segments it owns. The sum of all charges therefore always equals + /// the physical pool bytes — after admission, eviction, removal, + /// corruption cleanup, and rollback alike. Deterministic: identical + /// map states produce identical ownership. + fn recompute_all_charges(inner: &mut L2Inner) { + let mut ownership: HashMap = HashMap::new(); + for digest in inner.segments.keys() { + let Some(bytes) = inner.segments.get(digest).map(|h| h.bytes.len() as u64) else { + continue; + }; + let owner = inner + .map + .iter() + .filter(|(_, entry)| entry.payload.segment_digests().contains(&digest.as_str())) + .map(|(key, _)| key.clone()) + .min(); + if let Some(owner) = owner { + *ownership.entry(owner).or_insert(0) += bytes; + } + } + for (key, entry) in inner.map.iter_mut() { + entry.charge_bytes = ownership.get(key).copied().unwrap_or(0); + } + } + + /// Drop an entry's segments from the pool, decrementing the pool byte + /// total. Segments still referenced by another live entry, or pinned by + /// an in-flight admission (`protected`), stay. Zero-byte segments are + /// dropped without accounting (a pool without payload bytes must never + /// charge the budget). + fn release_entry_segments(&self, inner: &mut L2Inner, entry: &L2Entry, protected: &[String]) { + for digest in entry.payload.segment_digests() { + if protected.iter().any(|p| p == digest) { + continue; + } + let still_referenced = inner + .map + .values() + .any(|other| other.payload.segment_digests().contains(&digest)); + if still_referenced { + continue; + } + if let Some(handle) = inner.segments.remove(digest) { + let released = handle.bytes.len() as u64; + inner.bytes = inner.bytes.saturating_sub(released); + } + } + } + + /// Evict in deterministic LRU order until the pool fits `limit` bytes. + /// Shared segments are released only with their last referencing + /// entry; handles pinned by the in-flight admission (`protected`) are + /// never released; a victim that frees nothing is still counted as an + /// eviction. Every mutation is recorded in `journal` so a refused + /// admission can roll the evictions back exactly. + fn evict_to_limit( + &self, + inner: &mut L2Inner, + limit: u64, + protect_key: &str, + protected: &[String], + journal: &mut AdmitJournal, + ) -> Vec { + let mut evictions = Vec::new(); + while inner.bytes > limit { + // Deterministic LRU: lowest last_used wins; ties break on cache + // key so identical operation sequences produce identical + // evictions. + let victim = inner + .map + .iter() + .filter(|(key, _)| key.as_str() != protect_key) + .min_by(|a, b| a.1.last_used.cmp(&b.1.last_used).then_with(|| a.0.cmp(b.0))) + .map(|(key, _)| key.clone()); + let Some(victim) = victim else { break }; + let Some(removed) = inner.map.remove(&victim) else { + break; + }; + // Retained bytes from actual pool references, computed before + // release: segments of the victim that survivors still + // reference, or that the in-flight admission pins (its entry + // is not in the map yet, but it will own them). + let mut retained = 0u64; + for digest in removed.payload.segment_digests() { + let referenced_elsewhere = inner + .map + .values() + .any(|other| other.payload.segment_digests().contains(&digest)) + || protected.iter().any(|p| p == digest); + if referenced_elsewhere { + retained = retained.saturating_add( + inner + .segments + .get(digest) + .map(|handle| handle.bytes.len() as u64) + .unwrap_or(0), + ); + } + } + let before = inner.bytes; + // Charges are recomputed from the live map (journaled state + // is restored exactly; ownership follows the lowest key). + Self::recompute_all_charges(inner); + for digest in removed.payload.segment_digests() { + if protected.iter().any(|p| p == digest) { + continue; + } + let still_referenced = inner + .map + .values() + .any(|other| other.payload.segment_digests().contains(&digest)); + if still_referenced { + continue; + } + if let Some(handle) = inner.segments.remove(digest) { + let released = handle.bytes.len() as u64; + inner.bytes = inner.bytes.saturating_sub(released); + journal.released_bytes = journal.released_bytes.saturating_add(released); + journal.released_orphans.push((digest.to_string(), handle)); + } + } + let freed = before.saturating_sub(inner.bytes); + self.stats.evictions.fetch_add(1, Ordering::Relaxed); + journal.removed_entries.push((victim.clone(), removed)); + evictions.push(L2Eviction { + cache_key: victim, + freed_bytes: freed, + retained_bytes: retained, + }); + } + evictions + } +} + +/// Build the L2 cache key from the same coordinates L3 uses, plus the +/// identities the tier serves. Same coordinates under different identities +/// get different keys: an identity change cannot cross-contaminate. +pub fn l2_cache_key( + model_identity: &str, + state_identity: &str, + namespace: &str, + token_ids: &[i32], +) -> String { + let namespace_key = crate::tier::l3_namespace_key(namespace); + let prefix_key = crate::tier::l3_prefix_key(namespace, token_ids); + let mut hasher = blake3::Hasher::new(); + hasher.update(b"l2-cache-key-v1"); + hasher.update(model_identity.as_bytes()); + hasher.update(b"\0"); + hasher.update(state_identity.as_bytes()); + hasher.update(b"\0"); + hasher.update(namespace_key.as_bytes()); + hasher.update(prefix_key.as_bytes()); + format!("blake3:{}", hasher.finalize().to_hex()) +} + +#[cfg(test)] +mod tests { + use super::*; + + const DIGEST_B: &str = "b616719e0a0d39dc0fe85cd2d0a5e0e2f5e6e10b6b5a0a6f1a1c1d3e5f708a90"; + + thread_local! { + static FULL_WIRE: std::cell::RefCell> = const { std::cell::RefCell::new(Vec::new()) }; + } + + /// Build a synthetic wire of `len` bytes and its true BLAKE3 digest. + fn wire(len: usize, fill: u8) -> (Vec, String) { + // Position-dependent bytes so equal-length slices are never equal + // content: segment digests stay distinct across the wire. + let bytes: Vec = (0..len) + .map(|i| (fill as usize + i) % 251) + .map(|v| v as u8) + .collect(); + let digest = segment_digest(&bytes); + FULL_WIRE.with(|cell| *cell.borrow_mut() = bytes.clone()); + (bytes, digest) + } + + /// Single-segment full-state mirror over `len` wire bytes. The segment + /// key is the content digest of the whole wire, as L3 would produce. + fn single_segment_mirror(w: &[u8]) -> ExactStatePayloadMirror { + let len = w.len() as u64; + ExactStatePayloadMirror::FullState { + layout: L2Layout { + payload_kind: ExactStatePayloadKind::FullState, + total_bytes: len, + kv_bytes: len, + recurrent_bytes: 0, + segments: vec![(segment_digest(w), 0..len)], + }, + } + } + + /// Manifest-shaped mirror: digest-keyed segments cut at every + /// `segment_len` boundary, matching how `from_manifest` tiles. + fn manifest_shaped_mirror(w: &[u8], segment_len: u64) -> ExactStatePayloadMirror { + // `w` is a suffix of the test's full wire: segment digests are + // keyed by offset in that full wire so entries sharing a prefix + // also share segment identity. + let full = FULL_WIRE.with(|cell| cell.borrow().clone()); + let len = w.len() as u64; + let mut segments = Vec::new(); + let mut offset = 0u64; + while offset < len { + let end = (offset + segment_len).min(len); + let digest = segment_digest(&full[offset as usize..end as usize]); + segments.push((digest, offset..end)); + offset = end; + } + ExactStatePayloadMirror::FullState { + layout: L2Layout { + payload_kind: ExactStatePayloadKind::FullState, + total_bytes: len, + kv_bytes: len, + recurrent_bytes: 0, + segments, + }, + } + } + + fn key(namespace: &str, tokens: &[i32]) -> String { + l2_cache_key("model-a", "state-a", namespace, tokens) + } + + #[test] + fn admit_get_round_trip_serves_wire_bytes() { + let tier = L2Tier::new(1 << 20); + let k = key("ns", &[1, 2, 3]); + let (w, digest) = wire(64, 7); + tier.admit( + k.clone(), + 3, + digest.clone(), + &w, + single_segment_mirror(&w), + L2Origin::FromL3, + ) + .expect("admission must fit"); + let hit = tier.get(&k).expect("admitted key must hit"); + assert_eq!(hit.token_count, 3); + assert_eq!(hit.payload.byte_len(), 64); + assert_eq!(hit.payload_digest, digest); + // Round-trips into a serving payload with the right byte count and + // exactly the admitted wire bytes. + let payload = hit.to_payload(); + assert_eq!(payload.byte_len(), 64); + assert_eq!( + payload.kind(), + crate::payload::ExactStatePayloadKind::FullState + ); + let (bytes, _) = payload.full_state_bytes_timed().expect("full state"); + assert_eq!(bytes.as_ref(), &w[..], "served bytes must equal the wire"); + } + + #[test] + fn admission_digest_mismatch_refuses_and_stores_nothing() { + let tier = L2Tier::new(1 << 20); + let k = key("ns", &[1, 2, 3]); + let (w, _) = wire(64, 7); + let err = tier + .admit( + k.clone(), + 3, + DIGEST_B.to_string(), + &w, + single_segment_mirror(&w), + L2Origin::Direct, + ) + .expect_err("a wire that does not hash to the claimed digest must be refused"); + assert!(matches!(err, L2InsertRefusal::DigestMismatch { .. })); + assert!(tier.peek(&k).is_none(), "refused bytes must not be stored"); + assert!(tier.get(&k).is_none()); + let stats = tier.stats(); + assert_eq!(stats.admission_rejects, 1); + assert_eq!(stats.entries, 0); + assert_eq!(stats.bytes, 0); + } + + #[test] + fn corrupted_wire_never_reaches_the_tier() { + // L2Origin::Direct with arbitrary bytes under a valid-looking + // digest is exactly the hole this closes: the digest check runs on + // the actual bytes. + let tier = L2Tier::new(1 << 20); + let k = key("ns", &[9]); + let (mut w, digest) = wire(128, 3); + w[42] ^= 0xff; // one flipped bit + let err = tier + .admit( + k.clone(), + 1, + digest, + &w, + single_segment_mirror(&w), + L2Origin::Direct, + ) + .expect_err("corrupted wire must be refused at admission"); + assert!(matches!(err, L2InsertRefusal::DigestMismatch { .. })); + assert_eq!(tier.stats().admission_rejects, 1); + assert!(tier.is_empty()); + } + + #[test] + fn peek_is_side_effect_free_for_lru() { + let tier = L2Tier::new(160); + let k1 = key("ns", &[1]); + let k2 = key("ns", &[2]); + let (w1, d1) = wire(64, 1); + let (w2, d2) = wire(64, 2); + tier.admit( + k1.clone(), + 1, + d1, + &w1, + single_segment_mirror(&w1), + L2Origin::FromL3, + ) + .expect("k1 fits"); + tier.admit( + k2.clone(), + 1, + d2, + &w2, + single_segment_mirror(&w2), + L2Origin::FromL3, + ) + .expect("k2 fits"); + // Probe k1 many times: recency must not move. + for _ in 0..10 { + assert!(tier.peek(&k1).is_some()); + } + // Insert a third entry: k1 (never truly used) must still be the + // LRU victim, not k2. + let k3 = key("ns", &[3]); + let (w3, d3) = wire(64, 3); + let evictions = tier + .admit( + k3.clone(), + 1, + d3, + &w3, + single_segment_mirror(&w3), + L2Origin::FromL3, + ) + .expect("k3 fits after eviction"); + assert_eq!(evictions.len(), 1); + assert_eq!(evictions[0].cache_key, k1, "probed-but-unused k1 is LRU"); + assert!(tier.peek(&k2).is_some(), "untouched k2 survives"); + } + + #[test] + fn recency_moves_only_on_verified_hit() { + let tier = L2Tier::new(160); + let k1 = key("ns", &[1]); + let k2 = key("ns", &[2]); + let (w1, d1) = wire(64, 1); + let (w2, d2) = wire(64, 2); + tier.admit( + k1.clone(), + 1, + d1, + &w1, + single_segment_mirror(&w1), + L2Origin::FromL3, + ) + .expect("k1 fits"); + tier.admit( + k2.clone(), + 1, + d2, + &w2, + single_segment_mirror(&w2), + L2Origin::FromL3, + ) + .expect("k2 fits"); + // A real hit on k1 makes k2 the victim of the next insertion. + assert!(tier.get(&k1).is_some()); + let k3 = key("ns", &[3]); + let (w3, d3) = wire(64, 3); + let evictions = tier + .admit(k3, 1, d3, &w3, single_segment_mirror(&w3), L2Origin::FromL3) + .expect("k3 fits after eviction"); + assert_eq!(evictions.len(), 1); + assert_eq!(evictions[0].cache_key, k2, "k2 is now LRU"); + assert!(tier.peek(&k1).is_some(), "recently hit k1 survives"); + } + + #[test] + fn prefix_growth_shares_segment_bytes_instead_of_duplicating() { + // 16 KiB of four 4 KiB segments; the shorter prefix shares the + // first three segments with the longer one. + let segment_len = 4096u64; + let total = segment_len * 4; + let tier = L2Tier::new(total * 2); + let short = key("ns", &[1, 2, 3]); + let long = key("ns", &[1, 2, 3, 4, 5]); + let (w, digest) = wire(total as usize, 5); + let short_len = segment_len * 3; + tier.admit( + short.clone(), + 3, + segment_digest(&w[..short_len as usize]), + &w[..short_len as usize], + manifest_shaped_mirror(&w[..short_len as usize], segment_len), + L2Origin::FromL3, + ) + .expect("short prefix admitted"); + + tier.admit( + long.clone(), + 5, + digest, + &w, + manifest_shaped_mirror(&w, segment_len), + L2Origin::FromL3, + ) + .expect("long prefix admitted"); + + let stats = tier.stats(); + // The long entry pays only for its one new (4th) segment. + assert_eq!( + stats.bytes, total, + "pool must hold distinct segment bytes once: got {}", + stats.bytes + ); + assert_eq!( + stats.logical_bytes, + total + short_len, + "logical bytes count both entries' full wires" + ); + assert_eq!(stats.segments, 4, "four distinct segments, not seven"); + assert_eq!(stats.shared_bytes_admitted, short_len); + // Both entries serve their own wire slices. + let hit = tier.get(&long).expect("long hit"); + let payload = hit.to_payload(); + let (bytes, _) = payload.full_state_bytes_timed().expect("bytes"); + assert_eq!(bytes.as_ref(), &w[..]); + let hit_short = tier.get(&short).expect("short hit"); + let payload_short = hit_short.to_payload(); + let (bytes_short, _) = payload_short.full_state_bytes_timed().expect("bytes"); + assert_eq!(bytes_short.as_ref(), &w[..short_len as usize]); + } + + #[test] + fn evicting_one_entry_keeps_shared_prefix_segments() { + let segment_len = 4096u64; + let total = segment_len * 4; + let tier = L2Tier::new(total * 2); + let short = key("ns", &[1]); + let long = key("ns", &[2]); + let (w, digest) = wire(total as usize, 6); + let short_len = segment_len * 3; + tier.admit( + short, + 3, + segment_digest(&w[..short_len as usize]), + &w[..short_len as usize], + manifest_shaped_mirror(&w[..short_len as usize], segment_len), + L2Origin::FromL3, + ) + .expect("short admitted"); + tier.admit( + long.clone(), + 5, + digest, + &w, + manifest_shaped_mirror(&w, segment_len), + L2Origin::FromL3, + ) + .expect("long admitted"); + // Removing the long entry frees only its exclusive tail segment. + let removed = tier.remove(&long).expect("long entry present"); + assert_eq!(removed.freed_bytes, segment_len); + assert_eq!( + removed.retained_bytes, short_len, + "shared prefix bytes are retained by the shorter entry" + ); + let stats = tier.stats(); + assert_eq!(stats.bytes, short_len); + assert_eq!(stats.segments, 3); + assert!(tier.get(&key("ns", &[1])).is_some(), "short entry intact"); + } + + #[test] + fn budget_evicts_lru_first_and_never_the_protected_entry() { + let tier = L2Tier::new(256); + let k1 = key("ns", &[1]); + let k2 = key("ns", &[2]); + let k3 = key("ns", &[3]); + let (w1, d1) = wire(100, 1); + let (w2, d2) = wire(100, 2); + let (w3, d3) = wire(100, 3); + tier.admit( + k1.clone(), + 1, + d1, + &w1, + single_segment_mirror(&w1), + L2Origin::FromL3, + ) + .expect("k1 fits"); + tier.admit( + k2.clone(), + 1, + d2, + &w2, + single_segment_mirror(&w2), + L2Origin::FromL3, + ) + .expect("k2 fits"); + // Touch k1 so k2 becomes the LRU victim. + assert!(tier.get(&k1).is_some()); + let evictions = tier + .admit( + k3.clone(), + 1, + d3, + &w3, + single_segment_mirror(&w3), + L2Origin::FromL3, + ) + .expect("k3 fits after eviction"); + assert_eq!( + evictions.len(), + 1, + "one entry must be evicted: {evictions:?}" + ); + assert_eq!(evictions[0].cache_key, k2, "LRU victim is k2"); + assert_eq!(evictions[0].freed_bytes, 100); + assert_eq!(evictions[0].retained_bytes, 0); + assert!(tier.peek(&k1).is_some(), "recently used k1 survives"); + assert!(tier.peek(&k3).is_some(), "just-admitted k3 survives"); + assert!(tier.peek(&k2).is_none(), "k2 was evicted"); + let stats = tier.stats(); + assert_eq!(stats.evictions, 1); + assert_eq!(stats.bytes, 200, "pool bytes must track survivors exactly"); + } + + #[test] + fn oversized_distinct_bytes_are_refused_without_evicting() { + let tier = L2Tier::new(128); + let k1 = key("ns", &[1]); + let (w1, d1) = wire(64, 1); + tier.admit( + k1.clone(), + 1, + d1, + &w1, + single_segment_mirror(&w1), + L2Origin::FromL3, + ) + .expect("fits"); + let (w2, d2) = wire(129, 2); + let err = tier + .admit( + key("ns", &[2]), + 1, + d2, + &w2, + single_segment_mirror(&w2), + L2Origin::FromL3, + ) + .expect_err("over-budget payload must be refused"); + assert_eq!(err, L2InsertRefusal::OverBudget { payload_bytes: 129 }); + assert!(tier.peek(&k1).is_some(), "refusal must not evict anything"); + assert_eq!(tier.stats().refused_bytes, 129); + } + + #[test] + fn empty_payload_is_refused_like_l3() { + let tier = L2Tier::new(1 << 20); + let empty_digest = segment_digest(&[]); + let err = tier + .admit( + key("ns", &[1]), + 1, + empty_digest, + &[], + single_segment_mirror(&[]), + L2Origin::FromL3, + ) + .expect_err("empty payloads must be refused"); + assert_eq!(err, L2InsertRefusal::EmptyPayload); + assert!(tier.is_empty()); + } + + #[test] + fn malformed_digest_is_refused_before_any_hashing() { + let tier = L2Tier::new(1 << 20); + let (w, _) = wire(16, 4); + let err = tier + .admit( + key("ns", &[1]), + 1, + "not-a-digest".to_string(), + &w, + single_segment_mirror(&w), + L2Origin::FromL3, + ) + .expect_err("malformed digest must be refused"); + assert_eq!(err, L2InsertRefusal::MalformedDigest); + } + + #[test] + fn readmit_replaces_and_keeps_accounting_exact() { + let tier = L2Tier::new(1 << 20); + let k = key("ns", &[9]); + let (w1, d1) = wire(100, 1); + let (w2, d2) = wire(40, 2); + tier.admit( + k.clone(), + 3, + d1, + &w1, + single_segment_mirror(&w1), + L2Origin::FromL3, + ) + .expect("first admission"); + let evictions = tier + .admit( + k.clone(), + 3, + d2.clone(), + &w2, + single_segment_mirror(&w2), + L2Origin::FromL3, + ) + .expect("replacement"); + assert!(evictions.is_empty()); + assert_eq!(tier.len(), 1); + assert_eq!( + tier.stats().bytes, + 40, + "replacement must release the old bytes" + ); + // New digest is the one served now. + let hit = tier.get(&k).expect("replacement hit"); + assert_eq!(hit.payload_digest, d2); + let payload = hit.to_payload(); + let (bytes, _) = payload.full_state_bytes_timed().expect("bytes"); + assert_eq!(bytes.as_ref(), &w2[..]); + } + + #[test] + fn identical_coordinates_under_different_identities_get_different_keys() { + let a = l2_cache_key("model-a", "state-a", "ns", &[1, 2]); + let b = l2_cache_key("model-b", "state-a", "ns", &[1, 2]); + let c = l2_cache_key("model-a", "state-b", "ns", &[1, 2]); + assert_ne!(a, b); + assert_ne!(a, c); + // Same coordinates, same identities: stable key. + let a2 = l2_cache_key("model-a", "state-a", "ns", &[1, 2]); + assert_eq!(a, a2); + // Different token paths differ. + assert_ne!(a, l2_cache_key("model-a", "state-a", "ns", &[1, 3])); + } + + #[test] + fn mirror_round_trips_every_payload_kind_from_manifest() { + // kv-recurrent: kv 24 bytes then recurrent 8, cut into two segments. + let (kv_wire, _) = wire(24, 3); + let (rec_wire, _) = wire(8, 4); + let wire_bytes: Vec = [kv_wire, rec_wire].concat(); + let digest = segment_digest(&wire_bytes); + let manifest = HandoffManifest { + version: MANIFEST_VERSION, + model_identity: "blake3:model".to_string(), + state_identity: "blake3:state".to_string(), + payload_kind: "kv-recurrent".to_string(), + total_bytes: wire_bytes.len() as u64, + payload_digest: digest.clone(), + segments: vec![ + HandoffSegmentRef { + index: 0, + offset: 0, + bytes: 16, + digest: segment_digest(&wire_bytes[..16]), + meta_json: None, + }, + HandoffSegmentRef { + index: 1, + offset: 16, + bytes: 16, + digest: segment_digest(&wire_bytes[16..]), + meta_json: None, + }, + ], + kv_bytes: 24, + recurrent_bytes: 8, + kv_desc_json: None, + token_count: 4, + continuation_token: 0, + expected_tokens: Vec::new(), + }; + let mirror = ExactStatePayloadMirror::from_manifest(&manifest).expect("manifest parses"); + assert_eq!(mirror.byte_len(), 32); + assert_eq!(mirror.kind(), ExactStatePayloadKind::KvRecurrent); + + let tier = L2Tier::new(1 << 20); + let k = key("ns", &[7]); + tier.admit(k.clone(), 4, digest, &wire_bytes, mirror, L2Origin::FromL3) + .expect("fits"); + let hit = tier.get(&k).expect("hit"); + let payload = hit.to_payload(); + assert_eq!(payload.byte_len(), 32); + let served_kv = payload + .kv_bytes() + .expect("kv bytes") + .expect("kv-recurrent has kv") + .into_owned(); + let served_rec = payload + .recurrent_state_bytes() + .expect("recurrent bytes") + .into_owned(); + assert_eq!(served_kv, wire_bytes[..24], "kv slice must match the wire"); + assert_eq!(served_rec, wire_bytes[24..], "recurrent slice matches"); + } + + #[test] + fn from_manifest_rejects_unknown_kind_and_bad_tiling() { + let mut manifest = HandoffManifest { + version: MANIFEST_VERSION, + model_identity: "m".to_string(), + state_identity: "s".to_string(), + payload_kind: "blob".to_string(), + total_bytes: 10, + payload_digest: "blake3:aa".to_string(), + segments: Vec::new(), + kv_bytes: 10, + recurrent_bytes: 0, + kv_desc_json: None, + token_count: 1, + continuation_token: 0, + expected_tokens: Vec::new(), + }; + let err = ExactStatePayloadMirror::from_manifest(&manifest) + .expect_err("unknown kind must be refused"); + assert!(matches!(err, L2InsertRefusal::UnknownPayloadKind(_))); + + manifest.payload_kind = "full-state".to_string(); + manifest.segments = vec![HandoffSegmentRef { + index: 0, + offset: 0, + bytes: 7, + digest: "blake3:seg".to_string(), + meta_json: None, + }]; + let err = ExactStatePayloadMirror::from_manifest(&manifest) + .expect_err("tiling mismatch must be refused"); + assert!(matches!(err, L2InsertRefusal::MalformedManifest(_))); + } + + #[test] + fn multi_segment_reads_reassemble_exact_wire() { + // 6 KiB in 1 KiB segments: every read path crosses many blocks. + let segment_len = 1024u64; + let total = segment_len * 6; + let (w, digest) = wire(total as usize, 9); + let tier = L2Tier::new(total * 2); + let k = key("ns", &[1]); + tier.admit( + k.clone(), + 6, + digest, + &w, + manifest_shaped_mirror(&w, segment_len), + L2Origin::FromL3, + ) + .expect("fits"); + let hit = tier.get(&k).expect("hit"); + let payload = hit.to_payload(); + let (bytes, reconstruct) = payload.full_state_bytes_timed().expect("bytes"); + assert_eq!(bytes.as_ref(), &w[..]); + // Multiple distinct segment storages materialize on read; the + // reconstruction length must equal the payload either way. + assert_eq!( + reconstruct.reconstruct_bytes, total, + "multi-segment reads materialize the wire exactly once" + ); + } + + #[test] + fn clear_releases_everything_and_reports_bytes() { + let tier = L2Tier::new(1 << 20); + for i in 0..5i32 { + let (w, d) = wire(64, i as u8 + 10); + tier.admit( + key("ns", &[i]), + 1, + d, + &w, + single_segment_mirror(&w), + L2Origin::FromL3, + ) + .expect("fits"); + } + let released = tier.clear(); + assert_eq!(released, 320); + assert!(tier.is_empty()); + assert_eq!(tier.stats().bytes, 0); + assert_eq!(tier.stats().segments, 0); + } + + #[test] + fn remove_is_exact() { + let tier = L2Tier::new(1 << 20); + let k = key("ns", &[4]); + let (w, d) = wire(64, 8); + tier.admit( + k.clone(), + 1, + d, + &w, + single_segment_mirror(&w), + L2Origin::FromL3, + ) + .expect("fits"); + let removed = tier.remove(&k).expect("present entry removes"); + assert_eq!(removed.freed_bytes, 64); + assert_eq!(removed.retained_bytes, 0); + assert!(tier.remove(&k).is_none(), "second remove is None"); + assert_eq!(tier.stats().bytes, 0); + assert_eq!(tier.stats().segments, 0); + } + + #[test] + fn identical_wire_same_key_readmit_keeps_its_own_segments() { + // The original failure: re-admitting an identical wire at the same + // key classified the existing pool segments as shared, released the + // old entry's last references, inserted no replacement handles, and + // left an entry whose segments were absent + // (`declared=14 restored=0 pool=0`). + let tier = L2Tier::new(1 << 20); + let k = key("ns", &[11]); + let segment_len = 16u64; + let (w, digest) = wire(64, 21); + for round in 0..3 { + let mirror = manifest_shaped_mirror(&w, segment_len); + tier.admit(k.clone(), 4, digest.clone(), &w, mirror, L2Origin::FromL3) + .expect("identical re-admit must be accepted"); + let hit = tier + .get(&k) + .unwrap_or_else(|| panic!("round {round}: re-admitted key must hit")); + let payload = hit.to_payload(); + let (bytes, _) = payload.full_state_bytes_timed().expect("bytes"); + assert_eq!( + bytes.as_ref(), + &w[..], + "round {round}: re-admitted entry must serve its full wire" + ); + let stats = tier.stats(); + assert_eq!(stats.entries, 1); + assert_eq!( + stats.segments, 4, + "round {round}: pool must still hold every segment" + ); + assert_eq!(stats.bytes, 64, "round {round}: pool bytes exact"); + } + } + + #[test] + fn eviction_cannot_release_segments_the_incoming_entry_shares() { + // Pressure case: the incoming entry shares its would-be victim's + // prefix segments. The victim is not protected by the cache-key + // filter (different key) and is not yet replaced in the map, so + // eviction could drop the shared handles before the new entry + // lands. The 64-byte budget equals the old entry's footprint, so + // the 16-byte tail reservation forces eviction mid-admission: + // only the victim's *exclusive* tail X frees (16 bytes), the + // shared prefix is pinned by the admission and survives + // byte-exact, and the pool lands exactly at budget with the new + // entry's segments. Layout: old = [S0 S1 S2 X], grown = + // [S0 S1 S2 T] with T different content from X. + let segment_len = 16u64; + let total = segment_len * 4; + let tier = L2Tier::new(total); + let (w, _) = wire(total as usize, 30); + let short_len = segment_len * 3; + + let old = key("ns", &[1]); + tier.admit( + old.clone(), + 4, + segment_digest(&w), + &w, + manifest_shaped_mirror(&w, segment_len), + L2Origin::FromL3, + ) + .expect("old entry admitted"); + + // The grown entry swaps the old tail for a different one. + let tail: Vec = (0..segment_len as usize) + .map(|i| (200usize + i) % 251) + .map(|v| v as u8) + .collect(); + let mut grown_wire = w[..short_len as usize].to_vec(); + grown_wire.extend_from_slice(&tail); + let grown_digest = segment_digest(&grown_wire); + let grown_segments = vec![ + (segment_digest(&w[..segment_len as usize]), 0..segment_len), + ( + segment_digest(&w[segment_len as usize..segment_len as usize * 2]), + segment_len..segment_len * 2, + ), + ( + segment_digest(&w[segment_len as usize * 2..short_len as usize]), + segment_len * 2..short_len, + ), + (segment_digest(&tail), short_len..total), + ]; + let grown_mirror = ExactStatePayloadMirror::FullState { + layout: L2Layout { + payload_kind: ExactStatePayloadKind::FullState, + total_bytes: total, + kv_bytes: total, + recurrent_bytes: 0, + segments: grown_segments, + }, + }; + let grown = key("ns", &[2]); + let evictions = tier + .admit( + grown.clone(), + 4, + grown_digest.clone(), + &grown_wire, + grown_mirror, + L2Origin::FromL3, + ) + .expect("admission must succeed by evicting the old entry"); + assert_eq!(evictions.len(), 1, "old entry is the victim"); + assert_eq!(evictions[0].cache_key, old); + assert_eq!( + evictions[0].freed_bytes, segment_len, + "only the victim's exclusive tail frees; the pinned prefix stays" + ); + assert_eq!( + evictions[0].retained_bytes, short_len, + "the shared prefix is retained by the incoming entry" + ); + + // The shared prefix segments must have survived the eviction. + let hit = tier.get(&grown).expect("grown entry hits"); + let payload = hit.to_payload(); + let (bytes, _) = payload.full_state_bytes_timed().expect("bytes"); + assert_eq!( + bytes.as_ref(), + &grown_wire[..], + "shared segments must survive the admission that evicted their old owner" + ); + let stats = tier.stats(); + assert_eq!(stats.entries, 1); + assert_eq!(stats.segments, 4); + assert_eq!( + stats.bytes, total, + "the pool is exactly the admitted entry's distinct bytes" + ); + assert!( + stats.bytes <= budget_from(&tier), + "the hard byte cap holds after a sharing admission" + ); + assert!(tier.peek(&old).is_none(), "old entry was evicted"); + } + + /// The budget the tier was built with (test-only mirror of the + /// constructor argument). + fn budget_from(tier: &L2Tier) -> u64 { + tier.stats().budget_bytes + } + + #[test] + fn admission_reusing_protected_bytes_cannot_exceed_the_budget() { + // 100-byte budget: X (60 bytes) is live, the incoming X+Y (120 + // bytes) shares X's segment. `new_bytes` is only Y's 60, X is + // pinned (shared), so eviction cannot free the excess: the + // admission must be refused and rolled back, never inserted at + // 120 bytes over a 100-byte budget. + let tier = L2Tier::new(100); + let k1 = key("ns", &[1]); + let k2 = key("ns", &[2]); + let (x, _) = wire(60, 1); + tier.admit( + k1.clone(), + 1, + segment_digest(&x), + &x, + single_segment_mirror(&x), + L2Origin::FromL3, + ) + .expect("X admitted"); + let (y, _) = wire(60, 2); + let mut xy = x.clone(); + xy.extend_from_slice(&y); + let xy_digest = segment_digest(&xy); + let mirror = ExactStatePayloadMirror::FullState { + layout: L2Layout { + payload_kind: ExactStatePayloadKind::FullState, + total_bytes: 120, + kv_bytes: 120, + recurrent_bytes: 0, + segments: vec![(segment_digest(&x), 0..60), (segment_digest(&y), 60..120)], + }, + }; + let err = tier + .admit(k2.clone(), 2, xy_digest, &xy, mirror, L2Origin::Direct) + .expect_err("protected overcommit must be refused"); + assert_eq!( + err, + L2InsertRefusal::ProtectedOvercommit { + pool_bytes: 120, + budget_bytes: 100, + }, + "the pool could only reach the budget by evicting the pinned X" + ); + // The tier is exactly as it was before the refused admission. + let stats = tier.stats(); + assert_eq!(stats.entries, 1); + assert_eq!(stats.bytes, 60); + assert_eq!(stats.segments, 1); + assert_eq!(stats.inserts, 1, "the refusal must not count an insert"); + assert_eq!( + stats.evictions, 0, + "rolled-back evictions must not be counted" + ); + assert!(tier.peek(&k1).is_some(), "X survived the refusal"); + assert!(tier.peek(&k2).is_none(), "the incoming entry was refused"); + // X still serves its exact bytes. + let hit = tier.get(&k1).expect("X hit"); + let payload = hit.to_payload(); + let (bytes, _) = payload.full_state_bytes_timed().expect("bytes"); + assert_eq!(bytes.as_ref(), &x[..]); + } + + #[test] + fn eviction_moves_shared_bytes_into_the_survivors_charge() { + // The short entry owns A,B; the long entry shares A,B and owns C. + // The long entry is charged only C. When the short entry is + // evicted, the pooled bytes do not change — so the survivor's + // charge must grow to the full pool, never drift below it. + let segment_len = 16u64; + let total = segment_len * 3; + let tier = L2Tier::new(1 << 20); + let short = key("ns", &[1]); + let long = key("ns", &[2]); + let (w, digest) = wire(total as usize, 70); + let short_len = segment_len * 2; + tier.admit( + short.clone(), + 2, + segment_digest(&w[..short_len as usize]), + &w[..short_len as usize], + manifest_shaped_mirror(&w[..short_len as usize], segment_len), + L2Origin::FromL3, + ) + .expect("short admitted"); + tier.admit( + long.clone(), + 3, + digest, + &w, + manifest_shaped_mirror(&w, segment_len), + L2Origin::FromL3, + ) + .expect("long admitted"); + assert_eq!( + tier.peek(&long).expect("long peeked").distinct_bytes, + total, + "before the eviction the long entry owns the shared pool outright \ + (lowest-key owner), so its charge covers the full physical pool" + ); + let removed = tier.remove(&short).expect("short present"); + assert_eq!( + removed.freed_bytes, 0, + "every short-entry segment stays in the pool under the long entry" + ); + assert_eq!( + removed.retained_bytes, short_len, + "retained bytes come from actual pool references" + ); + let stats = tier.stats(); + assert_eq!(stats.bytes, total); + assert_eq!( + tier.peek(&long).expect("long peeked").distinct_bytes, + total, + "the survivor's charge must cover the physical pool it now owns" + ); + // Aggregate invariant: the sum of all charges equals the pool. + let sum_of_charges = tier.peek(&long).expect("long").distinct_bytes; + assert_eq!(sum_of_charges, stats.bytes); + } + + #[test] + fn repeated_same_digest_segments_report_retained_bytes_from_the_pool() { + // One 16-byte segment laid out twice: the pool holds 16 bytes, the + // logical wire is 32. Removing the only entry frees 16 and retains + // nothing — `logical − freed` would have overstated retention. + let tier = L2Tier::new(1 << 20); + let k = key("ns", &[3]); + let (seg, _) = wire(16, 80); + let wire_bytes: Vec = [seg.clone(), seg.clone()].concat(); + let digest = segment_digest(&wire_bytes); + let mirror = ExactStatePayloadMirror::FullState { + layout: L2Layout { + payload_kind: ExactStatePayloadKind::FullState, + total_bytes: 32, + kv_bytes: 32, + recurrent_bytes: 0, + segments: vec![ + (segment_digest(&seg), 0..16), + (segment_digest(&seg), 16..32), + ], + }, + }; + tier.admit(k.clone(), 2, digest, &wire_bytes, mirror, L2Origin::FromL3) + .expect("admitted"); + let stats = tier.stats(); + assert_eq!(stats.bytes, 16, "the pool holds the segment once"); + assert_eq!( + tier.peek(&k).expect("peeked").distinct_bytes, + 16, + "the charge is the distinct pool bytes, not the logical wire" + ); + let removed = tier.remove(&k).expect("present"); + assert_eq!(removed.freed_bytes, 16); + assert_eq!(removed.retained_bytes, 0, "an emptied pool retains nothing"); + assert_eq!(tier.stats().bytes, 0); + assert_eq!(tier.stats().segments, 0); + } + + #[test] + fn removing_a_zero_charge_sharer_never_inflates_survivor_charges() { + // A, B, C all share segment X; only the lowest-key entry is ever + // charged for X. Removing the other sharers — whatever their key + // order — must leave the survivor's charge at exactly X's size, + // never 2x or 3x the physical pool. + let tier = L2Tier::new(1 << 20); + let ka = key("ns", &[1]); + let kb = key("ns", &[2]); + let kc = key("ns", &[3]); + let (x, dx) = wire(64, 90); + for k in [&ka, &kb, &kc] { + tier.admit( + k.clone(), + 1, + dx.clone(), + &x, + single_segment_mirror(&x), + L2Origin::FromL3, + ) + .expect("admitted"); + } + // After each removal, no surviving charge may exceed the pool. + let assert_charges_bounded = |tier: &L2Tier| { + let stats = tier.stats(); + for k in [ka.clone(), kb.clone(), kc.clone()] { + if let Some(peeked) = tier.peek(&k) { + assert!( + peeked.distinct_bytes <= stats.bytes, + "no charge may exceed the physical pool" + ); + } + } + stats + }; + // Remove the two zero-charge sharers in both orders. + tier.remove(&kb).expect("B removed"); + let stats = assert_charges_bounded(&tier); + assert_eq!(stats.bytes, 64); + tier.remove(&kc).expect("C removed"); + let stats = assert_charges_bounded(&tier); + assert_eq!(stats.bytes, 64); + // The remaining A owns X exactly once. + assert_eq!( + tier.peek(&ka).expect("A peeked").distinct_bytes, + 64, + "A's charge is X once, never the double- or triple-counted sum" + ); + assert_eq!(tier.stats().bytes, 64); + } + + #[test] + fn two_sharers_charge_moves_deterministically_to_the_lowest_key() { + let tier = L2Tier::new(1 << 20); + let ka = key("ns", &[10]); + let kb = key("ns", &[11]); + let (x, dx) = wire(48, 91); + for k in [&ka, &kb] { + tier.admit( + k.clone(), + 1, + dx.clone(), + &x, + single_segment_mirror(&x), + L2Origin::FromL3, + ) + .expect("admitted"); + } + // Exactly one of the two is charged (the lowest key), the other + // carries zero. + let charged_a = tier.peek(&ka).expect("A").distinct_bytes; + let charged_b = tier.peek(&kb).expect("B").distinct_bytes; + assert_eq!( + charged_a + charged_b, + 48, + "the sum of charges equals the physical pool" + ); + assert!(charged_a == 48 || charged_b == 48, "one owns X"); + assert!(charged_a == 0 || charged_b == 0, "the other pays nothing"); + // Remove whichever one that is not the owner: the charge sum is + // unchanged. + let (owner, zero) = if charged_a == 48 { + (ka.clone(), kb.clone()) + } else { + (kb.clone(), ka.clone()) + }; + tier.remove(&zero).expect("zero-charge sharer removed"); + assert_eq!(tier.stats().bytes, 48); + assert_eq!( + tier.peek(&owner).expect("owner peeked").distinct_bytes, + 48, + "the owner's charge is unchanged by a zero-charge removal" + ); + } + + #[test] + fn repeated_digest_with_survivor_reports_retained_once() { + // The removed entry references X twice in its layout; a survivor + // references X once. The pool retains X exactly once, so + // `retained_bytes` must be size(X) — never the double count the + // raw layout iteration would produce. + let tier = L2Tier::new(1 << 20); + let k_removed = key("ns", &[20]); + let k_survivor = key("ns", &[21]); + let (x, _) = wire(32, 95); + let x_digest = segment_digest(&x); + + // Survivor: single-segment mirror over X. + tier.admit( + k_survivor.clone(), + 1, + x_digest.clone(), + &x, + single_segment_mirror(&x), + L2Origin::FromL3, + ) + .expect("survivor admitted"); + + // Removed entry: X laid out twice (X X). + let wire_bytes: Vec = [x.clone(), x.clone()].concat(); + let digest = segment_digest(&wire_bytes); + let mirror = ExactStatePayloadMirror::FullState { + layout: L2Layout { + payload_kind: ExactStatePayloadKind::FullState, + total_bytes: 64, + kv_bytes: 64, + recurrent_bytes: 0, + segments: vec![(x_digest.clone(), 0..32), (x_digest, 32..64)], + }, + }; + tier.admit( + k_removed.clone(), + 2, + digest, + &wire_bytes, + mirror, + L2Origin::FromL3, + ) + .expect("removed entry admitted"); + let stats = tier.stats(); + assert_eq!(stats.bytes, 32, "the pool holds X once"); + + let removed = tier.remove(&k_removed).expect("removed entry present"); + assert_eq!( + removed.retained_bytes, 32, + "retained is size(X) once, not the twice-referenced 64" + ); + assert_eq!(removed.freed_bytes, 0, "the survivor keeps X"); + assert_eq!(tier.stats().bytes, 32); + assert_eq!(tier.stats().segments, 1); + let hit = tier.get(&k_survivor).expect("survivor intact"); + let payload = hit.to_payload(); + let (bytes, _) = payload.full_state_bytes_timed().expect("bytes"); + assert_eq!(bytes.as_ref(), &x[..]); + } + + #[test] + fn same_digest_different_bytes_is_rejected_unless_replacing_same_key() { + // A second wire claiming an existing segment digest with + // different-length content must not steal or replace the live + // pool handle. + let tier = L2Tier::new(1 << 20); + let (w1, _) = wire(32, 41); + let k1 = key("ns", &[1]); + tier.admit( + k1.clone(), + 2, + segment_digest(&w1), + &w1, + single_segment_mirror(&w1), + L2Origin::FromL3, + ) + .expect("first entry admitted"); + + // Forge a fake digest; the wire integrity check would reject a + // mismatched whole-wire digest, so claim the real segment digest + // of another wire as the *layout segment* digest instead. Build a + // second wire whose layout claims k1's segment digest. + let (w2, d2) = wire(48, 42); + let stolen = segment_digest(&w1); + let mirror = ExactStatePayloadMirror::FullState { + layout: L2Layout { + payload_kind: ExactStatePayloadKind::FullState, + total_bytes: 48, + kv_bytes: 48, + recurrent_bytes: 0, + segments: vec![(stolen, 0..24), (segment_digest(&w2[24..]), 24..48)], + }, + }; + let k2 = key("ns", &[2]); + let err = tier + .admit(k2, 3, d2, &w2, mirror, L2Origin::Direct) + .expect_err( + "a layout that claims another entry's digest with different bytes \ + must be refused", + ); + assert!( + matches!(err, L2InsertRefusal::SegmentDigestMismatch { .. }), + "expected segment digest mismatch, got: {err:?}" + ); + // The first entry's handle is untouched and still serves its bytes. + let hit = tier.get(&k1).expect("first entry intact"); + let payload = hit.to_payload(); + let (bytes, _) = payload.full_state_bytes_timed().expect("bytes"); + assert_eq!(bytes.as_ref(), &w1[..]); + } + + #[test] + fn get_with_missing_segment_handle_is_a_cold_miss_without_recency() { + let tier = L2Tier::new(1 << 20); + let k = key("ns", &[5]); + let (w, d) = wire(64, 51); + tier.admit( + k.clone(), + 4, + d, + &w, + manifest_shaped_mirror(&w, 16), + L2Origin::FromL3, + ) + .expect("admitted"); + + // Simulate corruption: drop one handle directly out of the pool + // (test-only access through the public remove on a scratch entry + // would also free it, but here we remove the pool entry via the + // tier's own release path by admitting an exclusive same-digest + // layout is impossible — so exercise via a second tier sharing + // nothing is not needed; directly verify the downgrade path by + // clearing the pool). + tier.clear_pool_for_test(); + + let hit = tier.get(&k); + assert!(hit.is_none(), "missing handles must downgrade to a miss"); + let stats = tier.stats(); + assert_eq!(stats.misses, 1, "the miss counter must move"); + assert_eq!(stats.hits, 0); + // The corrupt entry is removed: the next get is also a miss, not a + // partial hit, and no panic occurs. + assert!(tier.get(&k).is_none()); + assert_eq!(tier.stats().misses, 2); + assert!( + tier.peek(&k).is_none(), + "corrupt entry must be dropped, not left peekable" + ); + } + + #[test] + fn from_manifest_rejects_wrong_segment_index_and_offset() { + let (w, digest) = wire(32, 61); + let base = |index: u32, offset: u64, digest: String| HandoffManifest { + version: MANIFEST_VERSION, + model_identity: "m".to_string(), + state_identity: "s".to_string(), + payload_kind: "full-state".to_string(), + total_bytes: 32, + payload_digest: digest.clone(), + segments: vec![HandoffSegmentRef { + index, + offset, + bytes: 32, + digest: segment_digest(&w), + meta_json: None, + }], + kv_bytes: 32, + recurrent_bytes: 0, + kv_desc_json: None, + token_count: 1, + continuation_token: 0, + expected_tokens: Vec::new(), + }; + let manifest = base(1, 0, digest.clone()); + let err = ExactStatePayloadMirror::from_manifest(&manifest) + .expect_err("wrong segment index must be refused"); + assert!(matches!(err, L2InsertRefusal::MalformedManifest(_))); + + let manifest = base(0, 8, digest); + let err = ExactStatePayloadMirror::from_manifest(&manifest) + .expect_err("wrong segment offset must be refused"); + assert!(matches!(err, L2InsertRefusal::MalformedManifest(_))); + } +} diff --git a/crates/skippy-cache/src/lib.rs b/crates/skippy-cache/src/lib.rs index 724658aaab..01b59e3b2a 100644 --- a/crates/skippy-cache/src/lib.rs +++ b/crates/skippy-cache/src/lib.rs @@ -1,6 +1,7 @@ pub mod config; pub mod fsinfo; pub mod identity; +pub mod l2; pub mod l3; pub mod manager; pub mod payload; @@ -16,6 +17,10 @@ pub use identity::{ numerical_model_identity_for_stage, prefix_hash, prefix_hash_with_namespace, prefix_identity, prefix_identity_with_namespace, prefix_namespace_hash, }; +pub use l2::{ + ExactStatePayloadMirror, L2Eviction, L2Hit, L2InsertRefusal, L2Origin, L2Peek, L2Stats, L2Tier, + l2_cache_key, +}; pub use l3::{ GeometryBlock, GeometryKind, HandoffManifest, HandoffSegmentRef, HandoffSegmentStore, MANIFEST_VERSION, ManifestPin, PayloadGeometry, Reservation, SegmentHold, SegmentPut, diff --git a/crates/skippy-cache/src/payload/bytes.rs b/crates/skippy-cache/src/payload/bytes.rs index 063684bc11..a3201c1b32 100644 --- a/crates/skippy-cache/src/payload/bytes.rs +++ b/crates/skippy-cache/src/payload/bytes.rs @@ -25,7 +25,7 @@ pub(super) enum CacheBytesRepr { } #[derive(Debug, Clone)] -pub(super) struct CacheBlockRef { +pub(crate) struct CacheBlockRef { pub(super) hash: String, /// Shared indirection lets eviction materialize a surviving deduped block /// before releasing its former contiguous backing allocation. @@ -71,6 +71,45 @@ impl CacheBytes { } } + /// Crate-internal: build a block-backed view over shared immutable + /// storages without copying bytes. Each item is `(hash, storage, range)`; + /// `hash` is bookkeeping identity for the block (the L2 tier uses the + /// segment digest). When exactly one item covers its whole storage the + /// view borrows it contiguously; otherwise reads reconstruct in block + /// order. + pub(crate) fn from_shared_blocks( + len: u64, + blocks: impl IntoIterator>, Range)>, + ) -> Self { + let refs: Vec = blocks + .into_iter() + .map(|(hash, storage, range)| { + CacheBlockRef::new( + hash, + Arc::new(RwLock::new(CacheBlockBytes::new(storage, range))), + ) + }) + .collect(); + let contiguous = match refs.as_slice() { + [single] => { + let bytes = single + .bytes + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner); + (bytes.range.start == 0 && bytes.range.end == bytes.storage.len()) + .then(|| Arc::clone(&bytes.storage)) + } + _ => None, + }; + Self { + len, + repr: CacheBytesRepr::Blocks { + blocks: refs.into(), + contiguous, + }, + } + } + pub(super) fn blocks( len: u64, blocks: Vec, diff --git a/tools/xtask/data/console_print_allowlist.json b/tools/xtask/data/console_print_allowlist.json index e2567d009b..ed72273f89 100644 --- a/tools/xtask/data/console_print_allowlist.json +++ b/tools/xtask/data/console_print_allowlist.json @@ -3665,6 +3665,12 @@ "macro_name": "println!" } ], + "crates/skippy-bench/src/l2_tier.rs": [ + { + "line": 254, + "macro_name": "println!" + } + ], "crates/skippy-bench/src/local_single.rs": [ { "line": 196, @@ -3691,11 +3697,11 @@ ], "crates/skippy-bench/src/main.rs": [ { - "line": 34, + "line": 35, "macro_name": "eprintln!" }, { - "line": 42, + "line": 43, "macro_name": "eprintln!" } ],