diff --git a/CHANGELOG.md b/CHANGELOG.md index bead886cb..de96ca6ed 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,46 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added — memory + tier accounting spine (kernel M2 stage 2 / K4) + +`resident_bytes()` is now implemented by every storage plane, and the +elastic memory budget's used-term (`ShardDatabases::recompute_elastic_budget`, +GAP-1/PR #170) folds in all of them — previously kv+vector only. + +- `TextStore`/`TextIndex::resident_bytes()`: posting lists, term + dictionaries, FST fuzzy/prefix sidecars, per-document bookkeeping maps, + and TAG/NUMERIC secondary indexes. FTS memory was hard-coded 0 + everywhere it was published (elastic budget, MEMORY DOCTOR, Prometheus) + until this change. **Data-size-independent incremental accumulator** + (the publish-site read sums cached per-structure totals — O(schema + field count), bounded by `FT.CREATE` definitions, never corpus size; + same contract as `ColdIndex`/graph below) — an initial version was an + O(doc-count + + vocabulary) full-recompute walk invoked unconditionally every 100ms from + the shard eviction tick regardless of `maxmemory`, measured 6.4–21.3ms/call + at 50K–200K docs (>20% of the tick budget, recurring P99 spikes for every + command on that shard). Fixed before merge: `PostingStore`/ + `TermDictionary`/`TextIndex` each carry a cached total maintained + incrementally at every mutation site (index/delete/upsert/TAG/NUMERIC + update/FST rebuild), verified against a `#[cfg(test)]` ground-truth + full-walk after a mixed mutation sequence. +- `ColdIndex::resident_bytes()` (KV disk-offload bookkeeping): an O(1) + incremental accumulator (not a per-tick walk — sized for G2's "10x RAM" + scale target), charged into the shard's published KV memory. +- Graph resident bytes (already computed) now also feed the elastic + budget's used-term, not just the observability atomic. +- `moon_memory_bytes{kind="text"}` Prometheus gauge + `Text (FTS):` line + in `MEMORY DOCTOR`. Also fixes a pre-existing gap where + `moon_memory_bytes{kind="lua_scripts"}` was emitted but never primed. +- New `src/storage/tier.rs`: `ResidencyTier` (Hot/WarmReloadable/ColdStub) + + `TierPolicy` trait skeleton — types only, no plane adoption in this + milestone (that is M4). + +No eviction policy semantics changed: this widens what the existing +donor/hot formula sees, not how it decides. Verified against +`eviction_parity`/`eviction_parity_hash_disk_offload` (shards 1 and 4, +including the disk-offload `ColdIndex` path) with no behavior change. + ### Fixed — Windows CI: `replication_planes` used un-gated `libc::kill` `tests/replication_planes.rs`'s `sigkill` helper called `libc::kill` @@ -13,6 +53,7 @@ unconditionally — `libc` is not linked on Windows, so the `Check (Windows)` job failed to compile the suite on every `main` push since the Wave A merge. Now cfg-gated exactly like `tests/aof_multidb_kill9.rs` (`Child::kill` on non-unix). Test-only. + ### Fixed — graph CSR segments and text/vector sidecars were not crash-durable (K3, storage-kernel M2 stage 1) Extracted the vector engine's Stack-B temp+fsync+rename+dir-fsync diff --git a/src/admin/metrics_setup.rs b/src/admin/metrics_setup.rs index abaa4198e..deea4acf8 100644 --- a/src/admin/metrics_setup.rs +++ b/src/admin/metrics_setup.rs @@ -239,22 +239,31 @@ pub fn init_metrics( } } -/// Prime all 7 `moon_memory_bytes{kind=...}` series with `0.0` so they +/// Prime every `moon_memory_bytes{kind=...}` series with `0.0` so they /// appear in `/metrics` output from the first scrape, even when subsystems /// are feature-gated off or not yet initialized. /// /// NOTE: This scrape path intentionally does NOT call `mallctl("epoch")`. /// See the documented jemalloc leak at the `get_rss_bytes()` doc-comment /// (~1 MB / 20 s growth). `allocator_overhead` is computed as -/// `max(0, RSS − sum(other 6))` — the same formula MEMORY DOCTOR uses. +/// `max(0, RSS − sum(other kinds))` — the same formula MEMORY DOCTOR uses. fn prime_moon_memory_bytes() { for kind in [ "dashtable", "hnsw", + // K4 (kernel-m2-brief-2026-07-12 stage 2): text (FTS) resident + // bytes -- previously hard-coded 0 at the publish site, so this + // series existed nowhere until now. + "text", "csr", "wal", "sealed", "replication_backlog", + // Pre-existing gap fixed alongside the "text" addition above: + // `update_moon_memory_bytes` has emitted this kind since C4 + // (wave-5 hygiene), but it was never primed, so it silently + // didn't appear in `/metrics` until the first 15s update tick. + "lua_scripts", "allocator_overhead", ] { gauge!("moon_memory_bytes", "kind" => kind).set(0.0); @@ -1390,6 +1399,9 @@ fn update_moon_memory_bytes() { let mut dashtable: usize = 0; let mut hnsw: usize = 0; let sealed: usize = 0; // combined into hnsw from vector atomic (C5) + // K4 (kernel-m2-brief-2026-07-12 stage 2): text (FTS) resident bytes, + // previously hard-coded 0 at the publish site. + let mut text: usize = 0; let mut csr: usize = 0; let wal: usize = 0; // WalWriterV3 is stack-owned; not reachable here let mut backlog: usize = 0; @@ -1400,10 +1412,11 @@ fn update_moon_memory_bytes() { // C5 / M4: `read_memory_sum()` replaces per-shard `read_db(…)` locks. dashtable = shard_dbs.read_memory_sum(); - // Store memory: sum published per-shard vector/graph atomics. + // Store memory: sum published per-shard vector/text/graph atomics. // Values are refreshed by each shard's 100ms tick (publish_store_memory). for mem in shard_dbs.store_memory_per_shard.iter() { hnsw += mem.vector.load(Ordering::Relaxed); + text += mem.text.load(Ordering::Relaxed); // graph is cfg-gated at publish time; the atomic is always present. csr += mem.graph.load(Ordering::Relaxed); // C4 (wave-5 hygiene): Lua script-cache byte estimate. @@ -1418,11 +1431,12 @@ fn update_moon_memory_bytes() { } } - let other_sum = dashtable + hnsw + csr + wal + sealed + backlog + lua; + let other_sum = dashtable + hnsw + text + csr + wal + sealed + backlog + lua; let alloc_overhead = rss.saturating_sub(other_sum); gauge!("moon_memory_bytes", "kind" => "dashtable").set(dashtable as f64); gauge!("moon_memory_bytes", "kind" => "hnsw").set(hnsw as f64); + gauge!("moon_memory_bytes", "kind" => "text").set(text as f64); gauge!("moon_memory_bytes", "kind" => "csr").set(csr as f64); gauge!("moon_memory_bytes", "kind" => "wal").set(wal as f64); gauge!("moon_memory_bytes", "kind" => "sealed").set(sealed as f64); diff --git a/src/command/server_admin.rs b/src/command/server_admin.rs index 7ee13a510..ad06db777 100644 --- a/src/command/server_admin.rs +++ b/src/command/server_admin.rs @@ -411,27 +411,34 @@ fn memory_doctor() -> Frame { let csr_bytes: usize; let wal_bytes: usize = 0; let lua_bytes: usize; + // K4 (kernel-m2-brief-2026-07-12 stage 2): text (FTS) resident bytes, + // previously hard-coded 0 at the publish site. + let text_bytes: usize; if let Some(shard_dbs) = crate::admin::metrics_setup::get_global_shard_databases() { // KV memory: sum of per-shard published atomics. Lock-free. dashtable_bytes = shard_dbs.read_memory_sum(); - // Store memory: sum published per-shard vector/graph atomics. + // Store memory: sum published per-shard vector/text/graph atomics. let mut vec_total = 0usize; + let mut text_total = 0usize; let mut csr_total = 0usize; let mut lua_total = 0usize; for mem in shard_dbs.store_memory_per_shard.iter() { vec_total += mem.vector.load(Ordering::Relaxed); + text_total += mem.text.load(Ordering::Relaxed); csr_total += mem.graph.load(Ordering::Relaxed); // C4 (wave-5 hygiene): Lua script-cache byte estimate. lua_total += mem.lua.load(Ordering::Relaxed); } hnsw_bytes = vec_total; + text_bytes = text_total; csr_bytes = csr_total; lua_bytes = lua_total; } else { dashtable_bytes = 0; hnsw_bytes = 0; + text_bytes = 0; csr_bytes = 0; lua_bytes = 0; } @@ -448,6 +455,7 @@ fn memory_doctor() -> Frame { // ── Computed overhead ──────────────────────────────────────────────── let tracked_sum = dashtable_bytes + hnsw_bytes + + text_bytes + csr_bytes + wal_bytes + sealed_bytes @@ -469,6 +477,8 @@ fn memory_doctor() -> Frame { "DashTable dominates RSS (>50%). Consider increasing --initial-keyspace-hint to reduce segment splits." } else if hnsw_bytes > half_rss { "HNSW (vector) dominates RSS (>50%). Consider compacting (FT.COMPACT) or reducing ef_construction." + } else if text_bytes > half_rss { + "Text (FTS) dominates RSS (>50%). Consider FT.COMPACT to build FST sidecars, or reviewing indexed field cardinality." } else if csr_bytes > half_rss { "CSR (graph) dominates RSS (>50%). Review graph index sizes." } else if allocator_overhead > half_rss { @@ -502,6 +512,12 @@ fn memory_doctor() -> Frame { humanize_bytes(hnsw_bytes), pct(hnsw_bytes, rss) ); + let _ = writeln!( + out, + " Text (FTS): {} ({:.1}%)", + humanize_bytes(text_bytes), + pct(text_bytes, rss) + ); let _ = writeln!( out, " CSR (graph): {} ({:.1}%)", diff --git a/src/shard/persistence_tick.rs b/src/shard/persistence_tick.rs index b3e5666b4..ec71eace1 100644 --- a/src/shard/persistence_tick.rs +++ b/src/shard/persistence_tick.rs @@ -325,7 +325,14 @@ pub(crate) fn run_eviction_tick( s.store_memory .vector .store(mutable + immutable, Ordering::Relaxed); - s.store_memory.text.store(0, Ordering::Relaxed); // TextStore has no aggregate API yet + // K4 (kernel-m2-brief-2026-07-12 stage 2): TextStore now has a real + // resident_bytes() aggregate (posting lists, term dicts, FST + // sidecars, TAG/NUMERIC indexes) -- this was hard-coded 0, making + // FTS memory invisible to the elastic budget, MEMORY DOCTOR, and + // Prometheus. + s.store_memory + .text + .store(s.text_store.resident_bytes(), Ordering::Relaxed); #[cfg(feature = "graph")] { let graph_bytes = s.graph_store.resident_bytes(); @@ -351,10 +358,22 @@ pub(crate) fn run_eviction_tick( // an O(1) accumulator read). Published unconditionally: MEMORY DOCTOR // and the Prometheus KV gauge read this atomic even when maxmemory is // unlimited — gating it on maxmemory > 0 left them at a permanent 0. + // + // K4: also charge each db's ColdIndex (disk-offload bookkeeping RAM + // -- see storage::tiered::cold_index::ColdIndex::resident_bytes doc + // comment). This is a per-db O(1) accumulator read, same complexity + // class as estimated_memory() itself, so folding it in here does not + // change this tick's cost -- and it is intentionally NOT folded into + // Database::estimated_memory()/resident_bytes() themselves, which + // stay untouched O(1) hot-path reads for the per-write eviction + // pre-gate (inline_write_can_skip_eviction / try_evict_if_needed). let used = crate::shard::slice::with_shard(|s| { s.databases .iter() - .map(|db| db.estimated_memory()) + .map(|db| { + db.estimated_memory() + + db.cold_index.as_ref().map_or(0, |ci| ci.resident_bytes()) + }) .sum::() }); shard_databases.publish_memory(shard_id, used); diff --git a/src/shard/shared_databases.rs b/src/shard/shared_databases.rs index 1cf0ce765..125bb1371 100644 --- a/src/shard/shared_databases.rs +++ b/src/shard/shared_databases.rs @@ -251,6 +251,14 @@ impl ShardDatabases { // siblings while its true footprint was already over base, and the // pressure cascade then compared a vector-INCLUSIVE used against a // budget inflated by that donation. Two Relaxed loads per shard. + // + // K4 (kernel-m2-brief-2026-07-12 stage 2): extend the same fix to + // text (FTS) and graph resident bytes — a text- or graph-heavy/ + // KV-light shard was exactly as misclassifiable as an idle donor as + // the vector case above (same mechanism, different plane). KV's own + // ColdIndex bytes are already folded into `memory_per_shard` at the + // publish site (persistence_tick.rs), so they flow through here for + // free. Four Relaxed loads per shard. let used: SmallVec<[usize; 16]> = self .memory_per_shard .iter() @@ -258,6 +266,8 @@ impl ShardDatabases { .map(|(kv, store)| { kv.load(Ordering::Relaxed) .saturating_add(store.vector.load(Ordering::Relaxed)) + .saturating_add(store.text.load(Ordering::Relaxed)) + .saturating_add(store.graph.load(Ordering::Relaxed)) }) .collect(); let budget = crate::storage::eviction::compute_elastic_budget(shard_id, base, &used); @@ -1411,6 +1421,53 @@ mod tests { assert_eq!(shared.recompute_elastic_budget(2, &rt), 100); } + /// K4 (kernel-m2-brief-2026-07-12 stage 2): the same donor/hot + /// misclassification the vector fix (A4) addressed above applies + /// identically to text (FTS) and graph resident bytes -- the elastic + /// budget's used-term counted kv+vector ONLY until this commit. Mirrors + /// `recompute_elastic_budget_vector_heavy_shard_not_donor` exactly, but + /// splits the "hidden" RAM across text on one shard and graph on + /// another, so both plane wires are exercised in one test. + #[test] + fn recompute_elastic_budget_text_and_graph_heavy_shards_not_donors() { + let shared = new_shared(4, 1); + let rt = rt_config(400, 4); // base = 100 per shard + + shared.publish_memory(0, 120); // hot + shared.publish_memory(1, 10); // KV-light... + shared.store_memory_per_shard[1] + .text + .store(200, Ordering::Relaxed); // ...but 200 of FTS RAM + shared.publish_memory(2, 10); // KV-light... + shared.store_memory_per_shard[2] + .graph + .store(200, Ordering::Relaxed); // ...but 200 of graph RAM + shared.publish_memory(3, 10); + + // Blind-to-text/graph math: shards 1 and 2 look idle (10 < 100) and + // each donate 90 — surplus 90+90+90=270 to the one hot shard ⇒ 370. + // Aware: shards 1 and 2 are each truly at 210 > base — HOT, not + // donors. Only shard 3 (true 10) donates 90. Three hot shards split + // the 90 surplus ⇒ 100 + 30 = 130 each. + assert_eq!( + shared.recompute_elastic_budget(0, &rt), + 130, + "text/graph-heavy shards must not be classified as idle donors" + ); + assert_eq!( + shared.recompute_elastic_budget(1, &rt), + 130, + "the text-heavy shard itself is hot and shares the pool" + ); + assert_eq!( + shared.recompute_elastic_budget(2, &rt), + 130, + "the graph-heavy shard itself is hot and shares the pool" + ); + // The true idle shard keeps base. + assert_eq!(shared.recompute_elastic_budget(3, &rt), 100); + } + #[test] fn recompute_elastic_budget_correct_beyond_smallvec_inline_capacity() { // The per-call `used` snapshot is a `SmallVec<[usize; 16]>` — pin diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 59c88eab3..08d2b9a5d 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -15,6 +15,7 @@ pub mod hotkey; pub mod intset; pub mod listpack; pub mod stream; +pub mod tier; pub mod tiered; pub use db::Database; diff --git a/src/storage/tier.rs b/src/storage/tier.rs new file mode 100644 index 000000000..bd8156ce5 --- /dev/null +++ b/src/storage/tier.rs @@ -0,0 +1,154 @@ +//! Shared tier-ladder vocabulary for cross-plane memory residency. +//! +//! K4 (kernel-m2-brief-2026-07-12 stage 2, item 5): **types only** in this +//! milestone (M2). No plane adopts [`ResidencyTier`] or [`TierPolicy`] yet +//! -- this module exists to give a name to a concept vector already +//! implements ad hoc (`src/vector/segment/holder.rs`'s `mutable` / +//! `immutable` (HOT) vs `warm` fields, `WarmSearchSegment`'s HOT->WARM +//! transition), so KV (disk-offload `ColdIndex`), graph (CSR segments), +//! and FTS can adopt the same vocabulary incrementally in M4 instead of +//! each inventing a bespoke tier concept. +//! +//! Non-goal (per the K4 brief): forcing one eviction policy on all planes. +//! Each plane keeps its own trigger/threshold logic (idle timeout, byte +//! cap, memory pressure) -- this module only standardizes the RESIDENCY +//! STATE an entry/segment can report itself in, and the query surface a +//! future shared driver (idle sweep, memory-pressure cascade) could use to +//! ask "what tier is this in" / "may this demote" without reaching into +//! plane internals. +//! +//! Adopting a plane onto this trait is explicitly OUT of scope here: doing +//! so touches each plane's compact/demote/promote call sites, which is real +//! behavior-affecting work that belongs in M4 with its own red/green tests, +//! not folded into an accounting-only milestone. + +/// A segment/entry's residency state along the shared tier ladder. +/// +/// `Hot` -> `WarmReloadable` -> `ColdStub`, in decreasing RAM cost and +/// increasing reload latency to serve a request. Not every plane +/// implements every tier: +/// +/// - KV disk-offload (`storage::tiered::cold_index::ColdIndex`) today only +/// has `Hot` (DashTable-resident) and `ColdStub` (spilled, index-only) -- +/// no reloadable middle tier. +/// - Vector immutable segments have `Hot` and `WarmReloadable` +/// (`WarmSearchSegment`, mmap-backed) but no `ColdStub`: the 2026-07-10 +/// two-stack decision deleted vector COLD/DiskANN entirely (see +/// `project_vector_two_stack_decision` in project memory). +/// - Graph CSR segments are currently `Hot`-only (no demotion path yet). +/// +/// This is exactly why the enum has three variants even though no single +/// plane uses all three today: it is the union of what every plane's +/// ladder could eventually reach, not a description of current behavior. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum ResidencyTier { + /// Fully materialized in RAM. Serving a request pays no reload cost. + Hot, + /// Partially resident (e.g. mmap-backed, quantized-codes-only, + /// index-only-with-on-disk-payload): serving a request may pay a + /// reload/decode cost, but the bookkeeping needed to KNOW what to + /// reload and where is still in RAM. + WarmReloadable, + /// A pointer/stub only. Serving a request requires a full reload from + /// durable storage before any work on this entry/segment can proceed. + ColdStub, +} + +impl ResidencyTier { + /// Whether this tier can still answer a query without a disk read + /// (`Hot`) versus needing one (`WarmReloadable` partially, `ColdStub` + /// fully). A cheap, plane-agnostic classification a shared driver could + /// use for e.g. "how much of this plane's data is disk-free right now". + #[must_use] + pub fn is_fully_resident(self) -> bool { + matches!(self, ResidencyTier::Hot) + } +} + +/// Per-plane policy hook for tier transitions. +/// +/// **Trait skeleton only (M2) -- no implementation exists yet.** Intended +/// shape for M4 adoption: a plane's segment/entry container implements this +/// so a future shared driver (idle sweep, memory-pressure cascade) can ask +/// "what tier is `id` in" / "should `id` demote" without knowing the +/// plane's internals, while the plane retains full control over its OWN +/// demote/promote mechanics -- this trait deliberately does NOT prescribe +/// *how* a demotion is carried out, only how it is queried and decided. +pub trait TierPolicy { + /// Opaque identifier for one entry/segment in this plane (e.g. a vector + /// segment id, a KV key, a graph CSR segment id). + type Id; + + /// Current residency tier of `id`, or `None` if `id` is unknown to this + /// plane's tier bookkeeping (plane doesn't track tiers for it, or it + /// does not exist). + fn tier_of(&self, id: &Self::Id) -> Option; + + /// Whether `id` is eligible to demote one step down the ladder right + /// now, per the plane's own idle/byte-cap/memory-pressure policy + /// (entirely plane-defined -- this trait imposes no threshold). + /// Returning `true` is advisory: the caller decides WHEN to actually + /// invoke the plane's demotion mechanics; this only answers "may I". + fn eligible_to_demote(&self, id: &Self::Id) -> bool; +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn is_fully_resident_true_only_for_hot() { + assert!(ResidencyTier::Hot.is_fully_resident()); + assert!(!ResidencyTier::WarmReloadable.is_fully_resident()); + assert!(!ResidencyTier::ColdStub.is_fully_resident()); + } + + #[test] + fn residency_tier_is_copy_eq_hash() { + // Compile-time + runtime smoke test that the derives hold: the + // trait skeleton above is meant to be usable as a map key / + // comparison target by a future shared driver. + use std::collections::HashSet; + let mut set = HashSet::new(); + set.insert(ResidencyTier::Hot); + set.insert(ResidencyTier::Hot); + set.insert(ResidencyTier::WarmReloadable); + assert_eq!(set.len(), 2); + } + + /// Minimal mock plane proving `TierPolicy` is implementable and usable + /// generically -- exercises the intended M4 adoption shape without + /// adopting any real plane onto it (that is explicitly out of scope + /// here; see module doc comment). + struct MockPlane { + tiers: std::collections::HashMap, + } + + impl TierPolicy for MockPlane { + type Id = u64; + + fn tier_of(&self, id: &Self::Id) -> Option { + self.tiers.get(id).copied() + } + + fn eligible_to_demote(&self, id: &Self::Id) -> bool { + matches!(self.tier_of(id), Some(ResidencyTier::Hot)) + } + } + + #[test] + fn tier_policy_skeleton_is_implementable() { + let mut tiers = std::collections::HashMap::new(); + tiers.insert(1u64, ResidencyTier::Hot); + tiers.insert(2u64, ResidencyTier::ColdStub); + let plane = MockPlane { tiers }; + + assert_eq!(plane.tier_of(&1), Some(ResidencyTier::Hot)); + assert_eq!(plane.tier_of(&2), Some(ResidencyTier::ColdStub)); + assert_eq!(plane.tier_of(&999), None); + + assert!(plane.eligible_to_demote(&1)); + assert!(!plane.eligible_to_demote(&2)); + assert!(!plane.eligible_to_demote(&999)); + } +} diff --git a/src/storage/tiered/cold_index.rs b/src/storage/tiered/cold_index.rs index c34ab8cbc..2759eb528 100644 --- a/src/storage/tiered/cold_index.rs +++ b/src/storage/tiered/cold_index.rs @@ -77,6 +77,32 @@ pub struct ColdIndex { /// sweep ([`Self::drain_pending_unlink`]). Pushed only on a zero-ref /// transition (rare), so it does not allocate on the common insert path. pending_unlink: Vec, + /// Running total of approximate resident bytes charged by [`Self::insert`] + /// / [`Self::remove`] / the sweep methods' direct removals / [`Self::clear_all`] + /// (K4 accounting spine, kernel-m2-brief-2026-07-12 stage 2). + /// + /// Deliberately an O(1) incremental accumulator, NOT an O(n) walk over + /// `map` computed at read time (the pattern used by + /// `graph::store::GraphStore::resident_bytes`, which is O(segment_count) + /// -- a much smaller bound). A cold index backing a disk-offloaded + /// dataset is exactly the structure G2 ("serve 10x RAM datasets") sizes + /// up to tens of millions of entries; an O(n) walk every 100ms shard + /// tick would regress the workload this index exists for. See + /// [`Self::resident_bytes`] for the read side. + resident_bytes: usize, +} + +/// Approximate fixed cost of one cold-index entry beyond the key bytes: the +/// `ColdLocation` value plus a `HashMap` bucket's control-byte/pointer +/// overhead. Not exact -- `hashbrown`'s SwissTable layout is an +/// implementation detail -- but monotonic, matching the approximation style +/// already used by `Database::entry_overhead` (WS6) and +/// `text::term_dict::TermDictionary::resident_bytes`. +const COLD_ENTRY_OVERHEAD: usize = std::mem::size_of::() + 48; + +#[inline] +fn cold_entry_cost(key_len: usize) -> usize { + key_len + COLD_ENTRY_OVERHEAD } impl ColdIndex { @@ -85,9 +111,19 @@ impl ColdIndex { map: HashMap::new(), file_refs: HashMap::new(), pending_unlink: Vec::new(), + resident_bytes: 0, } } + /// Approximate resident bytes of this index: O(1) read of the running + /// total maintained by every mutation site. See the field doc comment + /// on `resident_bytes` for why this is incremental rather than a + /// per-call walk. + #[inline] + pub fn resident_bytes(&self) -> usize { + self.resident_bytes + } + /// Increment a file's live-ref count. #[inline] fn ref_inc(&mut self, file_id: u64) { @@ -117,6 +153,7 @@ impl ColdIndex { /// never see such a file because no key references it anymore. pub fn insert(&mut self, key: Bytes, location: ColdLocation) { let new_file = location.file_id; + let key_len = key.len(); if let Some(old) = self.map.insert(key, location) { if old.file_id != new_file { if self.ref_dec(old.file_id) { @@ -124,9 +161,13 @@ impl ColdIndex { } self.ref_inc(new_file); } - // Same file_id (different slot/page): live-ref count is unchanged. + // Same file_id (different slot/page): live-ref count is + // unchanged. `HashMap::insert` keeps the pre-existing key on an + // overwrite (the new key argument, byte-equal, is dropped), so + // `resident_bytes` is unchanged too -- no delta to apply. } else { self.ref_inc(new_file); + self.resident_bytes += cold_entry_cost(key_len); } } @@ -140,6 +181,9 @@ impl ColdIndex { if self.ref_dec(old.file_id) { self.pending_unlink.push(old.file_id); } + self.resident_bytes = self + .resident_bytes + .saturating_sub(cold_entry_cost(key.len())); true } else { false @@ -152,6 +196,7 @@ impl ColdIndex { /// the manifest handle this method deliberately does not need. pub fn clear_all(&mut self) { self.map.clear(); + self.resident_bytes = 0; for (&file_id, _) in self.file_refs.iter() { self.pending_unlink.push(file_id); } @@ -314,6 +359,9 @@ impl ColdIndex { for key in &orphan_keys { if let Some(old) = self.map.remove(key.as_ref()) { stats.entries_reclaimed += 1; + self.resident_bytes = self + .resident_bytes + .saturating_sub(cold_entry_cost(key.len())); if self.ref_dec(old.file_id) { self.pending_unlink.push(old.file_id); } @@ -423,6 +471,9 @@ impl ColdIndex { for key in &expired_keys { if let Some(old) = self.map.remove(key.as_ref()) { stats.entries_reclaimed += 1; + self.resident_bytes = self + .resident_bytes + .saturating_sub(cold_entry_cost(key.len())); if self.ref_dec(old.file_id) { self.pending_unlink.push(old.file_id); } @@ -599,6 +650,76 @@ mod tests { assert!(idx.lookup(b"key1").is_none()); } + // ── K4 accounting spine: resident_bytes O(1) accumulator ───────────── + + #[test] + fn resident_bytes_zero_when_empty() { + assert_eq!(ColdIndex::new().resident_bytes(), 0); + } + + #[test] + fn resident_bytes_grows_on_insert_shrinks_on_remove() { + let mut idx = ColdIndex::new(); + let loc = ColdLocation { + file_id: 1, + page_idx: 0, + slot_idx: 0, + ttl_ms: None, + }; + idx.insert(Bytes::from_static(b"a_reasonably_long_key"), loc); + let after_one = idx.resident_bytes(); + assert!(after_one > 0); + + idx.insert(Bytes::from_static(b"another_key"), loc); + assert!(idx.resident_bytes() > after_one); + + idx.remove(b"another_key"); + assert_eq!(idx.resident_bytes(), after_one); + + idx.remove(b"a_reasonably_long_key"); + assert_eq!(idx.resident_bytes(), 0); + } + + #[test] + fn resident_bytes_overwrite_same_key_does_not_double_count() { + let mut idx = ColdIndex::new(); + let loc_a = ColdLocation { + file_id: 1, + page_idx: 0, + slot_idx: 0, + ttl_ms: None, + }; + let loc_b = ColdLocation { + file_id: 2, + page_idx: 0, + slot_idx: 1, + ttl_ms: None, + }; + idx.insert(Bytes::from_static(b"key1"), loc_a); + let after_first = idx.resident_bytes(); + // Overwrite the SAME key with a different location (re-eviction to a + // different file). Byte length is unchanged, so resident_bytes must + // not grow. + idx.insert(Bytes::from_static(b"key1"), loc_b); + assert_eq!(idx.resident_bytes(), after_first); + } + + #[test] + fn resident_bytes_zero_after_clear_all() { + let mut idx = ColdIndex::new(); + let loc = ColdLocation { + file_id: 1, + page_idx: 0, + slot_idx: 0, + ttl_ms: None, + }; + idx.insert(Bytes::from_static(b"key1"), loc); + idx.insert(Bytes::from_static(b"key2"), loc); + assert!(idx.resident_bytes() > 0); + idx.clear_all(); + assert_eq!(idx.resident_bytes(), 0); + } + /// Create a shard dir with a `data/` subdir and a dummy heap-NNNNNN.mpf /// file standing in for a batched multi-KV spill file. fn make_shard_with_heap(file_ids: &[u64]) -> tempfile::TempDir { @@ -649,10 +770,20 @@ mod tests { }, ); + let before_sweep = ci.resident_bytes(); + // Sweep ONLY the orphan key. ci.sweep_known_orphans(vec![Bytes::from_static(b"k_orphan")], shard_dir, None) .unwrap(); + // K4: resident_bytes must shrink by exactly the orphan's cost, via + // the direct `self.map.remove` inside `sweep_known_orphans` (the + // bypass site that does NOT go through the public `remove()`). + assert!( + ci.resident_bytes() < before_sweep, + "sweeping an orphan must shrink resident_bytes" + ); + // The co-located live key must remain resolvable AND its file present. assert!( ci.lookup(b"k_live").is_some(), @@ -777,6 +908,8 @@ mod tests { }, ); + assert!(ci.resident_bytes() > 0, "insert must charge resident_bytes"); + // Sweep strictly after expiry. The key was never read. let stats = ci .sweep_expired(2_000, shard_dir, None, MAX_EXPIRED_SWEEP_BATCH) @@ -786,6 +919,13 @@ mod tests { stats.entries_reclaimed, 1, "sweep must reclaim the never-read expired entry" ); + // K4: the direct `self.map.remove` bypass inside `sweep_expired` + // must also decrement resident_bytes. + assert_eq!( + ci.resident_bytes(), + 0, + "resident_bytes must drop to 0 once the only entry expires" + ); assert!( stats.bytes_reclaimed > 0, "sweep must reclaim the backing file's bytes (last live ref)" diff --git a/src/text/posting.rs b/src/text/posting.rs index 0da234dff..93a941d4a 100644 --- a/src/text/posting.rs +++ b/src/text/posting.rs @@ -86,6 +86,27 @@ impl PostingList { } } +/// Fixed per-term overhead charged exactly once, when a term's `PostingList` +/// entry is first created in `postings` (K4 P0 fix: this entry is kept +/// forever even after its last document is removed -- see `remove_doc`'s doc +/// comment -- so the cost is charged once and never refunded, matching that +/// contract). Approximates the `HashMap` bucket overhead +/// plus the `PostingList` struct shell (its growable contents are charged +/// separately via `POSTING_OCCURRENCE_COST`/`POSITION_COST`). +const POSTING_ENTRY_OVERHEAD: usize = 48 + std::mem::size_of::(); + +/// Fixed approximate cost of one (term, doc) occurrence: one `term_freqs` +/// `u32` slot plus an amortized per-id `RoaringBitmap` cost. A flat constant +/// -- not `RoaringBitmap::serialized_size()` -- because compressed bitmap +/// size is non-linear/non-additive across arbitrary insert/remove patterns +/// and cannot be delta-tracked in O(1); this is the same "monotonic signal, +/// not exact RSS" approximation style used by `ColdIndex`/`Database:: +/// entry_overhead` elsewhere in the accounting spine. +const POSTING_OCCURRENCE_COST: usize = 4 + 4; + +/// Fixed approximate cost of one tracked token position (`u32`). +const POSITION_COST: usize = std::mem::size_of::(); + /// Per-field inverted index storing term_id -> PostingList. pub struct PostingStore { postings: HashMap, @@ -95,6 +116,14 @@ pub struct PostingStore { /// re-index cliff (fts-upsert-incremental). Kept in sync with `postings`: `add_term_occurrence` /// records the edge on the new-doc branch; `remove_doc` erases the doc's entry. doc_terms: HashMap>, + /// K4 (P0 fix): O(1) cached total mirroring `estimated_bytes()`. + /// Maintained incrementally at every mutation site (`add_term_occurrence`, + /// `remove_doc`) instead of being recomputed by a full walk on every read + /// -- `estimated_bytes()` used to be an O(vocabulary) walk called + /// unconditionally every 100ms from the shard eviction tick, which does + /// not scale with corpus size. `estimated_bytes_ground_truth` + /// (`#[cfg(test)]`) is the walk this field must always match. + resident_bytes: usize, } impl PostingStore { @@ -103,6 +132,7 @@ impl PostingStore { Self { postings: HashMap::new(), doc_terms: HashMap::new(), + resident_bytes: 0, } } @@ -115,6 +145,7 @@ impl PostingStore { /// - `positions: Some(pos)` -- store positions; upgrades a no-position list to have positions /// - `positions: None` -- don't track positions for this occurrence; keeps existing positions if any pub fn add_term_occurrence(&mut self, term_id: u32, doc_id: u32, positions: Option>) { + let is_new_term = !self.postings.contains_key(&term_id); let posting = self.postings.entry(term_id).or_insert_with(|| { if positions.is_some() { PostingList::new_with_positions() @@ -122,13 +153,18 @@ impl PostingStore { PostingList::new_without_positions() } }); + if is_new_term { + self.resident_bytes += POSTING_ENTRY_OVERHEAD; + } if posting.doc_ids.contains(doc_id) { // Existing doc: increment at the rank-aligned index. let idx = posting.rank_index(doc_id); posting.term_freqs[idx] += 1; // Append positions if provided. + let mut added_positions = 0usize; if let Some(pos) = &positions { + added_positions = pos.len(); if let Some(pos_list) = &mut posting.positions { pos_list[idx].extend_from_slice(pos); } else { @@ -138,6 +174,7 @@ impl PostingStore { posting.positions = Some(pos_list); } } + self.resident_bytes += added_positions * POSITION_COST; } else { // New document: insert into the bitmap, then insert tf/positions AT THE RANK // INDEX (not push) so term_freqs/positions stay rank-aligned with doc_ids — correct @@ -145,11 +182,16 @@ impl PostingStore { posting.doc_ids.insert(doc_id); let idx = posting.rank_index(doc_id); posting.term_freqs.insert(idx, 1); + let mut added_positions = 0usize; match (&mut posting.positions, &positions) { - (Some(pos_list), Some(pos)) => pos_list.insert(idx, pos.clone()), + (Some(pos_list), Some(pos)) => { + added_positions = pos.len(); + pos_list.insert(idx, pos.clone()); + } (Some(pos_list), None) => pos_list.insert(idx, Vec::new()), (None, Some(pos)) => { // Upgrade: track positions for all docs; this doc's positions at idx. + added_positions = pos.len(); let mut pos_list = vec![Vec::new(); posting.term_freqs.len()]; pos_list[idx] = pos.clone(); posting.positions = Some(pos_list); @@ -160,6 +202,7 @@ impl PostingStore { // time `doc_id` joins `term_id`'s posting, so no de-dup is needed. `posting`'s borrow of // `self.postings` has ended (last use above), so this disjoint-field access is sound. self.doc_terms.entry(doc_id).or_default().push(term_id); + self.resident_bytes += POSTING_OCCURRENCE_COST + added_positions * POSITION_COST; } } @@ -209,12 +252,21 @@ impl PostingStore { if idx < posting.term_freqs.len() { let old_tf = posting.term_freqs.remove(idx); posting.doc_ids.remove(doc_id); + let mut freed_positions = 0usize; if let Some(pos_list) = &mut posting.positions { if idx < pos_list.len() { + freed_positions = pos_list[idx].len(); pos_list.remove(idx); } } removed.push((term_id, old_tf)); + // K4 (P0 fix): symmetric uncharge for the occurrence + its positions + // added by `add_term_occurrence`. The entry's `POSTING_ENTRY_OVERHEAD` + // is deliberately NOT refunded here -- the `postings` map entry itself + // survives (see below), matching the never-refunded charge on creation. + self.resident_bytes = self + .resident_bytes + .saturating_sub(POSTING_OCCURRENCE_COST + freed_positions * POSITION_COST); // The `postings` HashMap entry itself is kept even when empty // (see doc comment on `remove_doc` — callers rely on // `tf`/`doc_freq` for a "term with zero live docs" staying @@ -268,16 +320,33 @@ impl PostingStore { } /// Estimated memory usage in bytes. + /// + /// K4 (P0 fix): O(1) cached read. This used to be an O(vocabulary) walk + /// calling `RoaringBitmap::serialized_size()` per term -- fine as an + /// occasional diagnostic, but this is invoked unconditionally every + /// 100ms from the shard eviction tick (`persistence_tick.rs`), where an + /// O(n) walk does not scale with corpus size. See + /// `estimated_bytes_ground_truth` (`#[cfg(test)]`) for the equivalent + /// full-walk formula this cached value must always match. + #[must_use] pub fn estimated_bytes(&self) -> usize { - let mut total = 0; + self.resident_bytes + } + + /// Ground-truth full recompute of `estimated_bytes()`, using the exact + /// same fixed-cost formula as the incremental accumulator. Test-only: + /// exists solely to assert the incremental accumulator never drifts from + /// a from-scratch recount after a mixed mutation sequence. + #[cfg(test)] + pub(crate) fn estimated_bytes_ground_truth(&self) -> usize { + let mut total = 0usize; for posting in self.postings.values() { - total += posting.doc_ids.serialized_size(); - total += posting.term_freqs.len() * 4; + total += POSTING_ENTRY_OVERHEAD; + total += posting.doc_ids.len() as usize * POSTING_OCCURRENCE_COST; if let Some(ref pos_list) = posting.positions { for positions in pos_list { - total += positions.len() * 4; + total += positions.len() * POSITION_COST; } - total += pos_list.len() * std::mem::size_of::>(); } } total @@ -368,4 +437,109 @@ mod tests { "must not shrink while the posting still has live docs" ); } + + /// K4 (P0 fix): RED-first — the O(1) incremental `resident_bytes` + /// accumulator maintained by `add_term_occurrence`/`remove_doc` must + /// never drift from a from-scratch ground-truth recompute, across a + /// mixed sequence of new terms, repeat occurrences (tf bump + position + /// append), a position-tracking upgrade, and both full and partial doc + /// removal (including the term_id-shared-across-docs case that leaves a + /// posting with live docs after another doc is removed). + #[test] + fn estimated_bytes_matches_ground_truth_after_mixed_mutations() { + let mut store = PostingStore::new(); + assert_eq!(store.estimated_bytes(), 0); + assert_eq!( + store.estimated_bytes(), + store.estimated_bytes_ground_truth() + ); + + // New terms, some with positions, some without. + store.add_term_occurrence(1, 100, Some(vec![0, 3])); + store.add_term_occurrence(2, 100, None); + store.add_term_occurrence(3, 100, Some(vec![7])); + store.add_term_occurrence(1, 101, Some(vec![1])); + assert_eq!( + store.estimated_bytes(), + store.estimated_bytes_ground_truth() + ); + + // Repeat occurrence: tf bump + position append on an existing doc. + store.add_term_occurrence(1, 100, Some(vec![5, 6])); + assert_eq!( + store.estimated_bytes(), + store.estimated_bytes_ground_truth() + ); + + // Upgrade: term 2 had no position tracking, now gets one. + store.add_term_occurrence(2, 101, Some(vec![2])); + assert_eq!( + store.estimated_bytes(), + store.estimated_bytes_ground_truth() + ); + + // Shared term across many docs. + for doc_id in 200..210u32 { + store.add_term_occurrence(3, doc_id, Some(vec![doc_id])); + } + assert_eq!( + store.estimated_bytes(), + store.estimated_bytes_ground_truth() + ); + + // Partial removal: term 3 keeps live docs after doc 205 is removed. + store.remove_doc(205); + assert_eq!( + store.estimated_bytes(), + store.estimated_bytes_ground_truth() + ); + + // Full removal of a document touching multiple terms. + store.remove_doc(100); + assert_eq!( + store.estimated_bytes(), + store.estimated_bytes_ground_truth() + ); + + // Drain every remaining document -- resident_bytes must settle back + // to the entry-overhead-only floor (never below it: entries survive + // empty per the documented contract), matching ground truth exactly. + for doc_id in [101, 200, 201, 202, 203, 204, 206, 207, 208, 209] { + store.remove_doc(doc_id); + } + assert_eq!( + store.estimated_bytes(), + store.estimated_bytes_ground_truth() + ); + assert_eq!( + store.estimated_bytes(), + 3 * POSTING_ENTRY_OVERHEAD, + "3 terms ever created, all doc occurrences drained -- only entry overhead remains" + ); + } + + /// K4 (P0 fix): `estimated_bytes()` must be a pure O(1) load with no + /// iteration in the accessor -- enforced by construction here: the + /// accessor is called on a store sized large enough that an O(n) walk + /// would be trivially detectable by any reasonable wall-clock budget, + /// paired with the source-level guarantee that the method body is a + /// single field read (see the implementation above). + #[test] + fn estimated_bytes_is_o1_not_a_walk() { + let mut store = PostingStore::new(); + for term_id in 0..5_000u32 { + for doc_id in 0..20u32 { + store.add_term_occurrence(term_id, doc_id, Some(vec![doc_id])); + } + } + let start = std::time::Instant::now(); + for _ in 0..100_000 { + std::hint::black_box(store.estimated_bytes()); + } + let elapsed = start.elapsed(); + assert!( + elapsed < std::time::Duration::from_millis(200), + "100k reads of estimated_bytes() took {elapsed:?} -- looks like a walk, not O(1)" + ); + } } diff --git a/src/text/store.rs b/src/text/store.rs index cf372cebb..2cc29b6ca 100644 --- a/src/text/store.rs +++ b/src/text/store.rs @@ -18,6 +18,35 @@ use crate::text::types::{BM25Config, TextFieldDef}; #[cfg(feature = "text-index")] use crate::text::types::{NumericFieldDef, TagFieldDef}; +// ── K4 (P0 fix) accounting constants ─────────────────────────────────────── +// +// `TextIndex::resident_bytes()` used to be an O(n) full-recompute walk over +// every posting/term/TAG/NUMERIC entry, called unconditionally every 100ms +// from the shard eviction tick (`persistence_tick.rs`) regardless of whether +// `maxmemory` is even set -- measured 6.4ms/call at 50K docs, 21.3ms at +// 200K, recurring P99 spikes for every command on that shard. These +// constants back an O(1) incremental accumulator instead (mirroring +// `ColdIndex`'s `COLD_ENTRY_OVERHEAD` pattern): fixed per-entry/per- +// occurrence approximations updated at every mutation site rather than +// walked on every read. Not exact -- `hashbrown`'s SwissTable layout and +// `RoaringBitmap`'s compressed container format are implementation details +// -- but monotonic, matching `Database::entry_overhead` (WS6) and +// `TermDictionary::resident_bytes`'s established convention. + +/// Fixed per-entry `HashMap`/`BTreeMap` bucket-overhead constant. +const MAP_ENTRY_OVERHEAD: usize = 48; +/// Fixed approximate cost of one bit set in a `RoaringBitmap`. Not +/// `RoaringBitmap::serialized_size()` -- compressed bitmap size is +/// non-linear/non-additive across arbitrary insert/remove patterns and +/// cannot be delta-tracked in O(1) (same reasoning as `PostingStore`'s +/// `POSTING_OCCURRENCE_COST`). Only used by the TAG/NUMERIC accounting +/// helpers, which are `text-index`-only. +#[cfg(feature = "text-index")] +const ROARING_BIT_APPROX_COST: usize = 3; +/// Fixed approximate cost of a brand-new (empty) `RoaringBitmap` container. +#[cfg(feature = "text-index")] +const EMPTY_BITMAP_BASE_COST: usize = 8; + /// Modifier for a query term — controls expansion strategy (D-16). /// /// Exact terms use direct HashMap TermDictionary lookup (unchanged path). @@ -149,6 +178,17 @@ pub struct TextIndex { /// report), so every text index is currently tagged db 0 — /// behavior-preserving with pre-WS5a global semantics. pub db_index: u8, + + /// K4 (P0 fix): O(1) cached total for every `resident_bytes()` + /// contributor EXCEPT `field_postings`/`field_term_dicts` (those already + /// carry their own O(field_count) cached totals). Covers per-document + /// bookkeeping (`doc_field_lengths`, `key_hash_to_doc_id`, + /// `doc_id_to_key`, MVCC LSN maps) plus TAG/NUMERIC/FST sidecar + /// contributions. Maintained incrementally by the `charge_*`/`revoke_*` + /// helpers below at every mutation site -- never recomputed by a walk. + /// See `resident_bytes_ground_truth` (`#[cfg(test)]`) for the equivalent + /// full-walk formula this field must always match. + resident_bytes_extra: usize, } impl TextIndex { @@ -210,6 +250,7 @@ impl TextIndex { #[cfg(feature = "text-index")] doc_numeric_entries: HashMap::new(), db_index: 0, + resident_bytes_extra: 0, } } @@ -234,16 +275,31 @@ impl TextIndex { bm25_config: BM25Config, ) -> Self { let mut idx = Self::new(name, key_prefixes, text_fields, bm25_config); + // K4 (P0 fix): the outer per-field entry is seeded here (empty inner + // map/btree) so `search_tag`/`search_numeric_range` on a + // never-inserted field returns empty-but-present rather than + // missing-key. `resident_bytes_ground_truth` walks `tag_indexes`/ + // `numeric_indexes` regardless of whether the inner container is + // empty, so this seeding must be charged too -- otherwise every + // schema-declared TAG/NUMERIC field undercounts by one field entry. for tag_def in &tag_fields { + let is_new = !idx.tag_indexes.contains_key(&tag_def.field_name); idx.tag_indexes .entry(tag_def.field_name.clone()) .or_default(); + if is_new { + idx.resident_bytes_extra += tag_def.field_name.len() + MAP_ENTRY_OVERHEAD; + } } idx.tag_fields = tag_fields; for num_def in &numeric_fields { + let is_new = !idx.numeric_indexes.contains_key(&num_def.field_name); idx.numeric_indexes .entry(num_def.field_name.clone()) .or_default(); + if is_new { + idx.resident_bytes_extra += num_def.field_name.len() + MAP_ENTRY_OVERHEAD; + } } idx.numeric_fields = numeric_fields; idx @@ -265,9 +321,40 @@ impl TextIndex { self.next_doc_id += 1; self.key_hash_to_doc_id.insert(key_hash, id); self.doc_id_to_key.insert(id, Bytes::copy_from_slice(key)); + self.charge_new_doc_key(key.len()); id } + /// K4 (P0 fix): charge the bookkeeping cost of a genuinely NEW document + /// entering `key_hash_to_doc_id` + `doc_id_to_key`. Callers gate this on + /// "doc_id was just newly assigned" (never on an upsert of an existing + /// key, since `HashMap::insert` on the same key is a same-size + /// overwrite) so the charge fires exactly once per doc_id -- matching + /// `remove_doc_by_doc_id`'s single unconditional uncharge, which reads + /// the exact removed key length back from `doc_id_to_key.remove()`. + fn charge_new_doc_key(&mut self, key_len: usize) { + self.resident_bytes_extra += Self::doc_key_entry_cost(key_len); + } + + /// Symmetric uncharge for [`Self::charge_new_doc_key`], called from + /// `remove_doc_by_doc_id` with the exact key length of the removed + /// entry. Sharing `doc_key_entry_cost` between charge and uncharge + /// removes the risk of the two formulas drifting apart. + fn uncharge_doc_key(&mut self, key_len: usize) { + self.resident_bytes_extra = self + .resident_bytes_extra + .saturating_sub(Self::doc_key_entry_cost(key_len)); + } + + fn doc_key_entry_cost(key_len: usize) -> usize { + std::mem::size_of::() + + std::mem::size_of::() + + MAP_ENTRY_OVERHEAD // key_hash_to_doc_id entry + + std::mem::size_of::() + + key_len + + MAP_ENTRY_OVERHEAD // doc_id_to_key entry + } + /// Return `true` if a document is visible at the requested `as_of_lsn` /// snapshot. `as_of_lsn == 0` always returns `true` (no temporal filter, /// backwards-compatible with pre-v0.1.10 callers). @@ -303,7 +390,12 @@ impl TextIndex { #[inline] pub fn set_doc_insert_lsn(&mut self, doc_id: u32, lsn: u64) { if lsn != 0 { + let is_new = !self.doc_id_to_insert_lsn.contains_key(&doc_id); self.doc_id_to_insert_lsn.insert(doc_id, lsn); + if is_new { + self.resident_bytes_extra += + std::mem::size_of::() + std::mem::size_of::() + MAP_ENTRY_OVERHEAD; + } } } @@ -368,10 +460,16 @@ impl TextIndex { id }; - // Store key mapping + // Store key mapping. K4 (P0 fix): the bookkeeping charge fires only + // on a genuinely new doc_id (`!is_upsert`) -- on upsert, both inserts + // below overwrite the SAME key_hash/doc_id with byte-size-identical + // content (same key, same doc_id), so charging again would double-count. self.key_hash_to_doc_id.insert(key_hash, doc_id); self.doc_id_to_key .insert(doc_id, Bytes::copy_from_slice(key)); + if !is_upsert { + self.charge_new_doc_key(key.len()); + } // Initialize field lengths for this document let field_count = self.text_fields.len(); @@ -418,6 +516,15 @@ impl TextIndex { self.field_stats[field_idx].total_field_length += token_count as u64; } + // K4 (P0 fix): charge only on a genuinely new doc -- on upsert this + // `insert` replaces an existing entry with a Vec of the SAME length + // (`field_count`, constant for this index), a byte-size-identical + // overwrite that must not be re-charged. + if !is_upsert { + self.resident_bytes_extra += std::mem::size_of::() + + field_count * std::mem::size_of::() + + MAP_ENTRY_OVERHEAD; + } self.doc_field_lengths.insert(doc_id, field_lengths); } @@ -598,7 +705,7 @@ impl TextIndex { { Ok(bytes) => match fst::Map::new(bytes) { Ok(map) => { - self.fst_maps[field_idx] = Some(map); + self.set_fst_map(field_idx, Some(map)); // Update high water mark: terms with id >= this were added post-compaction. self.field_term_dicts[field_idx].fst_high_water_mark = self.field_term_dicts[field_idx].next_id(); @@ -610,6 +717,26 @@ impl TextIndex { } } + /// K4 (P0 fix): replace `fst_maps[field_idx]` and re-sync its + /// `resident_bytes_extra` contribution in one step. This is a "re-sync + /// at structural event" (not a periodic walk): `fst::Map::as_fst().size()` + /// is an O(1) read of the underlying byte-slice length, and this only + /// runs at FST (re)build time (`build_fst`, `load_fst_sidecars`) -- + /// events that are already O(vocabulary) themselves, so folding in an + /// O(1)-per-field recount adds no asymptotic cost. + #[cfg(feature = "text-index")] + fn set_fst_map(&mut self, field_idx: usize, map: Option>>) { + if let Some(old) = &self.fst_maps[field_idx] { + self.resident_bytes_extra = self + .resident_bytes_extra + .saturating_sub(old.as_fst().size()); + } + if let Some(new_map) = &map { + self.resident_bytes_extra += new_map.as_fst().size(); + } + self.fst_maps[field_idx] = map; + } + /// Expand a single query term into matching term IDs via FST + HashMap fallback. /// /// Exact terms: direct TermDictionary lookup (unchanged path). @@ -859,26 +986,34 @@ impl TextIndex { } // Rebuild `doc_tag_entries[doc_id]`: keep untouched-field entries, drop touched-field entries. - let prior = self.doc_tag_entries.remove(&doc_id).unwrap_or_default(); + // K4 (P0 fix): uncharge the prior entry's cost ONLY when an entry + // actually existed (`doc_tag_entries` never stores an empty Vec -- + // see the `!next.is_empty()` guard below -- so `Some(_)` always + // means real, previously-charged content). + let prior_opt = self.doc_tag_entries.remove(&doc_id); + if let Some(prior_entries) = &prior_opt { + self.resident_bytes_extra = self + .resident_bytes_extra + .saturating_sub(Self::tag_entries_cost(prior_entries)); + } + let prior = prior_opt.unwrap_or_default(); let mut next: smallvec::SmallVec<[(Bytes, Bytes); 8]> = smallvec::SmallVec::new(); for (field, value) in prior.into_iter() { let is_touched = touched.iter().any(|f| f == &field); if is_touched { - if let Some(field_map) = self.tag_indexes.get_mut(&field) { - if let Some(bm) = field_map.get_mut(&value) { - bm.remove(doc_id); - if bm.is_empty() { - field_map.remove(&value); - } - } - } + self.tag_bitmap_revoke(&field, &value, doc_id); } else { next.push((field, value)); } } // Insert fresh entries for each touched field. - for tag_def in &self.tag_fields { + // K4 (P0 fix): clone the per-field def out of `self.tag_fields` up + // front so the loop body is free to call `&mut self` accounting + // helpers (`tag_bitmap_insert`) -- `for tag_def in &self.tag_fields` + // would hold an immutable borrow of `self` alive for the whole loop. + for i in 0..self.tag_fields.len() { + let tag_def = self.tag_fields[i].clone(); if tag_def.noindex { continue; } @@ -939,18 +1074,82 @@ impl TextIndex { } let canonical_field = tag_def.field_name.clone(); // Arc bump - let field_map = self.tag_indexes.entry(canonical_field.clone()).or_default(); for value in seen.into_iter() { - field_map.entry(value.clone()).or_default().insert(doc_id); + self.tag_bitmap_insert(&canonical_field, &value, doc_id); next.push((canonical_field.clone(), value)); } } if !next.is_empty() { + self.resident_bytes_extra += Self::tag_entries_cost(&next); self.doc_tag_entries.insert(doc_id, next); } } + /// K4 (P0 fix): insert `doc_id` into `tag_indexes[field][value]`, + /// charging the O(1) fixed-cost delta for any newly-created field/value/ + /// doc-bit. Shared by `tag_index_document`'s insert loop. + #[cfg(feature = "text-index")] + fn tag_bitmap_insert(&mut self, field: &Bytes, value: &Bytes, doc_id: u32) { + let field_is_new = !self.tag_indexes.contains_key(field); + let field_map = self.tag_indexes.entry(field.clone()).or_default(); + let value_is_new = !field_map.contains_key(value); + let bm = field_map.entry(value.clone()).or_default(); + let doc_is_new = !bm.contains(doc_id); + bm.insert(doc_id); + if doc_is_new { + self.resident_bytes_extra += ROARING_BIT_APPROX_COST; + } + if value_is_new { + self.resident_bytes_extra += value.len() + MAP_ENTRY_OVERHEAD + EMPTY_BITMAP_BASE_COST; + } + if field_is_new { + self.resident_bytes_extra += field.len() + MAP_ENTRY_OVERHEAD; + } + } + + /// K4 (P0 fix): revoke `doc_id` from `tag_indexes[field][value]`, + /// uncharging the O(1) fixed-cost delta symmetrically with + /// `tag_bitmap_insert`. The outer per-field entry is never uncharged -- + /// it is never removed from `tag_indexes` either (matches + /// `PostingStore`'s "entry survives empty" contract). Shared by both + /// `tag_index_document`'s revoke loop and `remove_doc_by_doc_id`. + #[cfg(feature = "text-index")] + fn tag_bitmap_revoke(&mut self, field: &Bytes, value: &Bytes, doc_id: u32) { + if let Some(field_map) = self.tag_indexes.get_mut(field) { + if let Some(bm) = field_map.get_mut(value) { + let was_present = bm.contains(doc_id); + bm.remove(doc_id); + if was_present { + self.resident_bytes_extra = self + .resident_bytes_extra + .saturating_sub(ROARING_BIT_APPROX_COST); + } + if bm.is_empty() { + field_map.remove(value); + self.resident_bytes_extra = self + .resident_bytes_extra + .saturating_sub(value.len() + MAP_ENTRY_OVERHEAD + EMPTY_BITMAP_BASE_COST); + } + } + } + } + + /// K4 (P0 fix): fixed-cost approximation of one `doc_tag_entries[doc_id]` + /// entry (mirrors the removed inline formula from the old + /// `resident_bytes()` walk). Pure function of the entries slice so it + /// can be called both before insert (to charge) and after remove (to + /// uncharge) without borrowing `self`. + #[cfg(feature = "text-index")] + fn tag_entries_cost(entries: &[(Bytes, Bytes)]) -> usize { + std::mem::size_of::() + + MAP_ENTRY_OVERHEAD + + entries + .iter() + .map(|(a, b)| a.len() + b.len() + 32) + .sum::() + } + /// LSN-aware wrapper around [`Self::search_field`] — post-filters the /// scored result list by MVCC visibility at `as_of_lsn`. /// @@ -1115,20 +1314,23 @@ impl TextIndex { } // Rebuild `doc_numeric_entries[doc_id]`: keep untouched-field entries, drop touched-field entries. - let prior = self.doc_numeric_entries.remove(&doc_id).unwrap_or_default(); + // K4 (P0 fix): uncharge the prior entry's cost ONLY when an entry + // actually existed (mirrors the TAG-side reasoning in + // `tag_index_document` -- `doc_numeric_entries` never stores an + // empty Vec, see the `!next.is_empty()` guard below). + let prior_opt = self.doc_numeric_entries.remove(&doc_id); + if let Some(prior_entries) = &prior_opt { + self.resident_bytes_extra = self + .resident_bytes_extra + .saturating_sub(Self::numeric_entries_cost(prior_entries)); + } + let prior = prior_opt.unwrap_or_default(); let mut next: smallvec::SmallVec<[(Bytes, ordered_float::OrderedFloat); 4]> = smallvec::SmallVec::new(); for (field, value) in prior.into_iter() { let is_touched = touched.iter().any(|f| f == &field); if is_touched { - if let Some(btree) = self.numeric_indexes.get_mut(&field) { - if let Some(bm) = btree.get_mut(&value) { - bm.remove(doc_id); - if bm.is_empty() { - btree.remove(&value); - } - } - } + self.numeric_bitmap_revoke(&field, &value, doc_id); } else { next.push((field, value)); } @@ -1174,6 +1376,7 @@ impl TextIndex { } let of = ordered_float::OrderedFloat(parsed); let canonical_field = num_def.field_name.clone(); + let field_is_new = !self.numeric_indexes.contains_key(&canonical_field); let btree = self .numeric_indexes .entry(canonical_field.clone()) @@ -1186,15 +1389,71 @@ impl TextIndex { ); continue; } - btree.entry(of).or_default().insert(doc_id); + // K4 (P0 fix): charge the O(1) fixed-cost delta for any + // newly-created field/value/doc-bit -- checked AFTER the + // cardinality cap so a dropped value is never charged. + let value_is_new = !btree.contains_key(&of); + let bm = btree.entry(of).or_default(); + let doc_is_new = !bm.contains(doc_id); + bm.insert(doc_id); + if doc_is_new { + self.resident_bytes_extra += ROARING_BIT_APPROX_COST; + } + if value_is_new { + self.resident_bytes_extra += + std::mem::size_of::() + MAP_ENTRY_OVERHEAD + EMPTY_BITMAP_BASE_COST; + } + if field_is_new { + self.resident_bytes_extra += canonical_field.len() + MAP_ENTRY_OVERHEAD; + } next.push((canonical_field, of)); } if !next.is_empty() { + self.resident_bytes_extra += Self::numeric_entries_cost(&next); self.doc_numeric_entries.insert(doc_id, next); } } + /// K4 (P0 fix): revoke `doc_id` from `numeric_indexes[field][value]`, + /// uncharging the O(1) fixed-cost delta. Mirrors `tag_bitmap_revoke`; + /// shared by `numeric_index_document`'s revoke loop and + /// `remove_doc_by_doc_id`. + #[cfg(feature = "text-index")] + fn numeric_bitmap_revoke( + &mut self, + field: &Bytes, + value: &ordered_float::OrderedFloat, + doc_id: u32, + ) { + if let Some(btree) = self.numeric_indexes.get_mut(field) { + if let Some(bm) = btree.get_mut(value) { + let was_present = bm.contains(doc_id); + bm.remove(doc_id); + if was_present { + self.resident_bytes_extra = self + .resident_bytes_extra + .saturating_sub(ROARING_BIT_APPROX_COST); + } + if bm.is_empty() { + btree.remove(value); + self.resident_bytes_extra = self.resident_bytes_extra.saturating_sub( + std::mem::size_of::() + MAP_ENTRY_OVERHEAD + EMPTY_BITMAP_BASE_COST, + ); + } + } + } + } + + /// K4 (P0 fix): fixed-cost approximation of one + /// `doc_numeric_entries[doc_id]` entry. Mirrors `tag_entries_cost`. + #[cfg(feature = "text-index")] + fn numeric_entries_cost(entries: &[(Bytes, ordered_float::OrderedFloat)]) -> usize { + std::mem::size_of::() + + MAP_ENTRY_OVERHEAD + + entries.iter().map(|(a, _)| a.len() + 16).sum::() + } + /// Resolve a NUMERIC range filter to sorted doc_ids. /// /// Uses `BTreeMap::range` — O(log N) seek + sequential bucket scan. The @@ -1274,6 +1533,161 @@ impl TextIndex { .sum() } + /// Approximate total resident bytes owned by this index: posting lists, + /// term dictionaries (BM25 path), FST sidecars (fuzzy/prefix expansion), + /// per-document bookkeeping maps (field lengths, key<->doc_id, MVCC + /// LSNs), and TAG/NUMERIC secondary indexes. + /// + /// K4 accounting spine (kernel-m2-brief-2026-07-12 stage 2): the sole + /// contributor to `TextStore::resident_bytes()`, which is folded into + /// the shard's `store_memory.text` atomic (elastic memory budget + /// used-term + MEMORY DOCTOR / Prometheus surfacing) -- previously + /// hard-coded 0. + /// + /// K4 (P0 fix): O(1) cached read -- `field_postings`/`field_term_dicts` + /// each carry their own O(field_count) cached total (already O(1) per + /// field), and everything else is folded into `resident_bytes_extra`, + /// maintained incrementally at every mutation site. This used to be an + /// O(doc-count + vocabulary) walk called unconditionally every 100ms + /// from the shard eviction tick, which does not scale with corpus size. + /// See `resident_bytes_ground_truth` (`#[cfg(test)]`) for the equivalent + /// full-walk formula this cached value must always match. + /// + /// Excluded as negligible/bounded, not doc-scaling: `AnalyzerPipeline` + /// (one stemmer + stop-word set per field, built once at FT.CREATE), + /// `FieldStats` (two scalars per field), `BM25Config` / `text_fields` / + /// `key_prefixes` / `name` (schema metadata, O(field count)). Every + /// `HashMap`/`BTreeMap` entry cost below uses the same fixed-overhead + /// approximation as `TermDictionary::resident_bytes` and + /// `graph::index::PropertyIndex::resident_bytes`'s `serialized_size()` + /// convention -- a monotonic signal for the budget, not exact RSS. + #[must_use] + pub fn resident_bytes(&self) -> usize { + let postings: usize = self + .field_postings + .iter() + .map(PostingStore::estimated_bytes) + .sum(); + let term_dicts: usize = self + .field_term_dicts + .iter() + .map(TermDictionary::resident_bytes) + .sum(); + postings + term_dicts + self.resident_bytes_extra + } + + /// Ground-truth full recompute of `resident_bytes()`, using the exact + /// same fixed-cost formulas as the incremental accumulators + /// (`PostingStore::estimated_bytes_ground_truth`, + /// `TermDictionary::resident_bytes_ground_truth`, and the TAG/NUMERIC/ + /// FST/bookkeeping formulas inlined below). Test-only: exists solely to + /// assert the incremental accumulators never drift from a from-scratch + /// recount after a mixed mutation sequence. + #[cfg(test)] + pub(crate) fn resident_bytes_ground_truth(&self) -> usize { + let postings: usize = self + .field_postings + .iter() + .map(PostingStore::estimated_bytes_ground_truth) + .sum(); + let term_dicts: usize = self + .field_term_dicts + .iter() + .map(TermDictionary::resident_bytes_ground_truth) + .sum(); + + #[cfg(feature = "text-index")] + let fst: usize = self + .fst_maps + .iter() + .filter_map(|m| m.as_ref()) + .map(|m| m.as_fst().size()) + .sum(); + #[cfg(not(feature = "text-index"))] + let fst: usize = 0; + + let doc_field_lengths: usize = self + .doc_field_lengths + .values() + .map(|v| { + std::mem::size_of::() + + v.len() * std::mem::size_of::() + + MAP_ENTRY_OVERHEAD + }) + .sum(); + let key_hash_to_doc_id = self.key_hash_to_doc_id.len() + * (std::mem::size_of::() + std::mem::size_of::() + MAP_ENTRY_OVERHEAD); + let doc_id_to_key: usize = self + .doc_id_to_key + .values() + .map(|k| std::mem::size_of::() + k.len() + MAP_ENTRY_OVERHEAD) + .sum(); + let lsn_maps = (self.doc_id_to_insert_lsn.len() + self.doc_id_to_delete_lsn.len()) + * (std::mem::size_of::() + std::mem::size_of::() + MAP_ENTRY_OVERHEAD); + + #[cfg(feature = "text-index")] + let tag: usize = self + .tag_indexes + .iter() + .map(|(field, inner)| { + field.len() + + MAP_ENTRY_OVERHEAD + + inner + .iter() + .map(|(v, bm)| { + v.len() + + MAP_ENTRY_OVERHEAD + + EMPTY_BITMAP_BASE_COST + + bm.len() as usize * ROARING_BIT_APPROX_COST + }) + .sum::() + }) + .sum::() + + self + .doc_tag_entries + .values() + .map(|entries| Self::tag_entries_cost(entries)) + .sum::(); + #[cfg(not(feature = "text-index"))] + let tag: usize = 0; + + #[cfg(feature = "text-index")] + let numeric: usize = self + .numeric_indexes + .iter() + .map(|(field, tree)| { + field.len() + + MAP_ENTRY_OVERHEAD + + tree + .values() + .map(|bm| { + std::mem::size_of::() + + MAP_ENTRY_OVERHEAD + + EMPTY_BITMAP_BASE_COST + + bm.len() as usize * ROARING_BIT_APPROX_COST + }) + .sum::() + }) + .sum::() + + self + .doc_numeric_entries + .values() + .map(|entries| Self::numeric_entries_cost(entries)) + .sum::(); + #[cfg(not(feature = "text-index"))] + let numeric: usize = 0; + + postings + + term_dicts + + fst + + doc_field_lengths + + key_hash_to_doc_id + + doc_id_to_key + + lsn_maps + + tag + + numeric + } + /// Hard-delete a document identified by `doc_id` from all inverted indexes. /// /// Removes: @@ -1316,44 +1730,63 @@ impl TextIndex { } // ── TAG field removal ───────────────────────────────────────────────── + // K4 (P0 fix): shared `tag_bitmap_revoke`/`tag_entries_cost` helpers + // -- same logic `tag_index_document`'s revoke loop uses -- so the two + // call sites cannot drift apart. #[cfg(feature = "text-index")] if let Some(entries) = self.doc_tag_entries.remove(&doc_id) { + self.resident_bytes_extra = self + .resident_bytes_extra + .saturating_sub(Self::tag_entries_cost(&entries)); for (field, value) in entries { - if let Some(field_map) = self.tag_indexes.get_mut(&field) { - if let Some(bm) = field_map.get_mut(&value) { - bm.remove(doc_id); - if bm.is_empty() { - field_map.remove(&value); - } - } - } + self.tag_bitmap_revoke(&field, &value, doc_id); } } // ── NUMERIC field removal ───────────────────────────────────────────── + // K4 (P0 fix): shared `numeric_bitmap_revoke`/`numeric_entries_cost`. #[cfg(feature = "text-index")] if let Some(entries) = self.doc_numeric_entries.remove(&doc_id) { + self.resident_bytes_extra = self + .resident_bytes_extra + .saturating_sub(Self::numeric_entries_cost(&entries)); for (field, value) in entries { - if let Some(btree) = self.numeric_indexes.get_mut(&field) { - if let Some(bm) = btree.get_mut(&value) { - bm.remove(doc_id); - if bm.is_empty() { - btree.remove(&value); - } - } - } + self.numeric_bitmap_revoke(&field, &value, doc_id); } } // ── Metadata cleanup ────────────────────────────────────────────────── - self.doc_field_lengths.remove(&doc_id); + // K4 (P0 fix): uncharge using the ACTUAL removed Vec's length -- + // self-correcting even if field_count ever varied per doc (it + // currently doesn't). + if let Some(lengths) = self.doc_field_lengths.remove(&doc_id) { + self.resident_bytes_extra = self.resident_bytes_extra.saturating_sub( + std::mem::size_of::() + + lengths.len() * std::mem::size_of::() + + MAP_ENTRY_OVERHEAD, + ); + } // Remove from key_hash -> doc_id map (need to find the key_hash). if let Some(key) = self.doc_id_to_key.remove(&doc_id) { let key_hash = xxhash_rust::xxh64::xxh64(&key, 0); self.key_hash_to_doc_id.remove(&key_hash); + self.uncharge_doc_key(key.len()); + } + if self.doc_id_to_insert_lsn.remove(&doc_id).is_some() { + self.resident_bytes_extra = self.resident_bytes_extra.saturating_sub( + std::mem::size_of::() + std::mem::size_of::() + MAP_ENTRY_OVERHEAD, + ); + } + // `doc_id_to_delete_lsn` currently has no insertion call site anywhere + // in the codebase (reserved for future v0.2 logical-delete wiring -- + // see the field's doc comment on `TextIndex`), so this `.remove()` is + // always a no-op today. Written defensively symmetric so a future + // insert-side wiring doesn't silently leak accounting. + if self.doc_id_to_delete_lsn.remove(&doc_id).is_some() { + self.resident_bytes_extra = self.resident_bytes_extra.saturating_sub( + std::mem::size_of::() + std::mem::size_of::() + MAP_ENTRY_OVERHEAD, + ); } - self.doc_id_to_insert_lsn.remove(&doc_id); - self.doc_id_to_delete_lsn.remove(&doc_id); } } @@ -1485,6 +1918,16 @@ impl TextStore { } } + /// Approximate total resident bytes across every text index on this + /// shard (K4 accounting spine). Sum of `TextIndex::resident_bytes()`; + /// see that method's doc comment for what is counted/excluded. `0` for + /// an empty store -- called from the shard's 100ms tick and published + /// into `store_memory.text`, which previously never left its + /// hard-coded-0 initial value. + pub fn resident_bytes(&self) -> usize { + self.indexes.values().map(TextIndex::resident_bytes).sum() + } + /// Collect schema-only metadata from all text indexes for persistence. pub fn collect_index_metas(&self) -> Vec { self.indexes @@ -1755,7 +2198,10 @@ impl TextStore { if field_idx < idx.fst_maps.len() { if let Some(bytes) = fst_bytes_opt { match fst::Map::new(bytes) { - Ok(map) => idx.fst_maps[field_idx] = Some(map), + // K4 (P0 fix): route through set_fst_map so the + // resident_bytes_extra accounting stays in sync + // (same helper build_fst uses). + Ok(map) => idx.set_fst_map(field_idx, Some(map)), Err(e) => tracing::warn!( "FST load failed for {}[{}]: {}", String::from_utf8_lossy(name.as_ref()), @@ -2268,6 +2714,183 @@ mod tests { assert!(idx.is_doc_visible_at(0, 77)); assert!(!idx.is_doc_visible_at(0, 76)); } + + // ── K4 accounting spine: TextIndex/TextStore resident_bytes ────────── + + #[test] + fn resident_bytes_zero_for_empty_index() { + let idx = make_index_with_docs(&[]); + assert_eq!(idx.resident_bytes(), 0); + } + + #[test] + fn resident_bytes_grows_with_indexed_docs() { + let empty = make_index_with_docs(&[]); + let indexed = make_index_with_docs(&[ + ("doc:1", "the quick brown fox jumps over the lazy dog"), + ("doc:2", "a completely different sentence about cats"), + ]); + assert!( + indexed.resident_bytes() > empty.resident_bytes(), + "indexed docs must grow resident_bytes: empty={} indexed={}", + empty.resident_bytes(), + indexed.resident_bytes() + ); + } + + /// RED-first (K4 stage 2 contract): an empty `TextStore` must report 0, + /// and creating an index + indexing documents must strictly grow the + /// store-level total -- the aggregate the elastic memory budget's + /// used-term (`ShardStoreMemory::text`) now sees. + #[test] + fn text_store_resident_bytes_grows_after_indexing() { + let mut store = TextStore::new(); + assert_eq!(store.resident_bytes(), 0, "empty store reports 0"); + + let idx = make_index_with_docs(&[ + ("doc:1", "the quick brown fox jumps over the lazy dog"), + ("doc:2", "a completely different sentence about cats"), + ("doc:3", "yet another document with distinct vocabulary"), + ]); + store + .create_index(Bytes::from_static(b"test_idx"), idx) + .expect("create_index"); + + assert!( + store.resident_bytes() > 0, + "store with an indexed document must report > 0" + ); + } + + /// K4 (P0 fix): RED-first -- the O(1) incremental accumulator + /// (`resident_bytes_extra` plus the per-field `PostingStore`/ + /// `TermDictionary` caches) must never drift from a from-scratch + /// ground-truth recompute, across a mixed sequence of: indexing N docs + /// across TEXT + TAG + NUMERIC fields, an upsert that changes all three, + /// an FST build (structural resync event), a partial hard-delete, and + /// draining the index back to empty. + #[test] + fn resident_bytes_matches_ground_truth_after_mixed_mutations() { + use crate::protocol::Frame; + use crate::text::types::{NumericFieldDef, TagFieldDef}; + + let text_field = TextFieldDef::new(Bytes::from_static(b"body")); + let tag_field = TagFieldDef { + field_name: Bytes::from_static(b"status"), + separator: b',', + case_sensitive: false, + sortable: false, + noindex: false, + }; + let numeric_field = NumericFieldDef { + field_name: Bytes::from_static(b"score"), + sortable: false, + noindex: false, + }; + let mut idx = TextIndex::new_with_schema( + Bytes::from_static(b"mixed_idx"), + Vec::new(), + vec![text_field], + vec![tag_field], + vec![numeric_field], + BM25Config::default(), + ); + assert_eq!(idx.resident_bytes(), idx.resident_bytes_ground_truth()); + + let make_args = |body: &str, status: &str, score: &str| { + vec![ + Frame::BulkString(Bytes::from_static(b"body")), + Frame::BulkString(Bytes::copy_from_slice(body.as_bytes())), + Frame::BulkString(Bytes::from_static(b"status")), + Frame::BulkString(Bytes::copy_from_slice(status.as_bytes())), + Frame::BulkString(Bytes::from_static(b"score")), + Frame::BulkString(Bytes::copy_from_slice(score.as_bytes())), + ] + }; + + // K4 (P0 fix): use the REAL xxh64 hash of the key, not a synthetic + // index -- `remove_doc_by_doc_id` recomputes `key_hash` from the + // stored key bytes via `xxhash_rust::xxh64::xxh64` to evict + // `key_hash_to_doc_id` (production callers, e.g. `spsc_handler.rs`, + // always pass the real hash). A synthetic key_hash would desync that + // recomputation from the map's actual key, silently leaking the + // `key_hash_to_doc_id` entry while still uncharging it -- a test-only + // trap, not a production accounting bug. + let key_hash_of = |key: &str| xxhash_rust::xxh64::xxh64(key.as_bytes(), 0); + + let docs = [ + ("doc:1", "the quick brown fox", "open,urgent", "1.5"), + ("doc:2", "a lazy dog sleeps", "closed", "2.0"), + ("doc:3", "quick fox jumps again", "open", "3.5"), + ]; + for (key, body, status, score) in docs { + let key_hash = key_hash_of(key); + let args = make_args(body, status, score); + idx.index_document(key_hash, key.as_bytes(), &args); + idx.tag_index_document(key_hash, key.as_bytes(), &args); + idx.numeric_index_document(key_hash, key.as_bytes(), &args); + assert_eq!( + idx.resident_bytes(), + idx.resident_bytes_ground_truth(), + "drift after indexing {key}" + ); + } + + // Upsert doc:1 -- new TEXT + TAG + NUMERIC content on the same key. + let key_hash = key_hash_of("doc:1"); + let args = make_args("totally different body text", "closed", "9.9"); + idx.index_document(key_hash, b"doc:1", &args); + idx.tag_index_document(key_hash, b"doc:1", &args); + idx.numeric_index_document(key_hash, b"doc:1", &args); + assert_eq!(idx.resident_bytes(), idx.resident_bytes_ground_truth()); + + // FST build -- structural resync event, not a periodic walk. + idx.build_fst(); + assert_eq!(idx.resident_bytes(), idx.resident_bytes_ground_truth()); + + // Partial hard-delete: remove doc:2 by its assigned doc_id (1 -- + // insertion order, 0-based, matches `docs` above). + idx.remove_doc_by_doc_id(1); + assert_eq!(idx.resident_bytes(), idx.resident_bytes_ground_truth()); + + // Drain the rest -- resident_bytes must settle back near zero, + // matching ground truth exactly (entry-overhead floors survive, + // matching PostingStore/TermDictionary's "entries never removed" + // contract). + idx.remove_doc_by_doc_id(0); + idx.remove_doc_by_doc_id(2); + assert_eq!(idx.resident_bytes(), idx.resident_bytes_ground_truth()); + } + + /// K4 (P0 fix): `resident_bytes()` must be a pure O(1) load with no + /// iteration in the accessor -- enforced by construction here: called + /// 100k times against a 5,000-doc index, well within a wall-clock budget + /// that an O(n) walk (the reviewer measured 6.4ms/call at 50K docs) + /// would blow through by orders of magnitude. + #[test] + fn resident_bytes_is_o1_not_a_walk() { + let keys: Vec = (0..5_000).map(|i| format!("doc:{i}")).collect(); + let docs: Vec<(&str, &str)> = keys + .iter() + .map(|k| { + ( + k.as_str(), + "the quick brown fox jumps over the lazy dog and keeps going", + ) + }) + .collect(); + let idx = make_index_with_docs(&docs); + + let start = std::time::Instant::now(); + for _ in 0..100_000 { + std::hint::black_box(idx.resident_bytes()); + } + let elapsed = start.elapsed(); + assert!( + elapsed < std::time::Duration::from_millis(300), + "100k reads of resident_bytes() took {elapsed:?} -- looks like a walk, not O(1)" + ); + } } // Plan 152-06 TAG storage tests live in a sibling file so runtime code in diff --git a/src/text/term_dict.rs b/src/text/term_dict.rs index 2fee5b611..7d1994396 100644 --- a/src/text/term_dict.rs +++ b/src/text/term_dict.rs @@ -10,6 +10,14 @@ use std::collections::HashMap; /// /// IDs are assigned sequentially starting from 0. Once assigned, a term's /// ID is stable for the lifetime of the dictionary. +/// Fixed per-entry `HashMap` bucket-overhead constant used to approximate +/// `TermDictionary::resident_bytes` (K4 accounting spine). Not exact -- +/// `hashbrown`'s SwissTable control-byte layout is an implementation detail +/// -- but monotonic, matching `Database`'s own `entry_overhead` +/// approximation (WS6) and `graph::index::PropertyIndex::resident_bytes`'s +/// convention. +const MAP_ENTRY_OVERHEAD: usize = 48; + pub struct TermDictionary { terms: HashMap, next_id: u32, @@ -17,6 +25,11 @@ pub struct TermDictionary { /// Used by the dual-path expansion strategy (D-12): FST covers ids < fst_high_water_mark, /// HashMap brute-force covers ids >= fst_high_water_mark. pub fst_high_water_mark: u32, + /// K4 (P0 fix): O(1) cached total mirroring `resident_bytes()`. + /// Maintained incrementally in `get_or_insert`'s new-term branch -- + /// `TermDictionary` has no deletion path (ids are stable for the + /// dictionary's lifetime), so insertion is the only mutation site. + resident_bytes: usize, } impl TermDictionary { @@ -26,6 +39,7 @@ impl TermDictionary { terms: HashMap::new(), next_id: 0, fst_high_water_mark: 0, + resident_bytes: 0, } } @@ -38,6 +52,7 @@ impl TermDictionary { } let id = self.next_id; self.next_id += 1; + self.resident_bytes += term.len() + std::mem::size_of::() + MAP_ENTRY_OVERHEAD; self.terms.insert(term.to_owned(), id); id } @@ -63,4 +78,83 @@ impl TermDictionary { pub fn next_id(&self) -> u32 { self.next_id } + + /// Approximate resident bytes: term string length + the `u32` id value + /// plus a fixed per-entry `HashMap` bucket-overhead constant (K4 + /// accounting spine). + /// + /// K4 (P0 fix): O(1) cached read, maintained incrementally by + /// `get_or_insert` -- this used to be an O(vocabulary size) walk called + /// unconditionally every 100ms from the shard eviction tick, which does + /// not scale with corpus size. See `resident_bytes_ground_truth` + /// (`#[cfg(test)]`) for the equivalent full-walk formula this cached + /// value must always match. + #[must_use] + pub fn resident_bytes(&self) -> usize { + self.resident_bytes + } + + /// Ground-truth full recompute of `resident_bytes()`, using the exact + /// same fixed-cost formula as the incremental accumulator. Test-only. + #[cfg(test)] + pub(crate) fn resident_bytes_ground_truth(&self) -> usize { + self.terms + .keys() + .map(|term| term.len() + std::mem::size_of::() + MAP_ENTRY_OVERHEAD) + .sum() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn resident_bytes_zero_when_empty() { + let dict = TermDictionary::new(); + assert_eq!(dict.resident_bytes(), 0); + } + + #[test] + fn resident_bytes_grows_with_inserts() { + let mut dict = TermDictionary::new(); + let before = dict.resident_bytes(); + dict.get_or_insert("hello"); + dict.get_or_insert("world"); + assert!(dict.resident_bytes() > before); + // Re-inserting an existing term must not double-count. + let after_dup = dict.resident_bytes(); + dict.get_or_insert("hello"); + assert_eq!(dict.resident_bytes(), after_dup); + } + + /// K4 (P0 fix): RED-first — the O(1) incremental accumulator must never + /// drift from a from-scratch ground-truth recompute. + #[test] + fn resident_bytes_matches_ground_truth_after_inserts() { + let mut dict = TermDictionary::new(); + assert_eq!(dict.resident_bytes(), dict.resident_bytes_ground_truth()); + for term in ["alpha", "beta", "gamma", "alpha", "delta", "beta"] { + dict.get_or_insert(term); + assert_eq!(dict.resident_bytes(), dict.resident_bytes_ground_truth()); + } + } + + /// K4 (P0 fix): `resident_bytes()` must be a pure O(1) load, not a walk. + #[test] + fn resident_bytes_is_o1_not_a_walk() { + let mut dict = TermDictionary::new(); + for i in 0..50_000u32 { + dict.get_or_insert(&format!("term-{i}")); + } + let start = std::time::Instant::now(); + for _ in 0..100_000 { + std::hint::black_box(dict.resident_bytes()); + } + let elapsed = start.elapsed(); + assert!( + elapsed < std::time::Duration::from_millis(200), + "100k reads of resident_bytes() took {elapsed:?} -- looks like a walk, not O(1)" + ); + } } diff --git a/tests/memory_prometheus_kinds.rs b/tests/memory_prometheus_kinds.rs index 0864d7f09..21b3b35ed 100644 --- a/tests/memory_prometheus_kinds.rs +++ b/tests/memory_prometheus_kinds.rs @@ -2,7 +2,7 @@ //! (Phase 190 Plan 03). //! //! Spawns the release moon binary with an admin port, loads a small -//! dataset, scrapes `/metrics`, and verifies all 7 subsystem kinds are +//! dataset, scrapes `/metrics`, and verifies all 9 subsystem kinds are //! present with their sum within +/-10% of `moon_rss_bytes`. //! //! Run with: @@ -26,7 +26,19 @@ fn redis_cli_available() -> bool { .unwrap_or(false) } +/// Resolve the release binary to spawn. Honors `MOON_BIN` when set so a +/// pinned, freshly-built ELF binary can be supplied explicitly -- required +/// inside the OrbStack Linux VM, where the shared checkout's +/// `target/release/moon` may be a macOS Mach-O binary that gets silently +/// host-proxied back to the Mac (the port never binds VM-side, producing a +/// 30s accept timeout with no obvious cause). See +/// `gotcha_orbstack_macho_binary_trap`. fn release_binary() -> std::path::PathBuf { + if let Ok(bin) = std::env::var("MOON_BIN") { + if !bin.trim().is_empty() { + return std::path::PathBuf::from(bin); + } + } std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("target/release/moon") } @@ -175,18 +187,25 @@ fn parse_rss_bytes(body: &str) -> Option { None } -const EXPECTED_KINDS: [&str; 7] = [ +// NOTE: `lua_scripts` was already emitted by `update_moon_memory_bytes` +// (C4, wave-5 hygiene) but missing from this list -- a pre-existing +// test/code mismatch this file's own count assertion should have caught. +// Fixed alongside the K4 "text" addition since both land in this file. +const EXPECTED_KINDS: [&str; 9] = [ "dashtable", "hnsw", + // K4 (kernel-m2-brief-2026-07-12 stage 2): text (FTS) resident bytes. + "text", "csr", "wal", "sealed", "replication_backlog", + "lua_scripts", "allocator_overhead", ]; #[test] -fn metrics_endpoint_emits_seven_memory_kinds() { +fn metrics_endpoint_emits_nine_memory_kinds() { let Some(m) = spawn_moon() else { return }; // Load 1000 string keys so DashTable has non-zero resident bytes. @@ -218,7 +237,7 @@ fn metrics_endpoint_emits_seven_memory_kinds() { thread::sleep(Duration::from_secs(2)); } - // ── Assert all 7 kinds present ────────────────────────────────────── + // ── Assert all 9 kinds present ────────────────────────────────────── for expected in &EXPECTED_KINDS { assert!( kinds.contains_key(*expected), @@ -230,8 +249,8 @@ fn metrics_endpoint_emits_seven_memory_kinds() { } assert_eq!( kinds.len(), - 7, - "Expected exactly 7 kinds, got {}: {kinds:?}", + 9, + "Expected exactly 9 kinds, got {}: {kinds:?}", kinds.len() ); @@ -245,7 +264,7 @@ fn metrics_endpoint_emits_seven_memory_kinds() { let sum: f64 = kinds.values().sum(); assert!( sum > 0.0, - "Sum of all 7 kinds is 0 — update hook may not have fired" + "Sum of all 9 kinds is 0 — update hook may not have fired" ); // ── Assert dashtable > 0 after loading 1000 keys ────────────────────