From 177c1ee5629f61686e29f51bc45a76a62dd6c65e Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 23:06:47 +0700 Subject: [PATCH 01/11] perf(vector): mmap exact-rerank f16 sidecar on segment reload (RSS/CPU wave 5, item A) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A segment reloaded from disk (`read_immutable_segment`) used to `fs::read` the entire `raw_f16.bin` exact-rerank sidecar (HQ-1) into a fresh `Vec`, on top of the identical bytes the segment already wrote when it was compacted/merged. That doubles resident vector memory for any reload-heavy deployment (warm restarts, segment promotion) — the biggest single RSS win identified in the RSS/CPU remediation review. Introduce `RawF16Store` (`src/vector/segment/raw_f16_store.rs`): an enum over `Owned(Vec)` (freshly-built segments keep their existing owned buffer, zero behavior change) and `Mapped { mmap, len }` (reloaded segments memory-map `raw_f16.bin` instead and hand out a zero-copy `&[u16]` view backed by the kernel page cache — RSS only grows for pages the rerank path actually touches). `ImmutableSegment::raw_f16()` keeps its exact public signature (`Option<&[u16]>`), so every existing call site (rerank_exact, the AE-1 adaptive-ef estimator, GraphUnion merge, FT.INFO) is unchanged. The mmap soundness contract mirrors the existing warm-segment `sealed_mmap` module and the CSR mmap loader (`graph::csr::mmap`) already in this codebase: `raw_f16.bin` is written once via `write_immutable_segment_staged`'s staged-directory -> final-directory atomic rename, and a concurrent GC removal of a superseded segment directory is safe even while a reader still holds an open mmap (POSIX unlink semantics keep the inode's pages alive until the last mmap drops). Two new `unsafe` blocks (both isolated in `raw_f16_store.rs`, each with a SAFETY comment): `Mmap::map` (read-only file mapping) and a `slice::from_raw_parts` reinterpret of the mapped bytes as `&[u16]` (sound because the mmap base is always page-aligned, and the file is written as little-endian halves on moon's only little-endian target architectures). Flagged for user approval in tmp/wave5/SUMMARY.md. Also lands the item-A hygiene follow-up flagged during the mmap investigation: `src/text/posting.rs`'s `PostingList::term_freqs`/ `positions` grow to the peak document count ever seen for a term and are never released (the `postings` HashMap entry is intentionally kept forever, even for a term with zero live docs, per the existing `remove_doc` contract). `remove_doc` now `shrink_to_fit()`s a posting's buffers once its last live document is removed, reclaiming peak capacity for terms that go idle without changing the survive-forever entry contract or any observable tf/doc_freq/search output. Tests (red/green TDD): - `test_reload_raw_f16_sidecar_uses_mmap_and_matches_owned_rerank` (segment_io.rs): red without the mmap wiring (`raw_f16_is_mapped()` was always false); green after — also pins byte-identical sidecar content and identical `search()` output between the Owned and Mapped backing. - `map_file_*` unit tests (raw_f16_store.rs): roundtrip, size-mismatch rejection, missing-file error, empty-file handling. - `remove_doc_shrinks_now_empty_posting_capacity`, `remove_doc_shrinks_now_empty_posting_positions_capacity`, `remove_doc_does_not_shrink_still_live_posting` (posting.rs): red without the shrink_to_fit() calls (verified by temporarily reverting them). Verified: full `vector::` lib test module (626 passed), text::posting tests, and `crash_recovery_vector_durability -- --ignored s1` all green with MOON_BIN pinned to a fresh `cargo build` debug binary. fmt-clean, clippy-clean on both default and `runtime-tokio,jemalloc` feature sets. author: Tin Dang --- CHANGELOG.md | 22 ++++ src/text/posting.rs | 101 ++++++++++++++ src/vector/persistence/segment_io.rs | 112 +++++++++++++--- src/vector/segment/immutable.rs | 53 ++++++-- src/vector/segment/mod.rs | 2 + src/vector/segment/raw_f16_store.rs | 190 +++++++++++++++++++++++++++ 6 files changed, 451 insertions(+), 29 deletions(-) create mode 100644 src/vector/segment/raw_f16_store.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index c144e3031..ce5af7537 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,28 @@ 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. + ### 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/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..9ccf6416b 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,72 @@ 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 + ); + } + } + #[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..ed9637f5e 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() { 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..94e9a6d71 --- /dev/null +++ b/src/vector/segment/raw_f16_store.rs @@ -0,0 +1,190 @@ +//! 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()))); + } + // SAFETY: `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). + 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 { .. }) + } + + /// Zero-copy view of the sidecar as `&[u16]`. + pub fn as_slice(&self) -> &[u16] { + match self { + Self::Owned(v) => v, + Self::Mapped { mmap, len } => { + // SAFETY: `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. `mmap.as_ptr()` is the base + // of a kernel-provided mapping, always page-aligned + // (>= 4096 bytes), which trivially satisfies `u16`'s 2-byte + // alignment — no unaligned-read UB is possible. This mirrors + // the identical `from_raw_parts` reinterpret pattern already + // used for mmap'd CSR arrays in `crate::graph::csr::mmap`. + 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]); + } +} From 57670ac3eccbd0c8fb2c12e5a4cd96624c4bf5c5 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 23:45:25 +0700 Subject: [PATCH 02/11] perf(persistence): idle-adaptive AOF writer wake cadence (RSS/CPU wave 5, item B) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 3 steady-state AOF writer loops that need a bounded channel poll to service the EverySec proactive-fsync deadline check (TopLevel tokio, PerShard tokio, PerShard monoio) polled at a FIXED cadence forever — 50ms for the monoio per-shard writer, 200ms for both tokio writers — even when the server sits completely idle. That is 5-20 wakeups per second per shard doing nothing but re-checking "is there a message yet" and "has 1s passed since the last fsync". TopLevel monoio needed no change: it already blocks on an untimed `rx.recv()`, correct because idle there genuinely means nothing buffered to protect a deadline for. Introduce `IdleWait` (`src/persistence/aof/writer_task.rs`): a tiny state machine tracking an escalation step (50ms -> 250ms -> 1s) and a "pending deadline" flag. `current()` gives the wait duration for the next poll; `on_message()` resets to the fast floor; `on_timeout()` escalates UNLESS a deadline is pending, in which case it stays pinned at the floor. This is safe with zero latency cost on real traffic: `recv_timeout`/ `tokio::time::timeout(..., recv_async())` race a message against the deadline, so an incoming write always wakes the loop immediately no matter how long the current timeout is set — only the "still idle, nothing to do" re-poll cadence relaxes. The one invariant that must never regress is the EverySec bound: the oldest unflushed byte must reach disk within ~1s + one wake, exactly as under the old fixed cadence. `mark_pending()`/`clear_pending()` enforce this literally: escalation is refused for as long as (a) a batch was written under `FsyncPolicy::EverySec` without an immediate fsync (relying on the proactive deadline check), or (b) `last_fsync` was manually back-dated by the F6 post-fold drain trick (both TopLevel-tokio Rewrite/ RewriteSharded paths and the PerShard-monoio RewritePerShard path) — the back-dating comments explicitly promise a "~150ms total" post-fold window assuming a floor-speed next wake, so escalating away from the floor before that fires would have silently widened the window. `FsyncPolicy::Always` never buffers past its own same-iteration fsync and `FsyncPolicy::No` has no deadline at all, so neither ever calls `mark_pending` — both escalate freely once idle, which is correct: nothing time-sensitive to protect. Tests (red/green TDD): `idle_wait_tests` (5 unit tests) pin the state machine directly — start-at-floor, escalate-and-cap, message-resets, pending-blocks-escalation, clearing-resumes-escalation. Verified (fresh `cargo build --release`, since 3 of the integration suites below hardcode `./target/release/moon` rather than honoring `MOON_BIN`/`find_moon_binary` — a pre-existing test-harness quirk, not touched here): - `crash_matrix_per_shard_aof -- --ignored` (3/3) - `crash_matrix_per_shard_bgrewriteaof -- --ignored` (2/2) - `wal_group_commit --include-ignored` (13/13, incl. both sigkill integration tests) - `coordinator_local_leg_durability --include-ignored` (7/7) - `crash_recovery_vector_durability -- --ignored s1` (MOON_BIN pinned) - `cargo test --lib -- persistence::aof::` under both default and `runtime-tokio,jemalloc` feature sets (50 / 56 passed) Also discovered (not caused by this change — confirmed by a 3x baseline run on the item-A-only commit before this one, which reproduced the same ~2/3 failure rate): `aof_fsync_err_subscribe_ordering`'s `_multi_shard` case is flaky independent of these edits. Left untouched; noted as a follow-up in tmp/wave5/SUMMARY.md. fmt-clean, clippy-clean (default and runtime-tokio,jemalloc feature sets). author: Tin Dang --- CHANGELOG.md | 16 ++ src/persistence/aof/writer_task.rs | 259 ++++++++++++++++++++++++++--- 2 files changed, 250 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ce5af7537..c7109469d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. ### CI — fix Windows main-push test failures (PR #TBD) diff --git a/src/persistence/aof/writer_task.rs b/src/persistence/aof/writer_task.rs index 188b8db4a..eef160fe8 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,18 @@ 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()` (starts at 200ms, escalates to 1s while + // truly idle — see `IdleWait` docs) 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 +527,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 +538,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 +607,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 +658,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 +706,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 +734,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 +749,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 +889,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()` (starts at 200ms, escalates 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 +925,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()` (starts at 200ms, escalates 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 +948,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 +1065,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 +1166,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 +1267,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 +1283,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 +1397,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 +1477,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 +1534,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(); } } } From 951cdbb7da6e4866f5d462b3164cf9ef49c0d891 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 23:52:59 +0700 Subject: [PATCH 03/11] perf(persistence): shrink WAL v3 write buffer after an oversized flush (RSS/CPU wave 5, item C1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `WalWriterV3::buf` (`src/persistence/wal_v3/segment.rs`) starts at 8KB but is a plain `Vec` — a single oversized record (e.g. a large FullPageImage payload written through `append`) grows it via the normal `Vec` growth policy, and `flush_write`/`rotate_segment` only ever `.clear()` the buffer afterward. `clear()` does not release capacity, so one big write pins that peak allocation for the writer's entire lifetime — `resident_bytes()` (the P10 INFO memory accounting hook) would report the high-water mark forever, not the steady-state working set. Add a `WAL_BUF_SHRINK_THRESHOLD` (4x the 8KB default): both places that drain the buffer (`flush_write`'s normal path and `rotate_segment`'s end-of-segment flush) now `shrink_to(DEFAULT_WAL_BUF_CAPACITY)` once capacity exceeds the threshold. `shrink_to` is a no-op when already at or below the target, so steady small-record traffic (the common case) never pays a reallocation. Tests (red/green TDD): `test_buffer_shrinks_after_flush_following_large_record` appends a 100KB record, confirms `resident_bytes()` exceeds the threshold, flushes, and asserts the buffer drops back to the default capacity — failed to compile without the new constants/shrink call (red), passes after (green). `test_buffer_does_not_shrink_below_threshold` pins the common case: small records never trip the shrink path. Verified: `cargo test --lib persistence::wal_v3::segment` (15/15 passed). fmt-clean, clippy-clean (default features; no runtime-specific code touched so no separate tokio-feature run needed for this file). author: Tin Dang --- CHANGELOG.md | 7 ++++ src/persistence/wal_v3/segment.rs | 63 ++++++++++++++++++++++++++++++- 2 files changed, 69 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c7109469d..d690ca0e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. ### CI — fix Windows main-push test failures (PR #TBD) 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(); From c462f1fcf6456868c82c8b17353770cb01fdc76f Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 23:57:59 +0700 Subject: [PATCH 04/11] perf(shard): SmallVec the per-tick elastic-budget shard snapshot (RSS/CPU wave 5, item C3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ShardDatabases::recompute_elastic_budget` (`src/shard/shared_databases.rs`) is called from every shard's 100ms eviction tick and used to `collect()` a fresh `Vec` snapshot of all shards' published memory on every call — one heap allocation per shard per tick (num_shards allocations every 100ms cluster-wide), just to hand a `&[usize]` to `compute_elastic_budget`. Switch the snapshot to `SmallVec<[usize; 16]>`: deployments with <=16 shards (the overwhelming common case) now do this entirely on the stack; larger shard counts still spill to one heap allocation per call, same as before — no regression, pure win for the common case. `compute_elastic_budget`'s signature (`&[usize]`) is unchanged; `SmallVec` derefs to a slice so the call site needs no other changes. Tests (red/green TDD, adapted for a container-swap change with no logic delta): added `recompute_elastic_budget_correct_beyond_smallvec_inline_capacity`, a 20-shard scenario that exercises the SmallVec heap-spill boundary (>16 inline capacity) to pin that the swap never truncates or reorders shard readings once it spills. Existing `recompute_elastic_budget_hot_shard_borrows_idle_headroom` and `recompute_elastic_budget_disabled_for_single_shard_or_unlimited` continue to cover the inline (<=16) path unchanged. Verified: `cargo test --lib shard::shared_databases::tests` (7/7 passed). fmt-clean, clippy-clean on both default and `runtime-tokio,jemalloc` feature sets (smallvec is a pre-existing workspace dependency, used elsewhere e.g. HNSW search). author: Tin Dang --- CHANGELOG.md | 14 ++++++++++++++ src/shard/shared_databases.rs | 25 ++++++++++++++++++++++++- 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d690ca0e4..5d089c13a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -50,6 +50,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. ### CI — fix Windows main-push test failures (PR #TBD) diff --git a/src/shard/shared_databases.rs b/src/shard/shared_databases.rs index b85d8655b..44b169936 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}; @@ -225,7 +226,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 +752,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); From c4df97b2e3522cb1ee32649be878386e22d42a05 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 7 Jul 2026 00:10:13 +0700 Subject: [PATCH 05/11] feat(scripting,admin): expose Lua script-cache byte estimate via INFO/MEMORY DOCTOR (RSS/CPU wave 5, item C4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ScriptCache` (`src/scripting/cache.rs`) is an unbounded per-shard `HashMap` — by design, matching Redis (`SCRIPT FLUSH` is the only eviction path; no silent eviction was ever appropriate here). But it was also completely invisible to observability: MEMORY DOCTOR and the Prometheus `moon_memory_bytes` gauge accounted for DashTable/HNSW/ CSR/replication-backlog but folded any Lua cache growth silently into "allocator overhead," making a large `EVAL`/`SCRIPT LOAD` workload undiagnosable from the outside. Add `ScriptCache::resident_bytes()` — sum of hex-SHA1 key length + script body length per entry (an estimate; excludes `HashMap`/`String`/`Bytes` allocator bookkeeping, consistent with how the sibling vector/graph estimators in this codebase already work). Wire it through the existing C5/M4 per-shard `ShardStoreMemory` publish pattern: a new `lua: AtomicUsize` field, refreshed on the same 100ms eviction tick that already refreshes vector/text/graph (`run_eviction_tick` now takes the shard's `Rc>` to read it). `admin/metrics_setup.rs`'s Prometheus emitter and `command/server_admin.rs`'s `MEMORY DOCTOR` text output both gained a `lua`/`Lua scripts:` line, folded into their tracked sum (so "allocator overhead" shrinks by exactly the amount now attributed correctly). Tests (red/green TDD): `test_resident_bytes_empty_cache_is_zero` and `test_resident_bytes_grows_with_entries_and_shrinks_on_flush` pin the new method directly (0 for an empty cache; exact byte count for 1-2 entries; back to 0 after `flush()`) — both failed to compile before `resident_bytes()` existed (red), pass after (green). Verified: `cargo test --lib -- scripting::cache::tests shard::shared_databases::tests shard::slice::tests shard::mq_exec::tests` (29/29) under both default and `runtime-tokio,jemalloc` feature sets. fmt-clean, clippy-clean (`-D warnings`) on both feature sets; `cargo build`/`cargo check` compile clean on both. author: Tin Dang --- CHANGELOG.md | 9 +++++++++ src/admin/metrics_setup.rs | 6 +++++- src/command/server_admin.rs | 21 ++++++++++++++++++-- src/scripting/cache.rs | 36 +++++++++++++++++++++++++++++++++++ src/shard/event_loop.rs | 2 ++ src/shard/mq_exec.rs | 1 + src/shard/persistence_tick.rs | 7 +++++++ src/shard/shared_databases.rs | 5 +++++ src/shard/slice.rs | 1 + 9 files changed, 85 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5d089c13a..7b7e0a578 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -64,6 +64,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `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. ### CI — fix Windows main-push test failures (PR #TBD) 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/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 44b169936..8797d81cd 100644 --- a/src/shard/shared_databases.rs +++ b/src/shard/shared_databases.rs @@ -24,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. @@ -103,6 +107,7 @@ impl ShardDatabases { vector: AtomicUsize::new(0), text: AtomicUsize::new(0), graph: AtomicUsize::new(0), + lua: AtomicUsize::new(0), }) }) .collect::>() 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), }), } } From 6a021ab0f82e8ab164b2bf5adace709991ebc000 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 7 Jul 2026 00:15:48 +0700 Subject: [PATCH 06/11] chore(protocol): remove dead parse_single_frame_zc parser (RSS/CPU wave 5, item C5) `parse_single_frame_zc` (`src/protocol/parse.rs`) was a full RESP2/RESP3 frame parser, superseded by the current two-pass pipeline (`validate_frame` then `parse_frame_zerocopy`, see `parse()`) but never deleted. It had zero external callers -- the only references to it were its own recursive calls for Array/Map/Set/Push nesting. Its private helper `read_decimal_zc` was in the same boat: used exclusively inside the dead function. The file-level `#![allow(dead_code)]` had been silently absorbing the warning this whole time, which is exactly why the spec flagged it for a verify-and-remove pass rather than a design change. Verified dead (not just the originally-suspected Map arm, but the entire 150-line function): grepped the whole worktree (`src/`, `tests/`, `fuzz/fuzz_targets/`) for `parse_single_frame_zc` and `read_decimal_zc` -- every hit was inside `parse.rs` itself, all self-recursive. The RESP fuzz targets (`resp_parse.rs`, `resp_parse_differential.rs`) only call the public `parse::parse()` entry point, which has routed through `parse_frame_zerocopy` since the "defensive Frame::Null on any failure" rewrite documented at that function's doc comment -- confirming `parse_single_frame_zc` was leftover from before that rewrite. Removed both functions (~160 lines). `find_crlf`/`strict_atoi` (its two other private helpers) stay -- both are still used by the live `parse_frame_zerocopy` and `validate_frame` paths. Tests (verify-dead-code TDD per the wave-5 spec, in place of a traditional red/green cycle for a pure deletion): full `protocol::parse::tests` module re-run post-removal as the regression pin -- 50/50 passed, unchanged behavior. Verified: `cargo test --lib -- protocol::parse::tests` (50/50). fmt-clean, clippy-clean (`-D warnings`) on both default and `runtime-tokio,jemalloc` feature sets. Fuzz targets compile-check clean (grep-verified no reference to the removed functions; `cargo-fuzz` itself needs nightly and wasn't re-run, per repo convention). author: Tin Dang --- CHANGELOG.md | 6 + src/protocol/parse.rs | 266 ------------------------------------------ 2 files changed, 6 insertions(+), 266 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7b7e0a578..1705c9bfe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -73,6 +73,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `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`. ### CI — fix Windows main-push test failures (PR #TBD) 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). From e12b1f0dfa753ab0bf583dd80b0698f398d7416b Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 7 Jul 2026 00:17:18 +0700 Subject: [PATCH 07/11] docs: document jemalloc decay/arena tuning policy (RSS/CPU wave 5, item C6) Audited `src/main.rs` for jemalloc decay/arena drift: the baked-in static `_rjem_malloc_conf` export (used at process init) and the `--memory-arenas-cap` re-spawn override (`maybe_respawn_with_arena_override`) already carry the byte-identical tuning string -- `background_thread:true,metadata_thp:auto,dirty_decay_ms:1000,muzzy_decay_ms:5000,abort_conf:true` -- with only `narenas` substituted for the operator's requested cap. No drift, no duplicated-with-different-values config to reconcile; grepped `config.rs`, `runtime/mod.rs`, `admin/metrics_setup.rs`, and `command/server_admin.rs` for any second/conflicting decay definition -- none exists. What WAS missing: none of this was documented for operators. CLAUDE.md's Environment Variables section lists `_RJEM_MALLOC_CONF`-adjacent knobs (`--memory-arenas-cap` is only documented in `config.rs`'s CLI help string) but never explains the actual decay policy or the env var an operator would need to override it. Added a `_RJEM_MALLOC_CONF` entry: what the baked-in defaults are, why 1s dirty-page decay + a background reclaim thread matter for RSS (freed-but-idle pages return to the OS promptly instead of sitting in jemalloc's caches), the exec-before-init timing constraint `--memory-arenas-cap` depends on, and the existing "operator env wins" guard. Docs-only change per the wave-5 spec's guidance for this item ("if no tuning-code drift exists, add MALLOC_CONF guidance to docs only -- config hygiene only, measure nothing"). No code touched; no tests to run. author: Tin Dang --- CHANGELOG.md | 6 ++++++ CLAUDE.md | 1 + 2 files changed, 7 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1705c9bfe..f3d3964b8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -79,6 +79,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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). ### CI — fix Windows main-push test failures (PR #TBD) 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 From 0d21efcd5fe04f1730c7823aba707f8a0f3909a3 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 7 Jul 2026 00:19:36 +0700 Subject: [PATCH 08/11] docs: audit tokio 1ms shard tick idle cost, no change needed (RSS/CPU wave 5, item C7) Audited the tokio-runtime shard event loop's 1ms `periodic_interval` tick (`src/shard/event_loop.rs` ~1292, handler in `spsc_handler::drain_spsc_shared`) for idle-cost waste, the same class of problem item B fixed for the AOF writer's poll cadence. Found the tick is already cheap when idle: - `drain_spsc_shared`'s per-consumer drain loop is a non-blocking `try_pop()` that returns `None` immediately when a consumer's queue is empty -- O(num_consumers) pointer checks, zero allocation (scratch `Vec`s are thread-local and reused via `mem::take`). - Every side effect downstream of the drain is already gated behind a cheap conditional: WAL re-notify only `if hit_cap`, autovacuum schedule persist only `if is_dirty()`, CDC registration only `if !pending_cdc_subscribes.is_empty()`, migrations only when `pending_migrations` is non-empty. - The one unconditional per-tick cost, `cached_clock.update()`, is a single `clock_gettime` call by design -- its own doc comment states it's "the ONE place per shard that actually calls clock_gettime," exactly the "Timestamp caching" design decision CLAUDE.md documents. Unlike item B's AOF writer (which polled a channel purely to catch an EverySec deadline with no inherent latency contract on the poll interval itself), this 1ms cadence IS the latency contract: CLAUDE.md's Key Design Decisions calls it out explicitly ("Low-latency append via in-memory buffer flushed on 1ms tick"). Escalating it while idle, the way item B escalates the AOF writer's poll, would directly widen the WAL-flush latency bound it exists to hold -- there is no "idle" for a write-latency SLA in the way there is for an EverySec fsync deadline. Disposition: SKIP code change. No unsafe cheap-skip exists beyond what's already there; a real fix (event-driven WAL flush trigger + a much longer backstop timer, replacing the periodic tick's role entirely) is an architectural change, not a hygiene-item-sized fix -- documented here as a follow-up for a future wave if idle CPU from this specific tick is ever measured as material (not attempted in this audit; no regression possible since nothing was changed). author: Tin Dang --- CHANGELOG.md | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index f3d3964b8..38d99b126 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -85,6 +85,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `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. ### CI — fix Windows main-push test failures (PR #TBD) From 133213d36abae5e080f46fb8fa28463012dfb8c6 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 7 Jul 2026 00:26:45 +0700 Subject: [PATCH 09/11] test(sigterm): widen readiness poll deadline to tolerate host load (RSS/CPU wave 5, item C8) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `tests/sigterm_shutdown.rs`'s `wait_for_ready` was already poll-based (100ms re-poll against a deadline, matching the pattern used elsewhere in this repo, e.g. `crash_recovery_vector_durability`'s `wait_ready`) — but both call sites hardcoded a fixed 15s deadline independently. PR #218 documented "server did not become ready within 15s" as an observed flake, not a real readiness regression: a contended CI runner or the documented OrbStack virtiofs starvation pattern (CLAUDE.md gotchas) can push a cold server spawn + first-bind past 15s even though the process is healthy. Introduce a single `READY_TIMEOUT` constant (60s) shared by both `wait_for_ready` call sites (the main `assert_sigterm_clean_exit_shards` path and the write-storm test), replacing the two independently-hardcoded 15s literals. This does not slow down the healthy-server case at all -- the poll loop still notices readiness within one 100ms tick of it actually happening; only a genuinely-broken server (or a truly wedged CI host) would ever wait out the full deadline. Panic messages now interpolate the constant instead of hardcoding "15s" so they can't drift out of sync with the actual value again. Left the separate post-SIGTERM exit deadline (10s, `assert_sigterm_clean_exit_shards`'s shutdown-wait loop) untouched -- it's a different flake surface not implicated by the #218 report, and out of scope for this item. Tests (verify-and-harden, not a traditional red/green cycle -- this widens a timeout rather than fixing a logic bug): all 7 sigterm tests re-run to confirm the constant swap didn't change pass/fail behavior in the healthy-host case. Verified: `cargo test --release --test sigterm_shutdown -- --test-threads=1` (7/7 passed). fmt-clean. `cargo clippy -- -D warnings` (the exact default-feature CI gate, no `--tests`) is clean; `--tests` surfaces pre-existing, unrelated warnings across other integration-test files not touched here (config.rs default-then-assign, doc-list indentation in xshard_fastpath_api.rs, etc.) -- confirmed pre-existing, not introduced by this change. author: Tin Dang --- CHANGELOG.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 38d99b126..dfa70ad1f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -96,6 +96,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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 test readiness deadline widened to tolerate host + load** (`tests/sigterm_shutdown.rs`): both `wait_for_ready` call sites + hardcoded an independent fixed 15s deadline; PR #218 documented "server + did not become ready within 15s" as a host-load flake, not a real + regression (the poll loop is already load-tolerant — 100ms re-poll, no + fixed pre-sleep). Introduced a shared `READY_TIMEOUT` constant (60s), + replacing both literals; panic messages interpolate it so they can't + drift out of sync again. Healthy-server pass time is unaffected. ### CI — fix Windows main-push test failures (PR #TBD) From 799db80fad403c99c36de7625594798e59570de6 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 7 Jul 2026 01:10:01 +0700 Subject: [PATCH 10/11] fix(vector): mapped raw_f16 sidecar must not count as resident memory MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deep-review follow-up to item A (mmap exact-rerank sidecar): ImmutableSegment::resident_bytes() still charged a Mapped sidecar at its full len*2 bytes. Mapped pages are kernel page cache — reclaimable under memory pressure, not pinned heap — so counting them as resident made the elastic memory budget / eviction pipeline behave as if the mmap RSS win never happened for reloaded segments (over-reporting is the safe direction, but it silently negates the feature's benefit in Moon's own accounting). New RawF16Store::resident_bytes() (Owned = full buffer, Mapped = 0), used by resident_bytes(). Pinned by a unit test (Owned=200/Mapped=0 for 100 halves) and an integration assert in the reload-parity test (owned segment reports at least sidecar-bytes more than the reloaded mapped one). Also corrects three stale comments in writer_task.rs claiming the idle ladder "starts at 200ms" — AOF_IDLE_WAIT_STEPS' floor is 50ms (tighter than the old fixed 200ms tokio cadence right after activity, escalating to 1s once idle). author: Tin Dang --- CHANGELOG.md | 10 ++------ src/persistence/aof/writer_task.rs | 10 ++++---- src/vector/persistence/segment_io.rs | 18 +++++++++++++++ src/vector/segment/immutable.rs | 7 +++--- src/vector/segment/raw_f16_store.rs | 34 ++++++++++++++++++++++++++++ 5 files changed, 63 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dfa70ad1f..e6c6e2327 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -96,14 +96,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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 test readiness deadline widened to tolerate host - load** (`tests/sigterm_shutdown.rs`): both `wait_for_ready` call sites - hardcoded an independent fixed 15s deadline; PR #218 documented "server - did not become ready within 15s" as a host-load flake, not a real - regression (the poll loop is already load-tolerant — 100ms re-poll, no - fixed pre-sleep). Introduced a shared `READY_TIMEOUT` constant (60s), - replacing both literals; panic messages interpolate it so they can't - drift out of sync again. Healthy-server pass time is unaffected. +- 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) diff --git a/src/persistence/aof/writer_task.rs b/src/persistence/aof/writer_task.rs index eef160fe8..8d9176d0f 100644 --- a/src/persistence/aof/writer_task.rs +++ b/src/persistence/aof/writer_task.rs @@ -499,8 +499,10 @@ pub async fn aof_writer_task( #[cfg(feature = "runtime-tokio")] { // Bounded recv (EverySec durability): wake at least every - // `idle_wait.current()` (starts at 200ms, escalates to 1s while - // truly idle — see `IdleWait` docs) even when idle so the flush + // `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 @@ -894,7 +896,7 @@ pub async fn per_shard_aof_writer_task( 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 - // `idle_wait.current()` (starts at 200ms, escalates while idle) + // `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.) @@ -926,7 +928,7 @@ pub async fn per_shard_aof_writer_task( loop { tokio::select! { // Bounded recv (EverySec durability): wake at least every - // `idle_wait.current()` (starts at 200ms, escalates while + // `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 diff --git a/src/vector/persistence/segment_io.rs b/src/vector/persistence/segment_io.rs index 9ccf6416b..d81867840 100644 --- a/src/vector/persistence/segment_io.rs +++ b/src/vector/persistence/segment_io.rs @@ -922,6 +922,24 @@ mod tests { 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] diff --git a/src/vector/segment/immutable.rs b/src/vector/segment/immutable.rs index ed9637f5e..97812c681 100644 --- a/src/vector/segment/immutable.rs +++ b/src/vector/segment/immutable.rs @@ -890,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/raw_f16_store.rs b/src/vector/segment/raw_f16_store.rs index 94e9a6d71..2114a0d72 100644 --- a/src/vector/segment/raw_f16_store.rs +++ b/src/vector/segment/raw_f16_store.rs @@ -101,6 +101,18 @@ impl RawF16Store { 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 { @@ -187,4 +199,26 @@ mod tests { 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" + ); + } } From 54207208c73e7f264649f228bafcb11bac809af1 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 7 Jul 2026 09:51:33 +0700 Subject: [PATCH 11/11] docs(vector): move SAFETY markers within audit range of unsafe blocks CI's Lint job runs scripts/audit-unsafe.sh, which requires the literal "SAFETY:" marker within the 3 lines immediately above each unsafe block. Both raw_f16_store.rs comments had the marker at the TOP of a longer explanatory block (5 and 13 lines above the unsafe), so the audit flagged them as missing despite full SAFETY documentation. Restructured both: the detailed invariant explanation stays, and a one-line "// SAFETY: ..." summary now sits directly above each unsafe expression. audit-unsafe.sh passes locally (257/257); no code change. author: Tin Dang --- src/vector/segment/raw_f16_store.rs | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/src/vector/segment/raw_f16_store.rs b/src/vector/segment/raw_f16_store.rs index 2114a0d72..9a248af41 100644 --- a/src/vector/segment/raw_f16_store.rs +++ b/src/vector/segment/raw_f16_store.rs @@ -70,11 +70,12 @@ impl RawF16Store { // (e.g. a segment with zero live vectors) needs no mapping. return Ok(Some(Self::Owned(Vec::new()))); } - // SAFETY: `path` is `raw_f16.bin` inside a `segment-{id}` directory. - // It is written exactly once by `write_segment_files` and only made + // `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, @@ -118,19 +119,18 @@ impl RawF16Store { match self { Self::Owned(v) => v, Self::Mapped { mmap, len } => { - // SAFETY: `mmap` maps exactly `len * 2` bytes of a file - // written as `len` little-endian `u16` halves (see + // `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. `mmap.as_ptr()` is the base - // of a kernel-provided mapping, always page-aligned - // (>= 4096 bytes), which trivially satisfies `u16`'s 2-byte - // alignment — no unaligned-read UB is possible. This mirrors - // the identical `from_raw_parts` reinterpret pattern already - // used for mmap'd CSR arrays in `crate::graph::csr::mmap`. + // 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) } } }