diff --git a/CHANGELOG.md b/CHANGELOG.md index c144e3031..e6c6e2327 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,99 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — RSS/CPU remediation wave 5 (PR #TBD) + +- **Item A — mmap the exact-rerank f16 sidecar on segment reload** + (`src/vector/segment/raw_f16_store.rs`, new `RawF16Store` enum): a segment + reloaded from disk used to `fs::read` the entire `raw_f16.bin` sidecar into + a second heap `Vec`, doubling resident vector memory for + reload-heavy deployments (warm starts, segment promotion). Reload now + memory-maps the file (`memmap2`, already a workspace dependency) and hands + out a zero-copy `&[u16]` view backed by the kernel page cache — RSS only + grows for pages the rerank path actually touches. Freshly-built segments + (compaction/merge) are unaffected — they keep their owned buffer. Rerank + parity (Owned vs Mapped, byte-identical sidecar + identical `search()` + output) is pinned by + `test_reload_raw_f16_sidecar_uses_mmap_and_matches_owned_rerank`. +- **Item A follow-up — text posting-list capacity reclaim** + (`src/text/posting.rs`): `PostingList::term_freqs`/`positions` grow to the + peak document count ever seen for a term and, per the existing + `remove_doc` contract, the `postings` HashMap entry is kept forever even + once a term has zero live documents. The buffers now `shrink_to_fit()` + once the last document leaves a posting, releasing peak capacity for + terms that go idle without changing the "entry survives" contract. +- **Item B — AOF writer idle wake made adaptive** (`src/persistence/aof/writer_task.rs`, + new `IdleWait` state machine): the 3 steady-state writer loops that need a + bounded channel poll to service the EverySec proactive-fsync deadline + (TopLevel tokio, PerShard tokio, PerShard monoio — TopLevel monoio blocks + on an untimed `rx.recv()` and needed no change) used to poll at a FIXED + cadence forever (50ms monoio / 200ms tokio), waking an idle server's AOF + writer thread 5-20 times a second doing nothing. The wait now escalates + 50ms → 250ms → 1s once a poll times out with nothing queued, and resets to + the floor the instant any message arrives — a real write always wakes the + loop immediately regardless of the current timeout, since the poll races + a message against the deadline. Escalation is refused (pinned at the + floor) whenever a write is buffered under `FsyncPolicy::EverySec` without + an immediate fsync, or `last_fsync` was manually back-dated (the F6 + post-fold drain trick) — the ~1.2s EverySec bound is provably unchanged. + `FsyncPolicy::Always`/`No` have no such deadline and escalate freely once + idle. +- **Item C1 — WAL v3 write buffer shrinks after an oversized flush** + (`src/persistence/wal_v3/segment.rs`): a single large record (e.g. a + FullPageImage) grew the 8KB write buffer to fit it, and `clear()` alone + never released that capacity — the peak allocation was pinned for the + writer's lifetime. `flush_write`/`rotate_segment` now `shrink_to` the + 8KB default once capacity exceeds 4x that, a no-op for the common + small-record case. +- **Item C2 — SearchScratch visited-set: already bitset-based (SKIP)** + (`src/vector/hnsw/search.rs`): the per-query search hot path already uses + a word-based `BitVec` (u64 words, `test_and_set`/`clear_all` memset), + thread-cached and reused across queries — no change needed. The other + `Vec` visited sets found in the vector module are all build-time/ + compaction/merge-oracle code, not the per-query path; `search_sq.rs` in + particular carries an explicit comment warning that a prior BitVec + conversion there caused correctness issues, so it was left untouched. +- **Item C3 — SmallVec the per-tick elastic-budget shard snapshot** + (`src/shard/shared_databases.rs`): `recompute_elastic_budget` (called + from every shard's 100ms eviction tick) `collect()`ed a fresh + `Vec` snapshot of all shards' published memory on every call. + Switched to `SmallVec<[usize; 16]>` — stack-only for the common <=16 + shard case, unchanged single heap allocation beyond that. +- **Item C4 — Lua script-cache byte estimate exposed via INFO/MEMORY + DOCTOR** (`src/scripting/cache.rs`, `ScriptCache::resident_bytes()`): the + per-shard Lua cache was invisible to observability — its growth folded + silently into "allocator overhead." Added a byte-estimate accounting + method, published per-shard via the existing C5/M4 `ShardStoreMemory` + tick pattern (new `lua` atomic), and surfaced in both the Prometheus + `moon_memory_bytes{kind="lua_scripts"}` gauge and `MEMORY DOCTOR`'s text + report. The cache itself remains intentionally unbounded (Redis parity — + `SCRIPT FLUSH` is the only eviction path); this is observability only. +- **Item C5 — removed dead `parse_single_frame_zc` RESP parser** + (`src/protocol/parse.rs`): a full ~150-line RESP2/RESP3 parser + superseded by the current `validate_frame` + `parse_frame_zerocopy` + pipeline, with zero external callers (only self-recursion) — silently + masked by the file's `#![allow(dead_code)]`. Removed along with its + exclusively-private helper `read_decimal_zc`. +- **Item C6 — jemalloc decay policy audited, docs added (SKIP code + change)** (`CLAUDE.md`): the baked-in `_rjem_malloc_conf` static and the + `--memory-arenas-cap` re-spawn override already carry byte-identical + `dirty_decay_ms:1000,muzzy_decay_ms:5000,background_thread:true` tuning + — no drift to reconcile. Added the missing operator-facing + `_RJEM_MALLOC_CONF` documentation (docs-only, no code changed). +- **Item C7 — tokio 1ms shard tick idle cost audited (SKIP)** + (`src/shard/event_loop.rs`, `src/shard/spsc_handler.rs`): the 1ms + `periodic_interval` tick's SPSC drain is already a non-blocking, + zero-allocation `try_pop()` loop, and every downstream side effect is + already gated behind a cheap conditional. The one unconditional cost + (`cached_clock.update()`, a single `clock_gettime`) is the documented + "Timestamp caching" design. Unlike item B's AOF writer poll, this 1ms + cadence IS the low-latency WAL-flush contract (CLAUDE.md), not + incidental idle waste — escalating it would widen that bound. No code + change; a real fix would be event-driven WAL triggering, an + architectural change out of scope here. +- Item C8 (sigterm readiness deadline) landed early via the Windows-CI PR + (#229) — see the CI section below. + ### CI — fix Windows main-push test failures (PR #TBD) - `test_poll_real_process_smoke` is now gated to Linux/macOS: `get_rss_bytes()` diff --git a/CLAUDE.md b/CLAUDE.md index b86f1ea6f..44f64b5f9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -96,6 +96,7 @@ orb run -m moon-dev bash -c 'sudo apt-get update -qq && sudo apt-get install -y - `MOON_URING_SPIN_US`, `MOON_URING_SQPOLL[_CPU]`, `MOON_URING_PLAIN` — io_uring-side experiment gates kept as documented diagnostics; all dead ends for the p=1 path (see `tmp/KV-FULLPROOF.md` Round 2). - `MOON_XSHARD_SPIN_BUDGET` / `MOON_XSHARD_SPIN_GATE` / `MOON_XSHARD_SPIN_MAX_CONNS` — diagnostic overrides for the C2 reply-side spin (`src/shard/slice.rs`; defaults 4096 iters / gate 2 / **solo-conn 1**). Budget `0` disables the spin entirely (the same-instance A/B knob that proved the c8P1 convoy). ⚠ The solo-conn ceiling (spin only when the conn is ALONE on its shard thread) is the L1 convoy fix — raising `MAX_CONNS` re-creates the s4 c8P1 collapse (a spinning conn starves its sibling AND the shard's SPSC drain, 0.45× vs Redis; fixed = 2.75× better, see `tmp/MULTISHARD-REDESIGN.md`). Bench-only knobs: never set in production. - `RUSTFLAGS="-C target-cpu=native"` — enable CPU-specific optimizations for benchmarking +- `_RJEM_MALLOC_CONF` — jemalloc's tuning knob (prefixed because `tikv-jemallocator` builds with the `_rjem_` symbol prefix; the unprefixed `MALLOC_CONF` has no effect). Moon bakes in `narenas:8,background_thread:true,metadata_thp:auto,dirty_decay_ms:1000,muzzy_decay_ms:5000,abort_conf:true` via a static `_rjem_malloc_conf` export (`src/main.rs`) — 1s dirty-page decay + a background reclaim thread so freed-but-idle pages return to the OS quickly instead of sitting in jemalloc's dirty/muzzy caches inflating RSS. `--memory-arenas-cap N` re-spawns the process (`execve`) with this exact string, narenas substituted, **before** jemalloc's one-time init reads it — `mallctl` after init is a documented no-op for `opt.narenas`. If `_RJEM_MALLOC_CONF` is already set in the environment, `--memory-arenas-cap` is a no-op (operator env wins, warns instead of clobbering). Only applies to `--features jemalloc` builds; `mimalloc` (the non-jemalloc default) has no equivalent decay knob in this codebase. ## Key Design Decisions diff --git a/src/admin/metrics_setup.rs b/src/admin/metrics_setup.rs index 2c55dda27..c65583257 100644 --- a/src/admin/metrics_setup.rs +++ b/src/admin/metrics_setup.rs @@ -1360,6 +1360,7 @@ fn update_moon_memory_bytes() { let mut csr: usize = 0; let wal: usize = 0; // WalWriterV3 is stack-owned; not reachable here let mut backlog: usize = 0; + let mut lua: usize = 0; if let Some(shard_dbs) = get_global_shard_databases() { // KV memory: sum of per-shard published atomics. Lock-free. @@ -1372,6 +1373,8 @@ fn update_moon_memory_bytes() { hnsw += mem.vector.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. + lua += mem.lua.load(Ordering::Relaxed); } } @@ -1382,7 +1385,7 @@ fn update_moon_memory_bytes() { } } - let other_sum = dashtable + hnsw + csr + wal + sealed + backlog; + let other_sum = dashtable + hnsw + csr + wal + sealed + backlog + lua; let alloc_overhead = rss.saturating_sub(other_sum); gauge!("moon_memory_bytes", "kind" => "dashtable").set(dashtable as f64); @@ -1391,6 +1394,7 @@ fn update_moon_memory_bytes() { gauge!("moon_memory_bytes", "kind" => "wal").set(wal as f64); gauge!("moon_memory_bytes", "kind" => "sealed").set(sealed as f64); gauge!("moon_memory_bytes", "kind" => "replication_backlog").set(backlog as f64); + gauge!("moon_memory_bytes", "kind" => "lua_scripts").set(lua as f64); gauge!("moon_memory_bytes", "kind" => "allocator_overhead").set(alloc_overhead as f64); // Update the existing RSS gauge in the same snapshot so the integration diff --git a/src/command/server_admin.rs b/src/command/server_admin.rs index 845d5dc79..c583eaf32 100644 --- a/src/command/server_admin.rs +++ b/src/command/server_admin.rs @@ -410,6 +410,7 @@ fn memory_doctor() -> Frame { #[cfg_attr(not(feature = "graph"), allow(unused_variables))] let csr_bytes: usize; let wal_bytes: usize = 0; + let lua_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. @@ -418,16 +419,21 @@ fn memory_doctor() -> Frame { // Store memory: sum published per-shard vector/graph atomics. let mut vec_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); 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; csr_bytes = csr_total; + lua_bytes = lua_total; } else { dashtable_bytes = 0; hnsw_bytes = 0; csr_bytes = 0; + lua_bytes = 0; } // Replication backlog via global state (same pattern as INFO replication). @@ -440,8 +446,13 @@ fn memory_doctor() -> Frame { let (allocator_name, arena_count) = allocator_info(); // ── Computed overhead ──────────────────────────────────────────────── - let tracked_sum = - dashtable_bytes + hnsw_bytes + csr_bytes + wal_bytes + sealed_bytes + repl_bytes; + let tracked_sum = dashtable_bytes + + hnsw_bytes + + csr_bytes + + wal_bytes + + sealed_bytes + + repl_bytes + + lua_bytes; let allocator_overhead = rss.saturating_sub(tracked_sum); // ── VSZ ratio recommendation ───────────────────────────────────────── @@ -515,6 +526,12 @@ fn memory_doctor() -> Frame { humanize_bytes(repl_bytes), pct(repl_bytes, rss) ); + let _ = writeln!( + out, + " Lua scripts: {} ({:.1}%)", + humanize_bytes(lua_bytes), + pct(lua_bytes, rss) + ); let _ = writeln!( out, " Allocator overhead: {} ({:.1}%)", diff --git a/src/persistence/aof/writer_task.rs b/src/persistence/aof/writer_task.rs index 188b8db4a..8d9176d0f 100644 --- a/src/persistence/aof/writer_task.rs +++ b/src/persistence/aof/writer_task.rs @@ -22,6 +22,150 @@ use super::group_commit::{ #[cfg(feature = "runtime-monoio")] use super::group_commit::{GroupCommitSink, commit_group_commit_batch}; +/// Idle-adaptive wake cadence for a background AOF writer's channel poll +/// (RSS/CPU wave 5, item B). +/// +/// The steady-state writer loops (PerShard monoio/tokio, TopLevel tokio — +/// TopLevel monoio blocks on an untimed `rx.recv()` and needs none of this) +/// poll their channel with a bounded timeout so the EverySec proactive-fsync +/// deadline check that follows every wake still fires when no new Appends +/// ever arrive. A FIXED cadence forever (previously 50ms monoio / 200ms +/// tokio) means an idle server's AOF writer thread wakes 5-20 times a +/// second doing nothing. Escalating the wait once a poll times out with +/// nothing queued costs nothing: the poll races a message against the +/// deadline, so a real write always wakes the loop immediately regardless +/// of how long the timeout is set — only the "still idle, re-check +/// nothing" cadence relaxes. +/// +/// # The one invariant that must never regress +/// +/// The EverySec bound ("the oldest unflushed byte reaches disk within ~1s + +/// one wake") must hold exactly as it did under the old fixed cadence. +/// [`IdleWait`] enforces this the same way the ground rules require: +/// escalation is refused (stays pinned at the fast floor) whenever +/// [`Self::mark_pending`] has been called and not yet cleared by +/// [`Self::clear_pending`] — i.e. whenever there is a write buffered under +/// `FsyncPolicy::EverySec` that has not yet been fsynced, or a manually +/// back-dated `last_fsync` (the F6 post-fold drain trick) representing an +/// imminent deadline. `FsyncPolicy::Always` never buffers past its own +/// batch (fsynced same-iteration) and `FsyncPolicy::No` has no deadline at +/// all, so neither ever calls `mark_pending` — both escalate freely once +/// idle, which is correct: there is nothing time-sensitive to protect. +struct IdleWait { + step: usize, + pending: bool, +} + +/// Escalation ladder: fast floor for responsiveness right after activity, +/// capped at 1s (never longer than the EverySec deadline itself). +const AOF_IDLE_WAIT_STEPS: &[std::time::Duration] = &[ + std::time::Duration::from_millis(50), + std::time::Duration::from_millis(250), + std::time::Duration::from_secs(1), +]; + +impl IdleWait { + fn new() -> Self { + Self { + step: 0, + pending: false, + } + } + + /// Wait duration to use for the next channel poll. + fn current(&self) -> std::time::Duration { + AOF_IDLE_WAIT_STEPS[self.step] + } + + /// A message (data or control) was just received: reset to the fast + /// floor so the very next poll — which re-checks the EverySec deadline + /// — happens promptly again, exactly like the old fixed cadence did. + fn on_message(&mut self) { + self.step = 0; + } + + /// The poll timed out with nothing queued. Escalates towards the max + /// step UNLESS a deadline is still pending (see struct docs) — in that + /// case the wait stays at its current (already-fast, since + /// `on_message` just reset it) step so the deadline is re-checked + /// promptly instead of drifting out to the escalated cadence. + fn on_timeout(&mut self) { + if !self.pending { + self.step = (self.step + 1).min(AOF_IDLE_WAIT_STEPS.len() - 1); + } + } + + /// Mark that `last_fsync` now represents an unflushed/imminent deadline + /// (a batch was buffered under `FsyncPolicy::EverySec` without an + /// immediate fsync, or `last_fsync` was manually back-dated). Blocks + /// further escalation until [`Self::clear_pending`]. + fn mark_pending(&mut self) { + self.pending = true; + } + + /// The pending deadline was satisfied (a proactive or batch fsync just + /// succeeded) — escalation may resume from here. + fn clear_pending(&mut self) { + self.pending = false; + } +} + +#[cfg(test)] +mod idle_wait_tests { + use super::*; + + #[test] + fn starts_at_fast_floor() { + let w = IdleWait::new(); + assert_eq!(w.current(), AOF_IDLE_WAIT_STEPS[0]); + } + + #[test] + fn timeouts_escalate_and_cap_at_max() { + let mut w = IdleWait::new(); + w.on_timeout(); + assert_eq!(w.current(), AOF_IDLE_WAIT_STEPS[1]); + w.on_timeout(); + assert_eq!(w.current(), AOF_IDLE_WAIT_STEPS[2]); + // Capped: further timeouts stay at the max step. + w.on_timeout(); + assert_eq!(w.current(), AOF_IDLE_WAIT_STEPS[2]); + } + + #[test] + fn message_resets_to_floor_from_any_step() { + let mut w = IdleWait::new(); + w.on_timeout(); + w.on_timeout(); + assert_eq!(w.current(), AOF_IDLE_WAIT_STEPS[2]); + w.on_message(); + assert_eq!(w.current(), AOF_IDLE_WAIT_STEPS[0]); + } + + #[test] + fn pending_deadline_blocks_escalation() { + let mut w = IdleWait::new(); + w.mark_pending(); + // Never escalates while a deadline is pending, no matter how many + // consecutive timeouts occur. + w.on_timeout(); + w.on_timeout(); + w.on_timeout(); + assert_eq!(w.current(), AOF_IDLE_WAIT_STEPS[0]); + } + + #[test] + fn clearing_pending_resumes_escalation() { + let mut w = IdleWait::new(); + w.mark_pending(); + w.on_timeout(); + assert_eq!(w.current(), AOF_IDLE_WAIT_STEPS[0]); + w.clear_pending(); + w.on_timeout(); + assert_eq!(w.current(), AOF_IDLE_WAIT_STEPS[1]); + } +} + /// A sync [`GroupCommitSink`] over a `std::fs::File` for the monoio writer /// loops. `write_all` appends raw bytes (an empty buffer — a zero-length /// H1-BARRIER `AppendSync` — is a no-op); `sync` does the single per-batch @@ -118,6 +262,12 @@ pub async fn aof_writer_task( // path is exercised identically under both runtimes (shards=1 TopLevel). #[cfg(feature = "runtime-tokio")] let fail_fsync_for_test = std::env::var("MOON_TEST_AOF_FSYNC_FAIL").as_deref() == Ok("1"); + // Idle-adaptive channel-poll wake cadence (RSS/CPU wave 5, item B) — see + // `IdleWait` docs above. Declared outside the `loop` below (not inside + // the per-iteration `#[cfg(runtime-tokio)]` block) so its escalation + // state survives across iterations. + #[cfg(feature = "runtime-tokio")] + let mut idle_wait = IdleWait::new(); // Monoio path: multi-part AOF (base RDB + incremental RESP) with sync I/O. // @@ -348,16 +498,20 @@ pub async fn aof_writer_task( loop { #[cfg(feature = "runtime-tokio")] { - // Bounded recv (EverySec durability): wake at least every 200ms even - // when idle so the flush deadline check after this select! is honored - // within its 1s bound. A long-lived `interval.tick()` select arm is + // Bounded recv (EverySec durability): wake at least every + // `idle_wait.current()` (50ms floor, escalates to 1s while + // truly idle — see `IdleWait` docs; tighter than the old fixed + // 200ms right after activity, far looser once idle) even when + // idle so the flush + // deadline check after this select! is honored within its 1s + // bound. A long-lived `interval.tick()` select arm is // fairness-starvable under sustained writes and unreliable when idle // (see the per-shard writer below, which hit exactly that) — the // bounded recv cannot starve. flume's recv future is drop-safe on // the Elapsed branch (no message consumed on timeout). let recv_result = tokio::select! { r = tokio::time::timeout( - std::time::Duration::from_millis(200), + idle_wait.current(), rx.recv_async(), ) => r, _ = cancel.cancelled() => { @@ -375,7 +529,7 @@ pub async fn aof_writer_task( match recv_result { // Timeout (Elapsed): no message — fall through to the EverySec // deadline check after this block. - Err(_) => {} + Err(_) => idle_wait.on_timeout(), // Channel disconnected — final sync + shut down. Ok(Err(_)) => { if !write_error { @@ -386,6 +540,11 @@ pub async fn aof_writer_task( break; } Ok(Ok(first)) => { + // A message just arrived (data or control): reset the idle + // wait to its fast floor so the deadline check after this + // block — and every subsequent poll while there is still + // buffered/pending data — happens at the tight cadence. + idle_wait.on_message(); // Group commit: drain whatever else is queued into a bounded // batch so one fsync covers all (TopLevel = plain RESP bytes). let mut batch = collect_group_commit_batch( @@ -450,6 +609,13 @@ pub async fn aof_writer_task( .any(|m| matches!(m, AofMessage::AppendSync { .. })), "everysec/no batch must contain no AppendSync" ); + if fsync == FsyncPolicy::EverySec { + // Bytes are buffered but not yet durable — + // pin the idle wait at its floor until the + // deadline check below (or a future + // iteration's) clears it. + idle_wait.mark_pending(); + } BatchAck::Synced }; let _ = group_commit::ack_batch(&mut batch, verdict); @@ -494,8 +660,12 @@ pub async fn aof_writer_task( // Back-date so the backlog drained right after the // rewrite reaches disk within ~100ms + wake floor, // not a full second later (mirrors the per-shard - // writer's post-rewrite back-dating). + // writer's post-rewrite back-dating). This is itself + // a pending deadline — pin the idle wait at its + // floor so escalation cannot push the next check + // past the intended window. last_fsync = Instant::now() - std::time::Duration::from_millis(900); + idle_wait.mark_pending(); } Some(AofMessage::RewriteSharded(shard_dbs)) => { // C4 TopLevel cooperative fold (tokio path): @@ -538,8 +708,10 @@ pub async fn aof_writer_task( // Back-date so the channel backlog that accumulated // during the blocking fold reaches disk within ~100ms // + wake floor — a SIGKILL shortly after rewrite - // completion must not take the tail with it. + // completion must not take the tail with it. Pin the + // idle wait at its floor until this deadline fires. last_fsync = Instant::now() - std::time::Duration::from_millis(900); + idle_wait.mark_pending(); } // [F6] TopLevel writer never owns per-shard files — routing // bug. Self-abort so the countdown completes + flag clears. @@ -564,11 +736,14 @@ pub async fn aof_writer_task( } } // EverySec deadline: the oldest unflushed byte reaches disk at - // most ~1.2s after it was written (1s deadline + 200ms wake - // floor). tokio's BufWriter holds up to 8KB in userspace — a - // SIGKILL takes that tail with it, so the bound must hold even - // when the recv arm is saturated with messages. Skip if torn: - // syncing past a partial record cannot recover it. + // most ~1.2s after it was written (1s deadline + wake floor — + // the wake floor only, never the escalated idle cadence: see + // `IdleWait`, which `mark_pending`/`clear_pending` keep pinned + // at the floor for exactly this check). tokio's BufWriter holds + // up to 8KB in userspace — a SIGKILL takes that tail with it, so + // the bound must hold even when the recv arm is saturated with + // messages. Skip if torn: syncing past a partial record cannot + // recover it. if fsync == FsyncPolicy::EverySec && !write_error && last_fsync.elapsed() >= std::time::Duration::from_secs(1) @@ -576,6 +751,7 @@ pub async fn aof_writer_task( let _ = writer.flush().await; let _ = writer.get_ref().sync_data().await; last_fsync = Instant::now(); + idle_wait.clear_pending(); } } } @@ -715,9 +891,13 @@ pub async fn per_shard_aof_writer_task( let mut writer = tokio::io::BufWriter::new(file); let mut last_fsync = Instant::now(); + // Idle-adaptive channel-poll wake cadence (RSS/CPU wave 5, item B) — + // see `IdleWait` docs near the top of this file. + let mut idle_wait = IdleWait::new(); // (No `interval` here: the EverySec flush deadline is enforced by the // timeout-bounded recv in the loop below, which wakes at least every - // 200ms regardless of message traffic. A long-lived `interval.tick()` + // `idle_wait.current()` (50ms floor, escalates to 1s while idle) + // regardless of message traffic. A long-lived `interval.tick()` // select arm is fairness-starvable under sustained writes and proved // unreliable when idle on this dedicated current-thread writer runtime.) @@ -747,19 +927,21 @@ pub async fn per_shard_aof_writer_task( loop { tokio::select! { - // Bounded recv (EverySec durability): wake at least every 200ms - // even when idle so the flush deadline after this select! is - // honored within its 1s bound. flume's recv future is drop-safe - // on the Elapsed branch (no message consumed on timeout); the - // Ok(Ok(msg)) path below captures the message with no loss. + // Bounded recv (EverySec durability): wake at least every + // `idle_wait.current()` (50ms floor, escalates to 1s while + // idle — see `IdleWait`) even when idle so the flush deadline + // after this select! is honored within its 1s bound. flume's + // recv future is drop-safe on the Elapsed branch (no message + // consumed on timeout); the Ok(Ok(msg)) path below captures + // the message with no loss. r = tokio::time::timeout( - std::time::Duration::from_millis(200), + idle_wait.current(), rx.recv_async(), ) => { // On Elapsed (timeout) `r` is Err: skip and fall through to // the EverySec deadline check after this select!. match r { - Err(_) => {} + Err(_) => idle_wait.on_timeout(), // Channel disconnected — final sync + shut down. Ok(Err(_)) => { let _ = writer.flush().await; @@ -768,6 +950,9 @@ pub async fn per_shard_aof_writer_task( break; } Ok(Ok(first)) => { + // A message just arrived: reset to the fast floor + // (see TopLevel writer above for the full rationale). + idle_wait.on_message(); // Group commit: drain a bounded batch so ONE fsync // makes all framed records (`[u64 lsn][u32 len][RESP]`) // durable. @@ -882,6 +1067,9 @@ pub async fn per_shard_aof_writer_task( } else { // EverySec/No: the deadline check fsyncs; no // AppendSync waiters under everysec/no. + if fsync == FsyncPolicy::EverySec { + idle_wait.mark_pending(); + } BatchAck::Synced }; let _ = group_commit::ack_batch(&mut batch, verdict); @@ -980,6 +1168,7 @@ pub async fn per_shard_aof_writer_task( let _ = writer.flush().await; let _ = writer.get_ref().sync_data().await; last_fsync = Instant::now(); + idle_wait.clear_pending(); } } } @@ -1080,6 +1269,9 @@ pub async fn per_shard_aof_writer_task( let mut write_error = false; let mut _dbg_processed: u64 = 0; let _dbg_start = Instant::now(); + // Idle-adaptive channel-poll wake cadence (RSS/CPU wave 5, item B) — + // see `IdleWait` docs near the top of this file. + let mut idle_wait = IdleWait::new(); // Test-only fault injection: if MOON_TEST_AOF_FSYNC_FAIL=1 is set in // the environment at writer task startup, every AppendSync ack resolves // as FsyncFailed instead of Synced. Read once before the loop so there @@ -1093,12 +1285,21 @@ pub async fn per_shard_aof_writer_task( // the 1s fsync window never fires → data loss on kill. // recv_timeout so the EverySec proactive fsync fires even when no new // Appends arrive after a fold (or when the client stops writing). - let first = match rx.recv_timeout(std::time::Duration::from_millis(50)) { - Ok(m) => Some(m), - // Timeout: no message in the 50ms window. Fall through (None) to + // The wait starts at `idle_wait`'s fast floor (50ms, matching the + // old fixed cadence) and escalates while genuinely idle — see + // `IdleWait` docs. + let first = match rx.recv_timeout(idle_wait.current()) { + Ok(m) => { + idle_wait.on_message(); + Some(m) + } + // Timeout: no message in the window. Fall through (None) to // the EverySec proactive fsync below so queued-but-unfsynced // appends are durable within the everysec contract even when idle. - Err(flume::RecvTimeoutError::Timeout) => None, + Err(flume::RecvTimeoutError::Timeout) => { + idle_wait.on_timeout(); + None + } Err(flume::RecvTimeoutError::Disconnected) => { if !write_error { if let Err(e) = file.flush().and_then(|_| file.sync_data()) { @@ -1198,6 +1399,9 @@ pub async fn per_shard_aof_writer_task( } else { // EverySec/No: the proactive fsync below makes the batch // durable; no AppendSync waiters under everysec/no. + if fsync == FsyncPolicy::EverySec { + idle_wait.mark_pending(); + } BatchAck::Synced }; let _ = group_commit::ack_batch(&mut batch, verdict); @@ -1275,9 +1479,15 @@ pub async fn per_shard_aof_writer_task( } else { // Back-date last_fsync by 900ms: the proactive check // (threshold=1s) fires within the next 100ms, covering - // any appends that arrived after the drain above. + // any appends that arrived after the drain above. This + // IS a pending deadline — pin the idle wait at its + // floor (`on_message` already reset it for this + // iteration; `mark_pending` keeps it there) so + // escalation cannot push the next check out past the + // ≤150ms window the comment above promises. last_fsync = Instant::now() - std::time::Duration::from_millis(900); + idle_wait.mark_pending(); } } } @@ -1326,6 +1536,7 @@ pub async fn per_shard_aof_writer_task( } else { crate::admin::metrics_setup::record_aof_fsync(t.elapsed().as_micros() as u64); last_fsync = Instant::now(); + idle_wait.clear_pending(); } } } diff --git a/src/persistence/wal_v3/segment.rs b/src/persistence/wal_v3/segment.rs index ce333fd08..b4f9b3c02 100644 --- a/src/persistence/wal_v3/segment.rs +++ b/src/persistence/wal_v3/segment.rs @@ -43,6 +43,16 @@ pub const WAL_V3_HEADER_SIZE: usize = 64; /// Default segment size: 16MB. pub const DEFAULT_SEGMENT_SIZE: u64 = 16 * 1024 * 1024; +/// Default (initial) capacity of the in-memory write buffer. +const DEFAULT_WAL_BUF_CAPACITY: usize = 8192; + +/// Capacity above which the write buffer is shrunk back to +/// [`DEFAULT_WAL_BUF_CAPACITY`] once fully drained. A single oversized +/// record (e.g. a large FullPageImage) grows the buffer to fit it, and a +/// plain `Vec::clear()` never releases that capacity — one big write would +/// otherwise pin peak-sized memory for the lifetime of the writer. +const WAL_BUF_SHRINK_THRESHOLD: usize = DEFAULT_WAL_BUF_CAPACITY * 4; + /// Represents a single WAL v3 segment file. #[derive(Debug, Clone)] pub struct WalSegment { @@ -141,7 +151,7 @@ impl WalWriterV3 { segment_size, current_sequence: next_seq, current_file: None, - buf: Vec::with_capacity(8192), + buf: Vec::with_capacity(DEFAULT_WAL_BUF_CAPACITY), write_offset: 0, next_lsn, base_lsn: 0, @@ -182,6 +192,12 @@ impl WalWriterV3 { file.write_all(&self.buf)?; self.write_offset += self.buf.len() as u64; self.buf.clear(); + // Release peak capacity from an oversized record (e.g. a large + // FullPageImage) rather than pinning it for the writer's + // lifetime; `shrink_to` is a no-op below the target capacity. + if self.buf.capacity() > WAL_BUF_SHRINK_THRESHOLD { + self.buf.shrink_to(DEFAULT_WAL_BUF_CAPACITY); + } } Ok(()) @@ -423,6 +439,9 @@ impl WalWriterV3 { file.write_all(&self.buf)?; self.write_offset += self.buf.len() as u64; self.buf.clear(); + if self.buf.capacity() > WAL_BUF_SHRINK_THRESHOLD { + self.buf.shrink_to(DEFAULT_WAL_BUF_CAPACITY); + } } file.sync_data()?; } @@ -745,6 +764,48 @@ mod tests { assert_eq!(count, 3); } + #[test] + fn test_buffer_shrinks_after_flush_following_large_record() { + let tmp = tempfile::tempdir().unwrap(); + let wal_dir = tmp.path().join("wal"); + let mut writer = WalWriterV3::new(0, &wal_dir, DEFAULT_SEGMENT_SIZE).unwrap(); + + assert_eq!(writer.resident_bytes(), DEFAULT_WAL_BUF_CAPACITY); + + // A single oversized record forces the buffer well past the + // shrink threshold (4x default). + let huge_payload = vec![0xABu8; 100_000]; + writer.append(WalRecordType::Command, &huge_payload); + assert!(writer.resident_bytes() > WAL_BUF_SHRINK_THRESHOLD); + + writer.flush_sync().unwrap(); + + // The buffer must release the peak capacity back down to (near) + // default once fully drained, so one giant record doesn't pin + // memory forever. + assert!( + writer.resident_bytes() <= DEFAULT_WAL_BUF_CAPACITY, + "expected buffer to shrink back to default capacity, got {}", + writer.resident_bytes() + ); + } + + #[test] + fn test_buffer_does_not_shrink_below_threshold() { + let tmp = tempfile::tempdir().unwrap(); + let wal_dir = tmp.path().join("wal"); + let mut writer = WalWriterV3::new(0, &wal_dir, DEFAULT_SEGMENT_SIZE).unwrap(); + + // Small records that never exceed the shrink threshold should + // never trigger a reallocation cycle (a flush is a no-op sizing + // decision as long as capacity stays under the threshold). + for i in 0..10 { + writer.append(WalRecordType::Command, format!("SET k{i} v{i}").as_bytes()); + } + writer.flush_sync().unwrap(); + assert!(writer.resident_bytes() <= WAL_BUF_SHRINK_THRESHOLD); + } + #[test] fn test_writer_segment_rotation() { let tmp = tempfile::tempdir().unwrap(); diff --git a/src/protocol/parse.rs b/src/protocol/parse.rs index 03b7a1c4b..e62a53ec5 100644 --- a/src/protocol/parse.rs +++ b/src/protocol/parse.rs @@ -45,272 +45,6 @@ pub fn parse(buf: &mut BytesMut, config: &ParseConfig) -> Result, } } -/// Single-pass parser that produces zero-copy frames using Bytes::slice(). -/// Works on a frozen `Bytes` buffer for Arc-backed sub-slicing. -fn parse_single_frame_zc( - buf: &Bytes, - pos: &mut usize, - config: &ParseConfig, - depth: usize, -) -> Result { - if depth > config.max_array_depth { - return Err(ParseError::Invalid { - message: format!( - "array nesting depth {} exceeds maximum {}", - depth, config.max_array_depth - ), - offset: *pos, - }); - } - if *pos >= buf.len() { - return Err(ParseError::Incomplete); - } - let type_byte = buf[*pos]; - *pos += 1; - - match type_byte { - b'+' => { - let crlf = find_crlf(buf, *pos).ok_or(ParseError::Incomplete)?; - let line = buf.slice(*pos..crlf); - *pos = crlf + 2; - Ok(Frame::SimpleString(line)) - } - b'-' => { - let crlf = find_crlf(buf, *pos).ok_or(ParseError::Incomplete)?; - let line = buf.slice(*pos..crlf); - *pos = crlf + 2; - Ok(Frame::Error(line)) - } - b':' => { - let crlf = find_crlf(buf, *pos).ok_or(ParseError::Incomplete)?; - let line = &buf[*pos..crlf]; - let n = strict_atoi(line).ok_or_else(|| ParseError::Invalid { - message: format!("invalid integer: {:?}", String::from_utf8_lossy(line)), - offset: *pos, - })?; - *pos = crlf + 2; - Ok(Frame::Integer(n)) - } - b'$' => { - let len = read_decimal_zc(buf, pos)?; - if len == -1 { - return Ok(Frame::Null); - } - if len < 0 { - return Err(ParseError::Invalid { - message: format!("invalid bulk string length: {}", len), - offset: *pos, - }); - } - let len = len as usize; - if len > config.max_bulk_string_size { - return Err(ParseError::Invalid { - message: format!( - "bulk string size {} exceeds maximum {}", - len, config.max_bulk_string_size - ), - offset: *pos, - }); - } - let remaining = buf.len() - *pos; - if remaining < len + 2 { - return Err(ParseError::Incomplete); - } - // ZERO-COPY: Bytes::slice() does Arc refcount bump, no memcpy - let data = buf.slice(*pos..*pos + len); - *pos += len + 2; - Ok(Frame::BulkString(data)) - } - b'*' => { - let count = read_decimal_zc(buf, pos)?; - if count == -1 { - return Ok(Frame::Null); - } - if count < 0 { - return Err(ParseError::Invalid { - message: format!("invalid array count: {}", count), - offset: *pos, - }); - } - let count = count as usize; - if count > config.max_array_length { - return Err(ParseError::Invalid { - message: format!( - "array length {} exceeds maximum {}", - count, config.max_array_length - ), - offset: *pos, - }); - } - let mut items = FrameVec::with_capacity(count); - for _ in 0..count { - items.push(parse_single_frame_zc(buf, pos, config, depth + 1)?); - } - Ok(Frame::Array(items)) - } - b'%' => { - let count = read_decimal_zc(buf, pos)?; - if count == -1 { - return Ok(Frame::Null); - } - if count < 0 { - return Err(ParseError::Invalid { - message: "invalid map count".into(), - offset: *pos, - }); - } - let count = count as usize; - let mut entries = Vec::with_capacity(count); - for _ in 0..count { - let key = parse_single_frame_zc(buf, pos, config, depth + 1)?; - let val = parse_single_frame_zc(buf, pos, config, depth + 1)?; - entries.push((key, val)); - } - Ok(Frame::Map(entries)) - } - b'~' => { - let count = read_decimal_zc(buf, pos)?; - if count == -1 { - return Ok(Frame::Null); - } - if count < 0 { - return Err(ParseError::Invalid { - message: format!("invalid set count: {}", count), - offset: *pos, - }); - } - let count = count as usize; - if count > config.max_array_length { - return Err(ParseError::Invalid { - message: format!( - "set length {} exceeds maximum {}", - count, config.max_array_length - ), - offset: *pos, - }); - } - let mut items = FrameVec::with_capacity(count); - for _ in 0..count { - items.push(parse_single_frame_zc(buf, pos, config, depth + 1)?); - } - Ok(Frame::Set(items)) - } - b',' => { - let crlf = find_crlf(buf, *pos).ok_or(ParseError::Incomplete)?; - let line = &buf[*pos..crlf]; - let s = std::str::from_utf8(line).map_err(|_| ParseError::Invalid { - message: "invalid UTF-8 in double".into(), - offset: *pos, - })?; - let f = if s == "inf" { - f64::INFINITY - } else if s == "-inf" { - f64::NEG_INFINITY - } else { - s.parse::().map_err(|_| ParseError::Invalid { - message: format!("invalid double: {}", s), - offset: *pos, - })? - }; - *pos = crlf + 2; - Ok(Frame::Double(f)) - } - b'#' => { - if *pos + 2 >= buf.len() { - return Err(ParseError::Incomplete); - } - let val = buf[*pos]; - // Boolean format: #t\r\n or #f\r\n — exactly 1 char then CRLF - if (val != b't' && val != b'f') || buf[*pos + 1] != b'\r' || buf[*pos + 2] != b'\n' { - return Err(ParseError::Invalid { - message: format!("invalid boolean format at offset {}", *pos), - offset: *pos, - }); - } - *pos += 3; - Ok(Frame::Boolean(val == b't')) - } - b'_' => { - // RESP3 Null: `_\r\n` — verify CRLF immediately follows type byte - if *pos + 1 >= buf.len() { - return Err(ParseError::Incomplete); - } - if buf[*pos] != b'\r' || buf[*pos + 1] != b'\n' { - return Err(ParseError::Invalid { - message: format!( - "RESP3 null has trailing data before CRLF at offset {}", - *pos - ), - offset: *pos, - }); - } - *pos += 2; - Ok(Frame::Null) - } - b'=' => { - let len = read_decimal_zc(buf, pos)? as usize; - let remaining = buf.len() - *pos; - if remaining < len + 2 { - return Err(ParseError::Incomplete); - } - let payload = &buf[*pos..*pos + len]; - let encoding = Bytes::copy_from_slice(&payload[..3]); - let data = buf.slice(*pos + 4..*pos + len); - *pos += len + 2; - Ok(Frame::VerbatimString { encoding, data }) - } - b'(' => { - let crlf = find_crlf(buf, *pos).ok_or(ParseError::Incomplete)?; - let line = buf.slice(*pos..crlf); - *pos = crlf + 2; - Ok(Frame::BigNumber(line)) - } - b'>' => { - let count = read_decimal_zc(buf, pos)?; - if count == -1 { - return Ok(Frame::Null); - } - if count < 0 { - return Err(ParseError::Invalid { - message: format!("invalid push count: {}", count), - offset: *pos, - }); - } - let count = count as usize; - if count > config.max_array_length { - return Err(ParseError::Invalid { - message: format!( - "push length {} exceeds maximum {}", - count, config.max_array_length - ), - offset: *pos, - }); - } - let mut items = FrameVec::with_capacity(count); - for _ in 0..count { - items.push(parse_single_frame_zc(buf, pos, config, depth + 1)?); - } - Ok(Frame::Push(items)) - } - other => Err(ParseError::Invalid { - message: format!("unknown type byte: 0x{:02x}", other), - offset: *pos - 1, - }), - } -} - -/// Read a decimal integer from the frozen buffer (for zero-copy parser). -fn read_decimal_zc(buf: &Bytes, pos: &mut usize) -> Result { - let crlf = find_crlf(buf, *pos).ok_or(ParseError::Incomplete)?; - let line = &buf[*pos..crlf]; - let n = strict_atoi(line).ok_or_else(|| ParseError::Invalid { - message: format!("invalid decimal: {:?}", String::from_utf8_lossy(line)), - offset: *pos, - })?; - *pos = crlf + 2; - Ok(n) -} - /// Zero-copy frame extraction from a frozen `Bytes` buffer. /// Called AFTER validation succeeds, so all CRLF/atoi lookups should succeed. /// Uses `bytes.slice(start..end)` for zero-copy sub-slicing (Arc refcount bump only). diff --git a/src/scripting/cache.rs b/src/scripting/cache.rs index 9f2ed430e..9a0082838 100644 --- a/src/scripting/cache.rs +++ b/src/scripting/cache.rs @@ -35,6 +35,16 @@ impl ScriptCache { pub fn len(&self) -> usize { self.scripts.len() } + + /// Approximate resident bytes held by cached script bodies (C4 wave-5 + /// hygiene): the sum of each entry's hex-SHA1 key length plus its + /// source byte length. This is an estimate (it excludes `HashMap`/ + /// `String`/`Bytes` allocator bookkeeping overhead) intended for + /// observability only -- the cache itself remains unbounded, matching + /// Redis semantics (`SCRIPT FLUSH` is the only eviction path). + pub fn resident_bytes(&self) -> usize { + self.scripts.iter().map(|(k, v)| k.len() + v.len()).sum() + } } #[cfg(test)] @@ -72,6 +82,32 @@ mod tests { assert_eq!(cache.len(), 0); } + #[test] + fn test_resident_bytes_empty_cache_is_zero() { + let cache = ScriptCache::new(); + assert_eq!(cache.resident_bytes(), 0); + } + + #[test] + fn test_resident_bytes_grows_with_entries_and_shrinks_on_flush() { + let mut cache = ScriptCache::new(); + let sha1 = cache.load(Bytes::from_static(b"return 1")); + let after_one = cache.resident_bytes(); + // 40-byte hex key + 8-byte body. + assert_eq!(after_one, sha1.len() + 8); + + let sha2 = cache.load(Bytes::from_static(b"return 'a much longer script body'")); + let after_two = cache.resident_bytes(); + assert!(after_two > after_one); + assert_eq!( + after_two, + sha1.len() + 8 + sha2.len() + "return 'a much longer script body'".len() + ); + + cache.flush(); + assert_eq!(cache.resident_bytes(), 0); + } + #[test] fn test_sha1_deterministic() { let mut cache = ScriptCache::new(); diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index 1fff171c0..be9f2e300 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -1624,6 +1624,7 @@ impl super::Shard { &page_cache, &mut next_file_id, &mut wal_v3_writer, + &script_cache_rc, &spill_file_id, ); @@ -2127,6 +2128,7 @@ impl super::Shard { &page_cache, &mut next_file_id, &mut wal_v3_writer, + &script_cache_rc, &spill_file_id, ); // MQ trigger check: fire debounced triggers diff --git a/src/shard/mq_exec.rs b/src/shard/mq_exec.rs index bae203422..46af3b72c 100644 --- a/src/shard/mq_exec.rs +++ b/src/shard/mq_exec.rs @@ -557,6 +557,7 @@ mod tests { vector: AtomicUsize::new(0), text: AtomicUsize::new(0), graph: AtomicUsize::new(0), + lua: AtomicUsize::new(0), }), }) } diff --git a/src/shard/persistence_tick.rs b/src/shard/persistence_tick.rs index ebf1403a3..4c93ae3a5 100644 --- a/src/shard/persistence_tick.rs +++ b/src/shard/persistence_tick.rs @@ -339,6 +339,7 @@ pub(crate) fn run_eviction_tick( page_cache: &Option, next_file_id: &mut u64, wal_v3_writer: &mut Option, + script_cache: &std::rc::Rc>, spill_file_id: &std::rc::Rc>, ) { if let Some(spill_t) = spill_thread { @@ -387,6 +388,12 @@ pub(crate) fn run_eviction_tick( } #[cfg(not(feature = "graph"))] s.store_memory.graph.store(0, Ordering::Relaxed); + // C4 (wave-5 hygiene): publish the shard's Lua script-cache byte + // estimate alongside vector/text/graph so INFO/MEMORY DOCTOR and + // Prometheus stop reporting a permanent zero for Lua memory. + s.store_memory + .lua + .store(script_cache.borrow().resident_bytes(), Ordering::Relaxed); }); if server_config.disk_offload_enabled() diff --git a/src/shard/shared_databases.rs b/src/shard/shared_databases.rs index b85d8655b..8797d81cd 100644 --- a/src/shard/shared_databases.rs +++ b/src/shard/shared_databases.rs @@ -2,6 +2,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use parking_lot::{Mutex, MutexGuard}; +use smallvec::SmallVec; use crate::storage::Database; use crate::workspace::wal::{decode_workspace_create, decode_workspace_drop}; @@ -23,6 +24,10 @@ pub struct ShardStoreMemory { pub text: AtomicUsize, /// Resident bytes of GraphStore CSR segments. pub graph: AtomicUsize, + /// Approximate resident bytes of the shard's Lua `ScriptCache` (C4 wave-5 + /// hygiene). The cache itself stays unbounded (Redis parity -- `SCRIPT + /// FLUSH` is the only eviction path); this is observability only. + pub lua: AtomicUsize, } /// Shared infrastructure handle — the residual cross-shard state after M5. @@ -102,6 +107,7 @@ impl ShardDatabases { vector: AtomicUsize::new(0), text: AtomicUsize::new(0), graph: AtomicUsize::new(0), + lua: AtomicUsize::new(0), }) }) .collect::>() @@ -225,7 +231,10 @@ impl ShardDatabases { self.elastic_budgets[shard_id].store(0, Ordering::Relaxed); return 0; } - let used: Vec = self + // SmallVec: most deployments run <=16 shards, so this 100ms-tick + // snapshot stays fully on the stack; only larger shard counts spill + // to a single heap allocation (still one per call, same as before). + let used: SmallVec<[usize; 16]> = self .memory_per_shard .iter() .map(|a| a.load(Ordering::Relaxed)) @@ -748,6 +757,25 @@ mod tests { assert_eq!(shared.recompute_elastic_budget(1, &rt), 100); } + #[test] + fn recompute_elastic_budget_correct_beyond_smallvec_inline_capacity() { + // The per-call `used` snapshot is a `SmallVec<[usize; 16]>` — pin + // correctness both inline (<=16 shards, covered above) and once + // spilled to the heap (>16 shards) so the container swap never + // silently truncates or reorders shard readings. + const N: usize = 20; + let shared = new_shared(N, 1); + let rt = rt_config(N * 100, N); // base = 100 per shard + shared.publish_memory(0, 150); // hot + for i in 1..N { + shared.publish_memory(i, 10); + } + // Hot shard borrows (100-10)*19 = 1710 -> budget 1810. + assert_eq!(shared.recompute_elastic_budget(0, &rt), 1810); + // An idle shard keeps base. + assert_eq!(shared.recompute_elastic_budget(5, &rt), 100); + } + #[test] fn recompute_elastic_budget_disabled_for_single_shard_or_unlimited() { let shared = new_shared(1, 1); diff --git a/src/shard/slice.rs b/src/shard/slice.rs index 27694f06a..48a32b3cd 100644 --- a/src/shard/slice.rs +++ b/src/shard/slice.rs @@ -604,6 +604,7 @@ pub(crate) mod test_support { vector: AtomicUsize::new(0), text: AtomicUsize::new(0), graph: AtomicUsize::new(0), + lua: AtomicUsize::new(0), }), } } diff --git a/src/text/posting.rs b/src/text/posting.rs index d8a24893c..0da234dff 100644 --- a/src/text/posting.rs +++ b/src/text/posting.rs @@ -215,6 +215,21 @@ impl PostingStore { } } removed.push((term_id, old_tf)); + // 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 + // answerable without a fresh insert). But once the LAST doc + // leaves, the entry's Vec buffers have no reason to keep + // capacity sized for a document count of zero — release it. + // Reallocation on the next occurrence of this term is a + // one-time, bounded cost; the alternative is holding peak + // capacity forever for a term that may never recur. + if posting.doc_ids.is_empty() { + posting.term_freqs.shrink_to_fit(); + if let Some(pos_list) = &mut posting.positions { + pos_list.shrink_to_fit(); + } + } } } removed @@ -268,3 +283,89 @@ impl PostingStore { total } } + +#[cfg(test)] +mod tests { + use super::*; + + /// RSS/CPU wave 5 (item A hygiene follow-up): a posting's `term_freqs` + /// (and `positions`, when tracked) grow to the peak document count ever + /// seen for that term. The `postings` HashMap entry is intentionally + /// kept forever once created (existing contract — see `remove_doc` doc + /// comment), but the per-entry `Vec` buffers must not hold onto peak + /// capacity once every document has been removed. + #[test] + fn remove_doc_shrinks_now_empty_posting_capacity() { + let mut store = PostingStore::new(); + for doc_id in 0..500u32 { + store.add_term_occurrence(7, doc_id, None); + } + let peak_cap = store.get_posting(7).unwrap().term_freqs.capacity(); + assert!(peak_cap >= 500, "expected growth to >=500, got {peak_cap}"); + + for doc_id in 0..500u32 { + store.remove_doc(doc_id); + } + + // Entry survives (existing contract) ... + let posting = store.get_posting(7).expect("entry must survive removal"); + assert_eq!(posting.doc_ids.len(), 0); + assert_eq!(posting.tf(0), 0); + // ... but its buffer no longer holds peak capacity. + assert!( + posting.term_freqs.capacity() < peak_cap, + "expected shrink after last doc removed: peak={peak_cap} still={}", + posting.term_freqs.capacity() + ); + } + + /// Same shrink must apply to the `positions` buffer when position + /// tracking is enabled for the term. + #[test] + fn remove_doc_shrinks_now_empty_posting_positions_capacity() { + let mut store = PostingStore::new(); + for doc_id in 0..300u32 { + store.add_term_occurrence(3, doc_id, Some(vec![doc_id])); + } + let peak_cap = store + .get_posting(3) + .unwrap() + .positions + .as_ref() + .unwrap() + .capacity(); + assert!(peak_cap >= 300); + + for doc_id in 0..300u32 { + store.remove_doc(doc_id); + } + + let posting = store.get_posting(3).unwrap(); + let pos_cap = posting.positions.as_ref().unwrap().capacity(); + assert!( + pos_cap < peak_cap, + "expected positions shrink: peak={peak_cap} still={pos_cap}" + ); + } + + /// A term that still has live documents after a removal must not be + /// touched by the shrink (only a fully-emptied posting shrinks). + #[test] + fn remove_doc_does_not_shrink_still_live_posting() { + let mut store = PostingStore::new(); + for doc_id in 0..50u32 { + store.add_term_occurrence(1, doc_id, None); + } + let cap_before = store.get_posting(1).unwrap().term_freqs.capacity(); + + store.remove_doc(0); // one doc gone, 49 remain live + + let posting = store.get_posting(1).unwrap(); + assert_eq!(posting.doc_ids.len(), 49); + assert_eq!( + posting.term_freqs.capacity(), + cap_before, + "must not shrink while the posting still has live docs" + ); + } +} diff --git a/src/vector/persistence/segment_io.rs b/src/vector/persistence/segment_io.rs index df6b47740..d81867840 100644 --- a/src/vector/persistence/segment_io.rs +++ b/src/vector/persistence/segment_io.rs @@ -21,6 +21,7 @@ use crate::persistence::fsync::{fsync_directory, fsync_file}; use crate::vector::aligned_buffer::AlignedBuffer; use crate::vector::hnsw::graph::HnswGraph; use crate::vector::segment::immutable::{ImmutableSegment, MvccHeader}; +use crate::vector::segment::raw_f16_store::RawF16Store; use crate::vector::turbo_quant::collection::{CollectionMetadata, QuantizationConfig}; use crate::vector::types::DistanceMetric; @@ -550,26 +551,31 @@ pub fn read_immutable_segment( let sub_sign_bpv = (meta.padded_dimension as usize + 7) / 8; // 6b. raw_f16.bin — optional exact-rerank sidecar (HQ-1). Missing file - // (pre-sidecar segments) or a size mismatch → no sidecar; search falls - // back to quantized ADC distances. - let raw_f16: Option> = match fs::read(seg_dir.join("raw_f16.bin")) { - Ok(bytes) if bytes.len() == mvcc.len() * dim * 2 => Some( - bytes - .chunks_exact(2) - .map(|c| u16::from_le_bytes([c[0], c[1]])) - .collect(), - ), - Ok(bytes) => { - tracing::warn!( - "segment-{segment_id}: raw_f16.bin has {} bytes, expected {} — \ - ignoring sidecar (search degrades to quantized distances)", - bytes.len(), - mvcc.len() * dim * 2 - ); - None + // (pre-sidecar segments), an open/read error, or a size mismatch → no + // sidecar; search falls back to quantized ADC distances. Reload + // memory-maps the file instead of buffering a second heap copy (item A, + // RSS/CPU wave 5) — see `raw_f16_store` module docs for the mmap + // soundness contract this relies on. + let expected_halves = mvcc.len() * dim; + let raw_f16_path = seg_dir.join("raw_f16.bin"); + // missing file (pre-sidecar segment) or open/mmap error -> None, same as + // a size mismatch. + let raw_f16: Option = + RawF16Store::map_file(&raw_f16_path, expected_halves).unwrap_or_default(); + if raw_f16.is_none() { + // Distinguish "missing file" (expected, silent) from "present but + // wrong size" (corruption — worth a loud warning) without doing a + // second `map_file` call: a cheap metadata probe is enough. + if let Ok(actual_len) = fs::metadata(&raw_f16_path).map(|m| m.len()) { + if actual_len as usize != expected_halves * 2 { + tracing::warn!( + "segment-{segment_id}: raw_f16.bin has {actual_len} bytes, expected {} — \ + ignoring sidecar (search degrades to quantized distances)", + expected_halves * 2 + ); + } } - Err(_) => None, - }; + } let segment = ImmutableSegment::new( graph, @@ -584,7 +590,7 @@ pub fn read_immutable_segment( meta.live_count, meta.total_count, ) - .with_raw_f16(raw_f16) + .with_raw_f16_store(raw_f16) // R6: restore the compact-time estimate verbatim — never re-run the // estimator on the load path (it needs the raw sidecar + a full ladder // walk; that cost belongs on the compaction thread only). @@ -852,6 +858,90 @@ mod tests { assert_eq!(restored.total_count(), segment.total_count()); } + /// Item A (RSS/CPU wave 5): a segment reloaded from disk must back its + /// exact-rerank sidecar with a memory map, not a second heap `Vec` — + /// and the mapped view must decode byte-identical halves and produce + /// identical `search()` results to the original heap-owned segment. + #[test] + fn test_reload_raw_f16_sidecar_uses_mmap_and_matches_owned_rerank() { + let n = 40; + let dim = 64; + let (segment, collection) = build_test_segment(n, dim); + + // Synthetic BFS-ordered sidecar (content doesn't need to match the TQ + // codes for this test — only self-consistency between the Owned and + // Mapped views of the SAME bytes matters). + let mut raw_f16_bfs = vec![0u16; n * dim]; + for bfs in 0..n { + let orig_id = segment.graph().to_original(bfs as u32); + let mut v = lcg_f32(dim, orig_id ^ 0xABCD_1234); + normalize(&mut v); + let mut halves = Vec::new(); + crate::vector::f16::encode_f16_slice(&v, &mut halves); + raw_f16_bfs[bfs * dim..(bfs + 1) * dim].copy_from_slice(&halves); + } + let segment = segment.with_raw_f16(Some(raw_f16_bfs)); + assert!( + !segment.raw_f16_is_mapped(), + "freshly-built segment must stay heap-owned" + ); + + let tmp = tempfile::tempdir().unwrap(); + write_immutable_segment(tmp.path(), 1, &segment, &collection).unwrap(); + let (restored, _restored_col) = read_immutable_segment(tmp.path(), 1).unwrap(); + + assert!( + restored.raw_f16_is_mapped(), + "reloaded segment must back its raw_f16 sidecar with a memory map" + ); + + // Round-trip byte fidelity: mapped view decodes to the exact same + // halves the in-memory (Owned) segment holds. + assert_eq!(restored.raw_f16().unwrap(), segment.raw_f16().unwrap()); + + // Rerank parity: identical sidecar bytes through Owned vs Mapped + // storage must produce identical search() output for the same query. + let mut query = lcg_f32(dim, 999_999); + normalize(&mut query); + let padded = collection.padded_dimension; + let mut scratch_owned = + crate::vector::hnsw::search::SearchScratch::new(segment.graph().num_nodes(), padded); + let mut scratch_mapped = + crate::vector::hnsw::search::SearchScratch::new(restored.graph().num_nodes(), padded); + let owned_results = segment.search(&query, 5, 64, &mut scratch_owned); + let mapped_results = restored.search(&query, 5, 64, &mut scratch_mapped); + + assert_eq!(owned_results.len(), mapped_results.len()); + assert!(!owned_results.is_empty()); + for (a, b) in owned_results.iter().zip(mapped_results.iter()) { + assert_eq!(a.id.0, b.id.0); + assert!( + (a.distance - b.distance).abs() < 1e-4, + "distance mismatch: {} vs {}", + a.distance, + b.distance + ); + } + + // Memory accounting must reflect the mmap win: the mapped sidecar is + // kernel page cache, not pinned heap, so the reloaded segment must + // report at least the sidecar's bytes less than the heap-owned + // original (other components may also differ slightly across a + // reload; the exact Owned-vs-Mapped byte accounting is pinned by + // raw_f16_store's own unit tests). Counting mapped pages as resident + // would feed the elastic memory budget / eviction pipeline numbers + // as if the RSS win never happened. + let sidecar_bytes = n * dim * std::mem::size_of::(); + assert!( + segment.resident_bytes() >= restored.resident_bytes() + sidecar_bytes, + "mapped sidecar must not count toward resident_bytes \ + (owned={} mapped={} sidecar={})", + segment.resident_bytes(), + restored.resident_bytes(), + sidecar_bytes + ); + } + #[test] fn test_roundtrip_search_works() { let (segment, collection) = build_test_segment(50, 64); diff --git a/src/vector/segment/immutable.rs b/src/vector/segment/immutable.rs index 794c12ef5..97812c681 100644 --- a/src/vector/segment/immutable.rs +++ b/src/vector/segment/immutable.rs @@ -21,6 +21,7 @@ use crate::vector::hnsw::search::{ }; #[allow(unused_imports)] use crate::vector::hnsw::search_sq::hnsw_search_f32; +use crate::vector::segment::raw_f16_store::RawF16Store; use crate::vector::turbo_quant::collection::{CollectionMetadata, QuantizationConfig}; use crate::vector::turbo_quant::inner_product::{prepare_query_prod, score_l2_prod}; use crate::vector::turbo_quant::sq8::{decode_sq8, sq8_params}; @@ -92,7 +93,13 @@ pub struct ImmutableSegment { /// before top-k truncation — the returned distances are then true metric /// values to f16 tolerance instead of quantized ADC estimates. `None` for /// segments built without raw vectors (pre-sidecar disk segments). - raw_f16: Option>, + /// + /// Backed by [`RawF16Store`]: freshly-built segments own a `Vec` + /// (`Owned`); segments reloaded from disk memory-map `raw_f16.bin` + /// instead (`Mapped`) so RSS only grows for pages the rerank path + /// actually touches. See `raw_f16_store` module docs for the mmap + /// soundness contract. + raw_f16: Option, /// Compact-time adaptive-ef estimate (AE-1): the smallest ladder ef at /// which this segment's OWN sampled queries reach the target recall @@ -143,9 +150,12 @@ impl ImmutableSegment { } } - /// Attach the exact-rerank sidecar (HQ-1): BFS-ordered f16 copies of the - /// original vectors, `dimension` halves per entry. Builder-style so the - /// many `new()` call sites without raw vectors stay untouched. + /// Attach the exact-rerank sidecar (HQ-1) from an owned buffer: + /// BFS-ordered f16 copies of the original vectors, `dimension` halves + /// per entry. Builder-style so the many `new()` call sites without raw + /// vectors stay untouched. Used by compaction/merge, which always + /// construct a fresh owned buffer — for the disk-reload path (which + /// wants to memory-map instead), see [`Self::with_raw_f16_store`]. #[must_use] pub fn with_raw_f16(mut self, raw_f16: Option>) -> Self { if let Some(ref buf) = raw_f16 { @@ -155,15 +165,40 @@ impl ImmutableSegment { "raw_f16 sidecar must hold dimension halves per BFS entry" ); } - self.raw_f16 = raw_f16; + self.raw_f16 = raw_f16.map(RawF16Store::Owned); + self + } + + /// Attach the exact-rerank sidecar (HQ-1) from a pre-built + /// [`RawF16Store`] — used by `segment_io::read_immutable_segment` to + /// attach a memory-mapped sidecar without materializing a second heap + /// copy. See [`Self::with_raw_f16`] for the owned-buffer variant. + #[must_use] + pub fn with_raw_f16_store(mut self, store: Option) -> Self { + if let Some(ref s) = store { + debug_assert_eq!( + s.len(), + self.mvcc.len() * self.collection_meta.dimension as usize, + "raw_f16 sidecar must hold dimension halves per BFS entry" + ); + } + self.raw_f16 = store; self } /// The exact-rerank sidecar, if this segment carries one (BFS-ordered, /// `dimension` u16 halves per entry). Used by segment persistence and - /// GraphUnion merge to propagate the sidecar. + /// GraphUnion merge to propagate the sidecar. Zero-copy in both the + /// heap-owned and memory-mapped case. pub fn raw_f16(&self) -> Option<&[u16]> { - self.raw_f16.as_deref() + self.raw_f16.as_ref().map(RawF16Store::as_slice) + } + + /// `true` when the exact-rerank sidecar is backed by a memory map + /// (segment reloaded from disk) rather than a heap `Vec` (freshly built + /// segment). Exposed for tests/diagnostics only — not on any hot path. + pub fn raw_f16_is_mapped(&self) -> bool { + matches!(&self.raw_f16, Some(s) if s.is_mapped()) } /// The compact-time adaptive-ef estimate for this segment (AE-1), if one @@ -242,7 +277,7 @@ impl ImmutableSegment { const ADAPTIVE_EF_EPSILON: f32 = 0.005; const ADAPTIVE_EF_LADDER: &[usize] = &[24, 32, 48, 64, 96, 128, 192, 256]; - let raw = self.raw_f16.as_deref()?; + let raw = self.raw_f16()?; let dim = self.collection_meta.dimension as usize; let n = self.mvcc.len(); if dim == 0 || n < ADAPTIVE_EF_K * 8 || raw.len() < n * dim { @@ -394,7 +429,7 @@ impl ImmutableSegment { /// outside a 4× ADC oversample is rare; re-scoring the full ef-wide beam /// costs ~ef·dim f16 decodes per segment for negligible recall beyond that. fn rerank_exact(&self, candidates: &mut SmallVec<[SearchResult; 32]>, query: &[f32], k: usize) { - let Some(raw) = self.raw_f16.as_deref() else { + let Some(raw) = self.raw_f16() else { return; }; if candidates.is_empty() { @@ -855,10 +890,9 @@ impl ImmutableSegment { let norms = self.residual_norms.len() * std::mem::size_of::(); let sub = self.sub_centroid_signs.len(); let mvcc = self.mvcc.len() * std::mem::size_of::(); - let sidecar = self - .raw_f16 - .as_ref() - .map_or(0, |v| v.len() * std::mem::size_of::()); + // Mapped sidecars report 0: their pages are kernel page cache, not + // pinned heap — see RawF16Store::resident_bytes. + let sidecar = self.raw_f16.as_ref().map_or(0, RawF16Store::resident_bytes); graph + tq + qjl + norms + sub + mvcc + sidecar } diff --git a/src/vector/segment/mod.rs b/src/vector/segment/mod.rs index 9ba26c74d..2a947bfe7 100644 --- a/src/vector/segment/mod.rs +++ b/src/vector/segment/mod.rs @@ -3,6 +3,7 @@ pub mod holder; pub mod immutable; pub mod ivf; pub mod mutable; +pub mod raw_f16_store; pub use compaction::{ CompactionError, MergeMode, MergeStats, compact, merge_immutable, needs_vacuum, @@ -11,3 +12,4 @@ pub use holder::{SegmentHolder, SegmentList}; pub use immutable::ImmutableSegment; pub use ivf::IvfSegment; pub use mutable::MutableSegment; +pub use raw_f16_store::RawF16Store; diff --git a/src/vector/segment/raw_f16_store.rs b/src/vector/segment/raw_f16_store.rs new file mode 100644 index 000000000..9a248af41 --- /dev/null +++ b/src/vector/segment/raw_f16_store.rs @@ -0,0 +1,224 @@ +//! Storage backend for the exact-rerank f16 sidecar (HQ-1). +//! +//! An [`ImmutableSegment`](super::immutable::ImmutableSegment) built fresh by +//! compaction/merge already owns its `raw_f16` buffer as a `Vec` — no +//! extra work needed there. A segment **reloaded from disk** used to +//! re-materialize that same buffer with a fresh `fs::read` + decode, which +//! doubles resident memory for the sidecar on every warm start / segment +//! promotion. [`RawF16Store::Mapped`] instead memory-maps `raw_f16.bin` and +//! hands out a zero-copy `&[u16]` view backed by the kernel page cache — RSS +//! only grows for pages the rerank path actually touches. +//! +//! # The seal contract this relies on +//! +//! `raw_f16.bin` lives inside a `segment-{id}/` directory written by +//! [`crate::vector::persistence::segment_io::write_segment_files`] and made +//! visible only via the staged-directory -> final-directory atomic rename in +//! [`crate::vector::persistence::segment_io::write_immutable_segment_staged`]. +//! After that rename: +//! - No moon code ever opens `raw_f16.bin` for writing again — the file is +//! part of an immutable segment. +//! - The only later operation on the containing directory is a whole-directory +//! removal during GC (`run_snapshot_job` / `sweep_orphans_from_disk`) once +//! the segment is superseded by a merge. `unlink`/`remove_dir_all` on POSIX +//! does not invalidate an already-established `mmap` of a file that still +//! has an open mapping — the kernel keeps the inode's pages alive until the +//! last reference (including mmaps) drops. So a concurrent GC racing an +//! in-flight reader is safe, not just the common case. +//! +//! This mirrors the existing warm-segment seal contract documented in +//! [`crate::vector::persistence::sealed_mmap`] and the CSR mmap loader in +//! [`crate::graph::csr::mmap`] — same invariant, applied to a new file. + +use std::fs::File; +use std::io; +use std::path::Path; + +use memmap2::Mmap; + +/// Backing storage for an immutable segment's exact-rerank f16 sidecar. +pub enum RawF16Store { + /// Freshly-built segment (compaction/merge): the sidecar buffer it + /// already owns in memory. + Owned(Vec), + /// Segment reloaded from disk: zero-copy view into `raw_f16.bin`'s page + /// cache. `len` is the number of `u16` halves (== `mmap.len() / 2`). + Mapped { mmap: Mmap, len: usize }, +} + +impl RawF16Store { + /// Memory-map `path` (a sealed `raw_f16.bin`, see module docs) read-only. + /// + /// Returns `Ok(None)` when the file's byte length does not match + /// `expected_halves * 2` — the caller treats this exactly like a missing + /// file (size mismatch means a corrupt/truncated sidecar; search falls + /// back to quantized ADC distances rather than trusting bad data). + /// + /// # Errors + /// + /// Propagates `File::open`/`metadata`/`mmap` I/O errors (including + /// "not found" for pre-sidecar segments) — callers map those to "no + /// sidecar" too. + pub fn map_file(path: &Path, expected_halves: usize) -> io::Result> { + let file = File::open(path)?; + let len_bytes = file.metadata()?.len() as usize; + if len_bytes != expected_halves * 2 { + return Ok(None); + } + if len_bytes == 0 { + // memmap2 refuses to map a zero-length file; an empty sidecar + // (e.g. a segment with zero live vectors) needs no mapping. + return Ok(Some(Self::Owned(Vec::new()))); + } + // `path` is `raw_f16.bin` inside a `segment-{id}` directory. It is + // written exactly once by `write_segment_files` and only made + // visible via the staged-dir -> final-dir atomic rename in + // `write_immutable_segment_staged` (see module docs for the full + // seal contract, including why a racing GC removal is also safe). + // SAFETY: read-only map of a sealed, write-once file (see above). + let mmap = unsafe { Mmap::map(&file) }?; + Ok(Some(Self::Mapped { + mmap, + len: expected_halves, + })) + } + + /// Number of `u16` halves stored. + pub fn len(&self) -> usize { + match self { + Self::Owned(v) => v.len(), + Self::Mapped { len, .. } => *len, + } + } + + /// `true` when this store holds no halves. + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// `true` when this store is backed by a memory map rather than a heap + /// buffer. Exposed for tests/diagnostics — not on any hot path. + pub fn is_mapped(&self) -> bool { + matches!(self, Self::Mapped { .. }) + } + + /// Heap bytes this store pins: the full buffer when `Owned`, `0` when + /// `Mapped` — mapped pages are kernel page cache, reclaimable under + /// memory pressure, so counting them as resident would make the elastic + /// memory budget / eviction pipeline behave as if the mmap RSS win never + /// happened for reloaded segments. + pub fn resident_bytes(&self) -> usize { + match self { + Self::Owned(v) => v.len() * std::mem::size_of::(), + Self::Mapped { .. } => 0, + } + } + + /// Zero-copy view of the sidecar as `&[u16]`. + pub fn as_slice(&self) -> &[u16] { + match self { + Self::Owned(v) => v, + Self::Mapped { mmap, len } => { + // `mmap` maps exactly `len * 2` bytes of a file written as + // `len` little-endian `u16` halves (see + // `write_segment_files`'s `h.to_le_bytes()` loop in + // segment_io.rs). Moon's only target architectures + // (x86_64, aarch64 — see CLAUDE.md "Target Platform") are + // little-endian, so a native `u16` read reproduces exactly + // the value the writer encoded; there is no target where + // this would silently byte-swap. This mirrors the identical + // `from_raw_parts` reinterpret pattern already used for + // mmap'd CSR arrays in `crate::graph::csr::mmap`. + // SAFETY: kernel mappings are page-aligned (satisfies u16's + // 2-byte alignment); length verified at map time (above). + unsafe { std::slice::from_raw_parts(mmap.as_ptr().cast::(), *len) } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + + fn write_halves(path: &Path, halves: &[u16]) { + let mut f = File::create(path).unwrap(); + for h in halves { + f.write_all(&h.to_le_bytes()).unwrap(); + } + f.sync_all().unwrap(); + } + + #[test] + fn map_file_roundtrips_halves() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raw_f16.bin"); + let halves: Vec = (0..256u16).collect(); + write_halves(&path, &halves); + + let store = RawF16Store::map_file(&path, halves.len()).unwrap().unwrap(); + assert!(store.is_mapped()); + assert_eq!(store.len(), halves.len()); + assert_eq!(store.as_slice(), halves.as_slice()); + } + + #[test] + fn map_file_rejects_size_mismatch() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raw_f16.bin"); + write_halves(&path, &[1, 2, 3]); + + // Claiming 10 halves (20 bytes) against an actual 6-byte file. + let store = RawF16Store::map_file(&path, 10).unwrap(); + assert!(store.is_none()); + } + + #[test] + fn map_file_missing_file_errors() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("does-not-exist.bin"); + assert!(RawF16Store::map_file(&path, 4).is_err()); + } + + #[test] + fn map_file_empty_expected_is_owned_empty() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raw_f16.bin"); + write_halves(&path, &[]); + + let store = RawF16Store::map_file(&path, 0).unwrap().unwrap(); + assert!(!store.is_mapped()); + assert!(store.is_empty()); + assert_eq!(store.as_slice(), &[] as &[u16]); + } + + #[test] + fn owned_store_is_not_mapped() { + let store = RawF16Store::Owned(vec![7, 8, 9]); + assert!(!store.is_mapped()); + assert_eq!(store.as_slice(), &[7u16, 8, 9]); + } + + /// Resident accounting: an Owned store pins its full buffer on the heap; + /// a Mapped store pins nothing (kernel page cache, reclaimable) — the + /// elastic memory budget / eviction pipeline must see the mmap RSS win, + /// not pretend the bytes are still resident. + #[test] + fn resident_bytes_owned_full_mapped_zero() { + let owned = RawF16Store::Owned(vec![0u16; 100]); + assert_eq!(owned.resident_bytes(), 200); + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raw_f16.bin"); + let halves: Vec = (0..100u16).collect(); + write_halves(&path, &halves); + let mapped = RawF16Store::map_file(&path, halves.len()).unwrap().unwrap(); + assert!(mapped.is_mapped()); + assert_eq!( + mapped.resident_bytes(), + 0, + "mapped sidecar pages are page cache, not pinned heap" + ); + } +}