From aeea937a10cc84a921d2ffdbeceffae138e721b2 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 19:21:06 +0700 Subject: [PATCH 1/8] =?UTF-8?q?feat(vector):=20add=20BucketedKeyMap=20?= =?UTF-8?q?=E2=80=94=20bucketed=20CoW=20keymap=20(256=20buckets)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit VectorIndex's key_hash -> V maps were single Arc> instances. Under ft-search-off-eventloop, a live FT.SEARCH snapshot holds an extra Arc clone of the whole map while the event loop keeps processing writes; the next write's Arc::make_mut then sees refcount > 1 and full-clones the entire map synchronously on the shard event loop (~47MB @1M vectors). BucketedKeyMap shards the map into 256 fixed buckets, each an independently-cloneable Arc>, with bucket selected by (key_hash >> 56). A concurrent snapshot now pins at most 1/256th of the map per write instead of the whole thing. Manual Clone impl avoids a spurious V: Clone bound that #[derive(Clone)] would add. 8 unit tests: get/insert/remove/len round-trip against a reference HashMap under thousands of random ops, snapshot isolation, and the key property test — after snap = map.snapshot(), one insert leaves >= 255 buckets Arc::ptr_eq with the snapshot. Not yet wired into VectorIndex; that follows in the next commit. author: Tin Dang --- src/vector/keymap.rs | 426 +++++++++++++++++++++++++++++++++++++++++++ src/vector/mod.rs | 1 + 2 files changed, 427 insertions(+) create mode 100644 src/vector/keymap.rs diff --git a/src/vector/keymap.rs b/src/vector/keymap.rs new file mode 100644 index 00000000..b7912523 --- /dev/null +++ b/src/vector/keymap.rs @@ -0,0 +1,426 @@ +//! Bucketed copy-on-write key-hash map (RSS/CPU wave 4, defect 1 — +//! `Arc` CoW amplification). +//! +//! ## Problem +//! +//! `VectorIndex` keeps three `key_hash -> V` maps (`key_hash_to_key`, +//! `key_hash_to_global_id`, `key_hash_to_vec_checksum`) that used to be a +//! single flat `Arc>`. Every FT.SEARCH snapshot +//! (`SearchSnapshot::key_hash_to_key`) clones the `Arc` cheaply (a refcount +//! bump), but `ft-search-off-eventloop` lets the shard event loop keep +//! processing writes while that search `.await`s. The next HSET/HDEL calls +//! `Arc::make_mut` on the map — which sees refcount > 1 and deep-clones the +//! ENTIRE map (tens of MB at scale) SYNCHRONOUSLY on the shard thread, once +//! per write, for the whole lifetime of the live search. +//! +//! ## Fix +//! +//! Shard the map into [`NUM_BUCKETS`] independent `Arc>` +//! buckets, keyed by the top bits of the (already well-distributed) xxh64 +//! key hash. A search snapshot still clones the whole structure in O(1) — +//! [`NUM_BUCKETS`] refcount bumps, no data copy — but a concurrent +//! insert/remove only calls `Arc::make_mut` on the ONE bucket the key hashes +//! into. Worst case, a write during a live snapshot clones ~1/256th of the +//! map instead of all of it. +//! +//! Persistence (`keymap.bin`) is unaffected: [`BucketedKeyMap::iter`] walks +//! every bucket and the on-disk format is a flat, hash-keyed entry list that +//! was never order-sensitive (recovery re-inserts by key_hash into fresh +//! buckets — see `crate::vector::persistence::recover_v2`). + +use std::collections::HashMap; +use std::sync::Arc; + +/// Number of independent CoW buckets. Bucket index is the top 8 bits of the +/// `u64` key hash (`hash >> 56`), which is uniformly distributed for xxh64 +/// output — no separate mixing step needed. +pub const NUM_BUCKETS: usize = 256; + +/// A `key_hash -> V` map, internally sharded into [`NUM_BUCKETS`] +/// independently copy-on-write buckets. +/// +/// Cloning a `BucketedKeyMap` is O(`NUM_BUCKETS`) — a fixed number of `Arc` +/// refcount bumps, never a data copy. This is what makes search-snapshot +/// capture cheap (see `SearchSnapshot::key_hash_to_key`) while keeping +/// concurrent-write CoW cost bucket-scoped instead of map-wide. +pub struct BucketedKeyMap { + buckets: Box<[Arc>]>, +} + +impl BucketedKeyMap { + /// Create an empty map with all `NUM_BUCKETS` buckets allocated (each an + /// `Arc` around an empty `HashMap` — no per-bucket heap allocation until + /// the first insert into that bucket, since an empty `HashMap` doesn't + /// allocate). + pub fn new() -> Self { + let buckets: Vec>> = + (0..NUM_BUCKETS).map(|_| Arc::new(HashMap::new())).collect(); + Self { + buckets: buckets.into_boxed_slice(), + } + } + + /// Bucket index for a given key hash: top 8 bits. + #[inline] + fn bucket_of(key_hash: u64) -> usize { + (key_hash >> 56) as usize + } + + /// Look up a value by key hash. + #[inline] + pub fn get(&self, key_hash: &u64) -> Option<&V> { + self.buckets[Self::bucket_of(*key_hash)].get(key_hash) + } + + /// Whether `key_hash` is present. + #[inline] + pub fn contains_key(&self, key_hash: &u64) -> bool { + self.buckets[Self::bucket_of(*key_hash)].contains_key(key_hash) + } + + /// Insert a value, returning the previous value (if any). Only the + /// bucket `key_hash` maps into is touched — `Arc::make_mut` there clones + /// that ONE bucket's `HashMap` if (and only if) a snapshot is + /// concurrently holding a reference to it. + pub fn insert(&mut self, key_hash: u64, value: V) -> Option + where + V: Clone, + { + let idx = Self::bucket_of(key_hash); + Arc::make_mut(&mut self.buckets[idx]).insert(key_hash, value) + } + + /// Remove a value by key hash, returning it if present. Bucket-scoped, + /// same CoW cost profile as [`Self::insert`]. + pub fn remove(&mut self, key_hash: &u64) -> Option + where + V: Clone, + { + let idx = Self::bucket_of(*key_hash); + Arc::make_mut(&mut self.buckets[idx]).remove(key_hash) + } + + /// Entry-style get-or-insert: returns the existing value for `key_hash`, + /// or computes `default()` and inserts it. `default` is only invoked when + /// the key is absent (mirrors `HashMap::entry(..).or_insert_with`). + /// Bucket-scoped CoW, same as [`Self::insert`]. + pub fn get_or_insert_with(&mut self, key_hash: u64, default: impl FnOnce() -> V) -> &V + where + V: Clone, + { + let idx = Self::bucket_of(key_hash); + Arc::make_mut(&mut self.buckets[idx]) + .entry(key_hash) + .or_insert_with(default) + } + + /// Total number of entries across all buckets. O(`NUM_BUCKETS`). + pub fn len(&self) -> usize { + self.buckets.iter().map(|b| b.len()).sum() + } + + /// Whether the map has no entries. O(`NUM_BUCKETS`). + pub fn is_empty(&self) -> bool { + self.buckets.iter().all(|b| b.is_empty()) + } + + /// Iterate all `(key_hash, value)` pairs across every bucket. Used by the + /// persistence snapshot writer (order is not significant — the on-disk + /// keymap format is keyed by `key_hash`, not by insertion/iteration + /// order) and by admin paths that need every key (e.g. `FT.DROPINDEX + /// DD`). + pub fn iter(&self) -> impl Iterator { + self.buckets.iter().flat_map(|b| b.iter()) + } + + /// Iterate all key hashes across every bucket. + pub fn keys(&self) -> impl Iterator { + self.iter().map(|(k, _)| k) + } + + /// Iterate all values across every bucket. + pub fn values(&self) -> impl Iterator { + self.iter().map(|(_, v)| v) + } + + /// Cheap O(`NUM_BUCKETS`) snapshot: clones the bucket-`Arc` array (refcount + /// bumps only, zero data copy). Semantically identical to [`Clone::clone`] + /// — this named alias documents intent at capture sites (e.g. + /// `SearchSnapshot` construction) that want to read "take an isolated + /// view", not "duplicate the whole map". + pub fn snapshot(&self) -> Self { + self.clone() + } +} + +impl Default for BucketedKeyMap { + fn default() -> Self { + Self::new() + } +} + +// Manual `Clone` impl (rather than `#[derive(Clone)]`) so cloning a +// `BucketedKeyMap` never requires `V: Clone` — only `Arc>: Clone` is needed, which holds unconditionally (Arc's Clone impl has +// no bound on its inner type). `#[derive(Clone)]` would add a spurious `V: +// Clone` bound to the generated impl. +impl Clone for BucketedKeyMap { + fn clone(&self) -> Self { + Self { + buckets: self.buckets.clone(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap as StdHashMap; + + /// Deterministic LCG (matches the style already used for pseudo-random + /// test vectors elsewhere in the vector module, e.g. + /// `persistence::recover_v2::f32_blob`) — no external `rand` dependency + /// needed for a property-style test, and results are reproducible. + struct Lcg(u64); + impl Lcg { + fn next_u64(&mut self) -> u64 { + // Fixed-point LCG constants (Numerical Recipes). + self.0 = self.0.wrapping_mul(6364136223846793005).wrapping_add(1); + self.0 + } + } + + #[test] + fn new_map_is_empty() { + let m: BucketedKeyMap = BucketedKeyMap::new(); + assert_eq!(m.len(), 0); + assert!(m.is_empty()); + assert_eq!(m.get(&42), None); + } + + #[test] + fn insert_get_remove_single_key() { + let mut m: BucketedKeyMap = BucketedKeyMap::new(); + assert_eq!(m.insert(100, 7), None); + assert_eq!(m.get(&100), Some(&7)); + assert!(m.contains_key(&100)); + assert_eq!(m.len(), 1); + + // Overwrite returns the previous value. + assert_eq!(m.insert(100, 9), Some(7)); + assert_eq!(m.get(&100), Some(&9)); + assert_eq!(m.len(), 1); + + assert_eq!(m.remove(&100), Some(9)); + assert_eq!(m.get(&100), None); + assert!(!m.contains_key(&100)); + assert_eq!(m.len(), 0); + assert!(m.is_empty()); + } + + #[test] + fn get_or_insert_with_only_computes_on_vacant() { + let mut m: BucketedKeyMap = BucketedKeyMap::new(); + let mut calls = 0; + assert_eq!( + *m.get_or_insert_with(1, || { + calls += 1; + 5 + }), + 5 + ); + assert_eq!(calls, 1); + // Second call on the same (now-occupied) key must NOT invoke the closure. + assert_eq!( + *m.get_or_insert_with(1, || { + calls += 1; + 999 + }), + 5 + ); + assert_eq!(calls, 1); + } + + /// Property test: thousands of random insert/remove ops against a + /// `BucketedKeyMap` must match a plain reference `HashMap` at every step + /// (len, get, contains_key) and in final full-iteration content. + #[test] + fn property_matches_reference_hashmap_over_random_ops() { + let mut rng = Lcg(0xC0FFEE_u64); + let mut reference: StdHashMap = StdHashMap::new(); + let mut bucketed: BucketedKeyMap = BucketedKeyMap::new(); + + // Keep the key space small enough that removes/overwrites actually + // exercise existing entries, not just fresh inserts. + const KEY_SPACE: u64 = 500; + + for step in 0..5000u64 { + let raw = rng.next_u64(); + let key = raw % KEY_SPACE; + // Spread the low-space keys across the full u64 range (and thus + // across buckets) by mixing in a hashed key_hash instead of using + // `key` directly as the map key — mirrors real key_hash usage + // (xxh64 output), not a small dense integer. + let key_hash = key_hash_for_test(key); + let op = raw % 3; + match op { + 0 => { + // insert + let value = step; + let expected_prev = reference.insert(key_hash, value); + let actual_prev = bucketed.insert(key_hash, value); + assert_eq!(actual_prev, expected_prev, "insert mismatch at step {step}"); + } + 1 => { + // remove + let expected = reference.remove(&key_hash); + let actual = bucketed.remove(&key_hash); + assert_eq!(actual, expected, "remove mismatch at step {step}"); + } + _ => { + // get / contains_key (read-only) + assert_eq!( + bucketed.get(&key_hash), + reference.get(&key_hash), + "get mismatch at step {step}" + ); + assert_eq!( + bucketed.contains_key(&key_hash), + reference.contains_key(&key_hash), + "contains_key mismatch at step {step}" + ); + } + } + assert_eq!( + bucketed.len(), + reference.len(), + "len mismatch at step {step}" + ); + } + + assert_eq!(bucketed.len(), reference.len()); + assert_eq!(bucketed.is_empty(), reference.is_empty()); + for (k, v) in &reference { + assert_eq!( + bucketed.get(k), + Some(v), + "final content mismatch for key {k}" + ); + } + let mut bucketed_sorted: Vec<(u64, u64)> = bucketed.iter().map(|(k, v)| (*k, *v)).collect(); + bucketed_sorted.sort_unstable(); + let mut reference_sorted: Vec<(u64, u64)> = + reference.iter().map(|(k, v)| (*k, *v)).collect(); + reference_sorted.sort_unstable(); + assert_eq!(bucketed_sorted, reference_sorted); + } + + /// Deterministic mixer so a small dense key space still spreads across + /// the full u64 range (hence across all 256 buckets), matching how real + /// `key_hash` values (xxh64 of a Redis key) are distributed. + fn key_hash_for_test(key: u64) -> u64 { + key.wrapping_mul(0x9E3779B97F4A7C15).rotate_left(31) + } + + #[test] + fn snapshot_isolation_mutations_after_snapshot_are_invisible() { + let mut m: BucketedKeyMap = BucketedKeyMap::new(); + m.insert(1, 10); + m.insert(2, 20); + + let snap = m.snapshot(); + assert_eq!(snap.get(&1), Some(&10)); + assert_eq!(snap.get(&2), Some(&20)); + + // Mutate the LIVE map after the snapshot was taken. + m.insert(1, 999); // update + m.insert(3, 30); // new key + m.remove(&2); // delete + + // Snapshot must be unaffected by any of the above. + assert_eq!( + snap.get(&1), + Some(&10), + "snapshot must not see post-snapshot update" + ); + assert_eq!( + snap.get(&3), + None, + "snapshot must not see post-snapshot insert" + ); + assert_eq!( + snap.get(&2), + Some(&20), + "snapshot must not see post-snapshot remove" + ); + assert_eq!(snap.len(), 2); + + // Live map reflects all three mutations. + assert_eq!(m.get(&1), Some(&999)); + assert_eq!(m.get(&3), Some(&30)); + assert_eq!(m.get(&2), None); + assert_eq!(m.len(), 2); + } + + /// THE key property (spec, "Fix: bucketed CoW keymap"): after + /// `let snap = map.snapshot()`, a single `insert` on the live map must + /// leave at least 255 of the 256 buckets `Arc::ptr_eq` between `snap` + /// and the live map — proof the CoW clone triggered by that insert was + /// bucket-scoped, not map-wide. + #[test] + fn insert_under_live_snapshot_clones_only_one_bucket() { + let mut m: BucketedKeyMap = BucketedKeyMap::new(); + // Populate every bucket with at least one entry so a real HashMap + // clone (if it happened) would be observable — an empty bucket + // cloning is a degenerate case that wouldn't prove anything. + for b in 0..NUM_BUCKETS as u64 { + let key_hash = b << 56; // top byte = bucket index, rest zero + m.insert(key_hash, b as u32); + } + + let snap = m.snapshot(); + + // One insert into bucket 0's key range. + let target_key_hash = 0u64; // bucket_of(0) == 0 + m.insert(target_key_hash, 12345); + + let mut unchanged_buckets = 0usize; + for i in 0..NUM_BUCKETS { + if Arc::ptr_eq(&snap.buckets[i], &m.buckets[i]) { + unchanged_buckets += 1; + } + } + assert!( + unchanged_buckets >= NUM_BUCKETS - 1, + "expected at most 1 bucket to have cloned (bucket-scoped CoW), \ + but only {unchanged_buckets}/{NUM_BUCKETS} buckets stayed pointer-equal" + ); + // The touched bucket (0) must actually have diverged (proves the + // write really happened, not that make_mut was a no-op). + assert!( + !Arc::ptr_eq(&snap.buckets[0], &m.buckets[0]), + "bucket 0 should have cloned since the snapshot pinned its old Arc" + ); + + // Sanity: the live map has the new value, the snapshot has the old one. + assert_eq!(m.get(&target_key_hash), Some(&12345)); + assert_eq!(snap.get(&target_key_hash), Some(&0)); + } + + #[test] + fn clone_does_not_require_value_clone_bound() { + // A type that is NOT Clone — proves BucketedKeyMap::clone() (used by + // `snapshot()`) never requires `V: Clone`. + struct NotClone(#[allow(dead_code)] u8); + + let m: BucketedKeyMap = BucketedKeyMap::new(); + let cloned = m.clone(); // must compile without `V: Clone` + assert_eq!(cloned.len(), 0); + } + + #[test] + fn default_is_empty() { + let m: BucketedKeyMap = BucketedKeyMap::default(); + assert!(m.is_empty()); + } +} diff --git a/src/vector/mod.rs b/src/vector/mod.rs index a9cee433..8abe0ccb 100644 --- a/src/vector/mod.rs +++ b/src/vector/mod.rs @@ -9,6 +9,7 @@ pub mod filter; pub mod fusion; pub mod hnsw; pub mod index_persist; +pub mod keymap; pub mod metrics; pub mod mvcc; pub mod persistence; From 6316c9297f8d9783505e9e12f77c2274193eda7f Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 19:21:32 +0700 Subject: [PATCH 2/8] fix(vector): bucket-scoped CoW for keymap fields Closes the Arc full-clone amplification under live search snapshots (see previous commit for BucketedKeyMap rationale). Converts VectorIndex's three key_hash -> V maps from Arc> to BucketedKeyMap: key_hash_to_key: Arc> -> BucketedKeyMap key_hash_to_global_id: Arc> -> BucketedKeyMap key_hash_to_vec_checksum: Arc> -> BucketedKeyMap and SearchSnapshot.key_hash_to_key to match. All call sites that used Arc::make_mut(&mut idx.key_hash_to_*).insert/remove(...) now call .insert/.remove(...) directly on the BucketedKeyMap (the bucket-scoped CoW happens inside insert()); two SPSC insert paths use the new get_or_insert_with() helper in place of the former .entry(...).or_insert_with(...) pattern. Read paths (FT.SEARCH response building, session filtering, hybrid search, recovery) take &BucketedKeyMap instead of &HashMap; these are read-only signature changes with no behavior change since BucketedKeyMap mirrors HashMap's get/iter/len/is_empty API. On-disk keymap.bin format is unchanged: run_snapshot_job's persistence loop only calls .iter()/.get(), and recover_v2.rs rebuilds the map from the same on-disk entries via a plain insert loop. Touches: src/vector/store.rs, src/vector/segment/holder.rs, src/command/vector_search/{ft_search/execute,ft_search/response, session,hybrid,hybrid_multi,tests}.rs, src/shard/spsc_handler.rs, src/shard/scatter_hybrid.rs, src/vector/persistence/recover_v2.rs. author: Tin Dang --- .../vector_search/ft_search/execute.rs | 3 +- .../vector_search/ft_search/response.rs | 5 ++- src/command/vector_search/hybrid.rs | 19 +++----- src/command/vector_search/hybrid_multi.rs | 7 +-- src/command/vector_search/session.rs | 5 ++- src/command/vector_search/tests.rs | 33 +++++++------- src/shard/scatter_hybrid.rs | 2 +- src/shard/spsc_handler.rs | 26 +++++------ src/vector/persistence/recover_v2.rs | 13 +++--- src/vector/segment/holder.rs | 8 ++-- src/vector/store.rs | 45 ++++++++++--------- 11 files changed, 87 insertions(+), 79 deletions(-) diff --git a/src/command/vector_search/ft_search/execute.rs b/src/command/vector_search/ft_search/execute.rs index e22aef86..ed545d6e 100644 --- a/src/command/vector_search/ft_search/execute.rs +++ b/src/command/vector_search/ft_search/execute.rs @@ -15,6 +15,7 @@ use smallvec::SmallVec; use crate::protocol::Frame; use crate::vector::filter::FilterExpr; +use crate::vector::keymap::BucketedKeyMap; use crate::vector::store::VectorStore; use crate::vector::types::{DistanceMetric, SearchResult}; @@ -25,7 +26,7 @@ use super::response::build_search_response; pub(super) enum SearchRawResult { Ok { results: SmallVec<[SearchResult; 32]>, - key_hash_to_key: std::sync::Arc>, + key_hash_to_key: BucketedKeyMap, }, Error(Frame), } diff --git a/src/command/vector_search/ft_search/response.rs b/src/command/vector_search/ft_search/response.rs index 6d1a1999..4eee5198 100644 --- a/src/command/vector_search/ft_search/response.rs +++ b/src/command/vector_search/ft_search/response.rs @@ -14,6 +14,7 @@ use smallvec::SmallVec; use crate::protocol::Frame; use crate::vector::filter::FilterExpr; +use crate::vector::keymap::BucketedKeyMap; use crate::vector::types::SearchResult; use super::parse::{extract_param_blob, parse_filter_clause, parse_knn_query, parse_limit_clause}; @@ -30,7 +31,7 @@ use super::parse::{extract_param_blob, parse_filter_clause, parse_knn_query, par /// (e.g., legacy data restored from a snapshot without the key map). pub(crate) fn build_search_response( results: &SmallVec<[SearchResult; 32]>, - key_hash_to_key: &std::collections::HashMap, + key_hash_to_key: &BucketedKeyMap, offset: usize, count: usize, ) -> Frame { @@ -221,7 +222,7 @@ pub fn parse_ft_search_args( /// that stop at `total` document pairs remain backward-compatible. pub(crate) fn build_hybrid_response( results: &[SearchResult], - key_hash_to_key: &std::collections::HashMap, + key_hash_to_key: &BucketedKeyMap, dense_count: usize, sparse_count: usize, offset: usize, diff --git a/src/command/vector_search/hybrid.rs b/src/command/vector_search/hybrid.rs index 98f59b48..386336df 100644 --- a/src/command/vector_search/hybrid.rs +++ b/src/command/vector_search/hybrid.rs @@ -40,6 +40,7 @@ use bytes::Bytes; use crate::protocol::{Frame, FrameVec}; use crate::text::store::{TextIndex, TextSearchResult, TextStore}; +use crate::vector::keymap::BucketedKeyMap; use crate::vector::store::VectorStore; use crate::vector::types::{SearchResult, VectorId}; @@ -513,13 +514,7 @@ pub(super) fn run_dense_knn( k: usize, as_of_lsn: u64, committed: &roaring::RoaringTreemap, -) -> Result< - ( - Vec, - std::sync::Arc>, - ), - Frame, -> { +) -> Result<(Vec, BucketedKeyMap), Frame> { let field_opt = if field_name.is_empty() { None } else { @@ -643,7 +638,7 @@ pub(super) fn run_dense_knn( #[allow(clippy::too_many_arguments)] pub fn build_hybrid_local_response( fused: &[SearchResult], - vector_key_hash_to_key: &std::collections::HashMap, + vector_key_hash_to_key: &BucketedKeyMap, text_index: &TextIndex, bm25_count: usize, dense_count: usize, @@ -687,7 +682,7 @@ pub fn build_hybrid_local_response( pub(super) fn resolve_hybrid_doc_key( r: &SearchResult, - vector_map: &std::collections::HashMap, + vector_map: &BucketedKeyMap, text_index: &TextIndex, ) -> Bytes { if r.key_hash != 0 { @@ -1233,7 +1228,7 @@ mod tests { fn test_build_hybrid_response_shape() { // Pure-function test: fused results → response shape matches FT.SEARCH convention // [total, key, [__rrf_score, "N"], ..., bm25_hits, N, dense_hits, N, sparse_hits, N] - let mut key_map = std::collections::HashMap::new(); + let mut key_map = BucketedKeyMap::new(); key_map.insert(100u64, Bytes::from_static(b"doc:alpha")); key_map.insert(200u64, Bytes::from_static(b"doc:beta")); @@ -1305,7 +1300,7 @@ mod tests { #[cfg(feature = "text-index")] fn test_build_hybrid_response_vec_fallback() { // key_hash present but not in vector map AND not in text index → vec: fallback. - let key_map = std::collections::HashMap::new(); + let key_map: BucketedKeyMap = BucketedKeyMap::new(); let fused = vec![SearchResult { distance: -0.01, id: VectorId(77), @@ -1331,7 +1326,7 @@ mod tests { #[cfg(feature = "text-index")] fn test_build_hybrid_response_pagination() { // offset skips results; count truncates. - let mut key_map = std::collections::HashMap::new(); + let mut key_map = BucketedKeyMap::new(); for i in 0..5 { key_map.insert(i as u64, Bytes::from(format!("k{i}"))); } diff --git a/src/command/vector_search/hybrid_multi.rs b/src/command/vector_search/hybrid_multi.rs index 74130394..686a0406 100644 --- a/src/command/vector_search/hybrid_multi.rs +++ b/src/command/vector_search/hybrid_multi.rs @@ -20,6 +20,7 @@ use bytes::Bytes; use crate::protocol::{Frame, FrameVec}; use crate::text::store::{TextIndex, TextStore}; +use crate::vector::keymap::BucketedKeyMap; use crate::vector::store::VectorStore; use crate::vector::types::{SearchResult, VectorId}; @@ -209,7 +210,7 @@ pub fn encode_shard_hybrid_partial( dense: &[SearchResult], sparse: &[SearchResult], text_index: &TextIndex, - vector_key_hash_to_key: &std::collections::HashMap, + vector_key_hash_to_key: &BucketedKeyMap, ) -> Frame { let total_items = 3 + 4 * (bm25.len() + dense.len() + sparse.len()); let mut out: Vec = Vec::with_capacity(total_items); @@ -317,7 +318,7 @@ mod tests { let dense = vec![tsr(0.1, 3, 0xCC)]; let sparse = vec![tsr(0.2, 4, 0xDD), tsr(0.3, 5, 0xEE)]; - let mut vec_map: std::collections::HashMap = std::collections::HashMap::new(); + let mut vec_map: BucketedKeyMap = BucketedKeyMap::new(); vec_map.insert(0xCC, Bytes::from_static(b"doc:three")); vec_map.insert(0xDD, Bytes::from_static(b"doc:four")); vec_map.insert(0xEE, Bytes::from_static(b"doc:five")); @@ -370,7 +371,7 @@ mod tests { ))], crate::text::types::BM25Config::default(), ); - let vec_map: std::collections::HashMap = std::collections::HashMap::new(); + let vec_map: BucketedKeyMap = BucketedKeyMap::new(); let frame = encode_shard_hybrid_partial(&[], &[], &[], &text_index, &vec_map); let decoded = decode_shard_hybrid_partial(&frame).expect("decode empty"); assert!(decoded.bm25.is_empty()); diff --git a/src/command/vector_search/session.rs b/src/command/vector_search/session.rs index 19cf6e27..5846624b 100644 --- a/src/command/vector_search/session.rs +++ b/src/command/vector_search/session.rs @@ -12,6 +12,7 @@ use smallvec::SmallVec; use std::collections::HashMap; use crate::storage::db::Database; +use crate::vector::keymap::BucketedKeyMap; use crate::vector::types::SearchResult; /// Filter search results, removing any whose Redis key already exists in the @@ -26,7 +27,7 @@ use crate::vector::types::SearchResult; pub fn filter_session_results( results: &SmallVec<[SearchResult; 32]>, session_members: &HashMap, - key_hash_to_key: &HashMap, + key_hash_to_key: &BucketedKeyMap, ) -> SmallVec<[SearchResult; 32]> { if session_members.is_empty() { return results.clone(); @@ -59,7 +60,7 @@ pub fn record_session_results( results: &SmallVec<[SearchResult; 32]>, db: &mut Database, session_key: &[u8], - key_hash_to_key: &HashMap, + key_hash_to_key: &BucketedKeyMap, timestamp: f64, ) { if results.is_empty() { diff --git a/src/command/vector_search/tests.rs b/src/command/vector_search/tests.rs index 5aaff66e..87652557 100644 --- a/src/command/vector_search/tests.rs +++ b/src/command/vector_search/tests.rs @@ -1206,12 +1206,12 @@ fn test_parse_limit_zero() { #[test] fn test_build_search_response_paginated() { + use crate::vector::keymap::BucketedKeyMap; use crate::vector::types::{SearchResult, VectorId}; - use std::collections::HashMap; // Create 10 fake results let mut results: SmallVec<[SearchResult; 32]> = SmallVec::new(); - let mut key_map = HashMap::new(); + let mut key_map: BucketedKeyMap = BucketedKeyMap::new(); for i in 0u32..10 { results.push(SearchResult { id: VectorId(i), @@ -1245,8 +1245,8 @@ fn test_build_search_response_paginated() { #[test] fn test_build_search_response_limit_zero_zero() { + use crate::vector::keymap::BucketedKeyMap; use crate::vector::types::{SearchResult, VectorId}; - use std::collections::HashMap; let mut results: SmallVec<[SearchResult; 32]> = SmallVec::new(); for i in 0u32..5 { @@ -1256,7 +1256,7 @@ fn test_build_search_response_limit_zero_zero() { key_hash: 0, }); } - let key_map = HashMap::new(); + let key_map: BucketedKeyMap = BucketedKeyMap::new(); // LIMIT 0 0 -> count only, no docs let response = build_search_response(&results, &key_map, 0, 0); @@ -2101,6 +2101,7 @@ fn test_session_parse_session_clause_no_key() { #[test] fn test_session_filter_results_empty_session() { + use crate::vector::keymap::BucketedKeyMap; use crate::vector::types::{SearchResult, VectorId}; use std::collections::HashMap; @@ -2117,7 +2118,7 @@ fn test_session_filter_results_empty_session() { }, ]; let session_members: HashMap = HashMap::new(); - let mut key_hash_to_key = HashMap::new(); + let mut key_hash_to_key: BucketedKeyMap = BucketedKeyMap::new(); key_hash_to_key.insert(100u64, Bytes::from_static(b"doc:a")); key_hash_to_key.insert(200u64, Bytes::from_static(b"doc:b")); @@ -2127,6 +2128,7 @@ fn test_session_filter_results_empty_session() { #[test] fn test_session_filter_results_removes_seen() { + use crate::vector::keymap::BucketedKeyMap; use crate::vector::types::{SearchResult, VectorId}; use std::collections::HashMap; @@ -2150,7 +2152,7 @@ fn test_session_filter_results_removes_seen() { let mut session_members: HashMap = HashMap::new(); session_members.insert(Bytes::from_static(b"doc:a"), 1000.0); - let mut key_hash_to_key = HashMap::new(); + let mut key_hash_to_key: BucketedKeyMap = BucketedKeyMap::new(); key_hash_to_key.insert(100u64, Bytes::from_static(b"doc:a")); key_hash_to_key.insert(200u64, Bytes::from_static(b"doc:b")); key_hash_to_key.insert(300u64, Bytes::from_static(b"doc:c")); @@ -2163,8 +2165,8 @@ fn test_session_filter_results_removes_seen() { #[test] fn test_session_record_results() { + use crate::vector::keymap::BucketedKeyMap; use crate::vector::types::{SearchResult, VectorId}; - use std::collections::HashMap; let mut db = crate::storage::db::Database::new(); let results: SmallVec<[SearchResult; 32]> = smallvec::smallvec![ @@ -2179,7 +2181,7 @@ fn test_session_record_results() { key_hash: 200 }, ]; - let mut key_hash_to_key = HashMap::new(); + let mut key_hash_to_key: BucketedKeyMap = BucketedKeyMap::new(); key_hash_to_key.insert(100u64, Bytes::from_static(b"doc:a")); key_hash_to_key.insert(200u64, Bytes::from_static(b"doc:b")); @@ -2813,7 +2815,8 @@ fn insert_hybrid_doc( drop(snap); // Record key mapping - std::sync::Arc::make_mut(&mut idx.key_hash_to_key).insert(key_hash, Bytes::from(key.to_vec())); + idx.key_hash_to_key + .insert(key_hash, Bytes::from(key.to_vec())); // Insert sparse vector if let Some(ss) = idx.sparse_stores.get_mut(b"sparse_vec".as_ref()) { @@ -3351,7 +3354,7 @@ fn test_recommend_basic_with_vectors() { quantize_f32_to_sq(v, &mut sq); snap.mutable.append(key_hash, v, i as u64); drop(snap); - std::sync::Arc::make_mut(&mut idx.key_hash_to_key) + idx.key_hash_to_key .insert(key_hash, Bytes::from(key.to_vec())); } } @@ -3521,8 +3524,8 @@ fn test_ft_dropindex_dd_deletes_docs() { if let Some(idx) = store.get_index_mut(b"ddtest") { let h1 = xxhash_rust::xxh64::xxh64(&key1, 0); let h2 = xxhash_rust::xxh64::xxh64(&key2, 0); - std::sync::Arc::make_mut(&mut idx.key_hash_to_key).insert(h1, key1.clone()); - std::sync::Arc::make_mut(&mut idx.key_hash_to_key).insert(h2, key2.clone()); + idx.key_hash_to_key.insert(h1, key1.clone()); + idx.key_hash_to_key.insert(h2, key2.clone()); } // Verify keys exist in database @@ -3590,7 +3593,7 @@ fn test_ft_dropindex_preserves_docs() { // Register key in vector index if let Some(idx) = store.get_index_mut(b"preservetest") { let h1 = xxhash_rust::xxh64::xxh64(&key1, 0); - std::sync::Arc::make_mut(&mut idx.key_hash_to_key).insert(h1, key1.clone()); + idx.key_hash_to_key.insert(h1, key1.clone()); } // Drop index WITHOUT DD flag (using None for db since we don't need it) @@ -3642,7 +3645,7 @@ fn test_ft_dropindex_dd_case_insensitive() { let key = Bytes::from_static(b"c1:doc"); db.set(key.clone(), crate::storage::entry::Entry::new_hash()); if let Some(idx) = store.get_index_mut(b"casetest1") { - std::sync::Arc::make_mut(&mut idx.key_hash_to_key) + idx.key_hash_to_key .insert(xxhash_rust::xxh64::xxh64(&key, 0), key.clone()); } @@ -3692,7 +3695,7 @@ fn test_ft_dropindex_dd_case_insensitive() { let key = Bytes::from_static(b"c2:doc"); db.set(key.clone(), crate::storage::entry::Entry::new_hash()); if let Some(idx) = store.get_index_mut(b"casetest2") { - std::sync::Arc::make_mut(&mut idx.key_hash_to_key) + idx.key_hash_to_key .insert(xxhash_rust::xxh64::xxh64(&key, 0), key.clone()); } diff --git a/src/shard/scatter_hybrid.rs b/src/shard/scatter_hybrid.rs index f5f6af05..27b9892c 100644 --- a/src/shard/scatter_hybrid.rs +++ b/src/shard/scatter_hybrid.rs @@ -640,7 +640,7 @@ mod tests { .doc_id_to_key .insert(2, Bytes::from_static(b"doc:second")); - let mut vec_map = std::collections::HashMap::new(); + let mut vec_map = crate::vector::keymap::BucketedKeyMap::new(); vec_map.insert(300u64, Bytes::from_static(b"doc:dense-only")); vec_map.insert(400u64, Bytes::from_static(b"doc:sparse-only")); diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index d96b7f1a..0de13410 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -2880,11 +2880,11 @@ fn handle_vector_insert( for chunk in blob.chunks_exact(4) { f32_vec.push(f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]])); } - // Record original Redis key for FT.SEARCH response. COW via make_mut: - // clones only if a search snapshot holds the map concurrently (QP-1). - std::sync::Arc::make_mut(&mut idx.key_hash_to_key) - .entry(key_hash) - .or_insert_with(|| bytes::Bytes::copy_from_slice(key)); + // Record original Redis key for FT.SEARCH response. Bucket-scoped COW: + // clones only the ONE bucket if a search snapshot holds the map + // concurrently (QP-1 + RSS/CPU wave 4). + idx.key_hash_to_key + .get_or_insert_with(key_hash, || bytes::Bytes::copy_from_slice(key)); // Append to mutable segment. `insert_lsn` is the monotonic LSN allocated // by `auto_index_hset`; MVCC visibility (src/vector/mvcc/visibility.rs) // compares against query snapshot_lsn to enforce FT.SEARCH AS_OF and @@ -2931,12 +2931,12 @@ fn handle_vector_insert( crate::vector::metrics::add_vectors(1); // Record key_hash → global_id mapping for future metadata-only updates. - // COW via make_mut, mirroring key_hash_to_key (QP-1). - std::sync::Arc::make_mut(&mut idx.key_hash_to_global_id).insert(key_hash, global_id); + // Bucket-scoped COW, mirroring key_hash_to_key (QP-1 + RSS/CPU wave 4). + idx.key_hash_to_global_id.insert(key_hash, global_id); // B2 (durability): mirror the same key_hash into the checksum map so the // two maps never drift — the B3 dedup rescan compares this checksum // against a freshly-hashed current value to decide unchanged-vs-changed. - std::sync::Arc::make_mut(&mut idx.key_hash_to_vec_checksum) + idx.key_hash_to_vec_checksum .insert(key_hash, xxhash_rust::xxh64::xxh64(&blob, 0)); // Populate payload index with all HASH fields (for filtered search) @@ -2979,11 +2979,11 @@ fn handle_vector_insert_field( for chunk in blob.chunks_exact(4) { f32_vec.push(f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]])); } - // Record original Redis key (shared across all fields). COW via make_mut: - // clones only if a search snapshot holds the map concurrently (QP-1). - std::sync::Arc::make_mut(&mut idx.key_hash_to_key) - .entry(key_hash) - .or_insert_with(|| bytes::Bytes::copy_from_slice(key)); + // Record original Redis key (shared across all fields). Bucket-scoped + // COW: clones only the ONE bucket if a search snapshot holds the map + // concurrently (QP-1 + RSS/CPU wave 4). + idx.key_hash_to_key + .get_or_insert_with(key_hash, || bytes::Bytes::copy_from_slice(key)); // Look up the additional field's SegmentHolder let fs = match idx.field_segments.get(field_name.as_ref()) { diff --git a/src/vector/persistence/recover_v2.rs b/src/vector/persistence/recover_v2.rs index a855804a..75246b08 100644 --- a/src/vector/persistence/recover_v2.rs +++ b/src/vector/persistence/recover_v2.rs @@ -46,6 +46,7 @@ use tracing::{info, warn}; use crate::protocol::Frame; use crate::text::store::TextStore; +use crate::vector::keymap::BucketedKeyMap; use crate::vector::persistence::manifest::{self, IndexManifest}; use crate::vector::persistence::segment_io; use crate::vector::segment::SegmentList; @@ -414,9 +415,9 @@ fn load_segments_and_keymap( let entries = manifest::read_keymap_tolerant(idx_dir, manifest.keymap_epoch).unwrap_or_default(); - let mut key_hash_to_key = std::collections::HashMap::with_capacity(entries.len()); - let mut key_hash_to_global_id = std::collections::HashMap::with_capacity(entries.len()); - let mut key_hash_to_vec_checksum = std::collections::HashMap::with_capacity(entries.len()); + let mut key_hash_to_key = BucketedKeyMap::new(); + let mut key_hash_to_global_id = BucketedKeyMap::new(); + let mut key_hash_to_vec_checksum = BucketedKeyMap::new(); for e in entries { if dropped_key_hashes.contains(&e.key_hash) { continue; @@ -441,9 +442,9 @@ fn load_segments_and_keymap( warm: Vec::new(), cold: Vec::new(), }); - idx.key_hash_to_key = Arc::new(key_hash_to_key); - idx.key_hash_to_global_id = Arc::new(key_hash_to_global_id); - idx.key_hash_to_vec_checksum = Arc::new(key_hash_to_vec_checksum); + idx.key_hash_to_key = key_hash_to_key; + idx.key_hash_to_global_id = key_hash_to_global_id; + idx.key_hash_to_vec_checksum = key_hash_to_vec_checksum; Some(IndexRecoveryCounters { loaded_segments, diff --git a/src/vector/segment/holder.rs b/src/vector/segment/holder.rs index 7131570f..5502102e 100644 --- a/src/vector/segment/holder.rs +++ b/src/vector/segment/holder.rs @@ -12,6 +12,7 @@ use smallvec::SmallVec; use crate::vector::diskann::segment::DiskAnnSegment; use crate::vector::filter::selectivity::{FilterStrategy, select_strategy}; use crate::vector::hnsw::search::SearchScratch; +use crate::vector::keymap::BucketedKeyMap; use crate::vector::persistence::warm_search::WarmSearchSegment; use crate::vector::segment::ivf::IvfSegment; use crate::vector::turbo_quant::encoder::padded_dimension; @@ -156,9 +157,10 @@ pub struct SearchSnapshot { pub scratch: SearchScratch, /// Key-hash → key map captured at START (§3 C1) so a mid-search delete cannot /// drop an entry this search still needs to resolve. Used by the response - /// builder, not the segment scan. `Arc` snapshot: capture is O(1); writers - /// copy-on-write via `Arc::make_mut` (QP-1). - pub key_hash_to_key: std::sync::Arc>, + /// builder, not the segment scan. Bucketed-CoW snapshot: capture is O(256) + /// (refcount bumps only); writers copy-on-write bucket-scoped via + /// `Arc::make_mut` on the single touched bucket (QP-1 + RSS/CPU wave 4). + pub key_hash_to_key: BucketedKeyMap, /// AE-1: true when `ef_search` came from the resolution heuristic (no /// user `EF_RUNTIME`) — saturation-certified segments may then run at /// their min-ef estimate. diff --git a/src/vector/store.rs b/src/vector/store.rs index e671ea56..cf9ccb85 100644 --- a/src/vector/store.rs +++ b/src/vector/store.rs @@ -12,6 +12,7 @@ use bytes::Bytes; use crate::storage::tiered::SegmentHandle; use crate::vector::filter::PayloadIndex; use crate::vector::hnsw::search::SearchScratch; +use crate::vector::keymap::BucketedKeyMap; use crate::vector::mvcc::manager::TransactionManager; use crate::vector::segment::compaction; use crate::vector::segment::{SegmentHolder, SegmentList}; @@ -191,22 +192,24 @@ pub struct VectorIndex { /// `vec:` form. Survives compaction and segment merging because /// it's keyed by the stable `key_hash`, not the volatile internal ID. /// - /// `Arc`-wrapped so search snapshots capture it in O(1) (QP-1: the previous - /// owned `HashMap` was deep-cloned per query — one entry per indexed vector). - /// Writers mutate via [`Arc::make_mut`]: copy-on-write triggers only when a - /// snapshot is concurrently alive, at most once per snapshot lifetime. - pub key_hash_to_key: Arc>, + /// Bucketed CoW (RSS/CPU wave 4, defect 1): 256 independent + /// `Arc` shards keyed by the top bits of `key_hash`, so search + /// snapshots still capture the whole map in O(1) (QP-1) but a concurrent + /// writer's `Arc::make_mut` only clones the ONE bucket its key hashes + /// into — not the entire multi-MB map — even while a search snapshot is + /// alive. See `crate::vector::keymap` module docs. + pub key_hash_to_key: BucketedKeyMap, /// Maps `key_hash` → `global_id` for metadata-only updates. /// /// When `HSET doc:1 category "science"` is called without a vector blob, /// the auto-indexer looks up the existing `global_id` here to update the /// PayloadIndex for that vector without re-inserting it. /// - /// `Arc`-wrapped for the same reason as `key_hash_to_key` (QP-1 O(1) - /// snapshot capture) — the durability write path (B2) clones this Arc - /// into background keymap-snapshot jobs without an O(n) copy on the - /// shard thread. - pub key_hash_to_global_id: Arc>, + /// Bucketed CoW for the same reason as `key_hash_to_key` (QP-1 O(1) + /// snapshot capture, bucket-scoped write clone) — the durability write + /// path (B2) clones this into background keymap-snapshot jobs without an + /// O(n) copy on the shard thread. + pub key_hash_to_global_id: BucketedKeyMap, /// Maps `key_hash` → `xxh64(vector-field bytes, seed 0)` (B2, durability). /// /// Computed where the raw HSET vector blob is in hand @@ -216,7 +219,7 @@ pub struct VectorIndex { /// rescan can tell "unchanged" keys (checksum matches → metadata-only /// rebuild) from "changed" keys (re-encode) without hashing every value /// twice across a restart. - pub key_hash_to_vec_checksum: Arc>, + pub key_hash_to_vec_checksum: BucketedKeyMap, /// Shard-relative directory this index persists segments/manifest/keymap /// under, i.e. `/idx-/` (B2). /// `None` when the store has no `persist_dir` configured (disk-offload @@ -1572,9 +1575,9 @@ impl VectorStore { scratch, collection, payload_index: PayloadIndex::new(), - key_hash_to_key: Arc::new(std::collections::HashMap::new()), - key_hash_to_global_id: Arc::new(std::collections::HashMap::new()), - key_hash_to_vec_checksum: Arc::new(std::collections::HashMap::new()), + key_hash_to_key: BucketedKeyMap::new(), + key_hash_to_global_id: BucketedKeyMap::new(), + key_hash_to_vec_checksum: BucketedKeyMap::new(), persist_dir: self.persist_dir.clone(), next_segment_id: floor_next_segment_id, next_snapshot_seq: floor_next_snapshot_seq, @@ -1700,9 +1703,9 @@ impl VectorStore { scratch, collection, payload_index: PayloadIndex::new(), - key_hash_to_key: Arc::new(std::collections::HashMap::new()), - key_hash_to_global_id: Arc::new(std::collections::HashMap::new()), - key_hash_to_vec_checksum: Arc::new(std::collections::HashMap::new()), + key_hash_to_key: BucketedKeyMap::new(), + key_hash_to_global_id: BucketedKeyMap::new(), + key_hash_to_vec_checksum: BucketedKeyMap::new(), persist_dir: self.persist_dir.clone(), next_segment_id: manifest.next_segment_id, next_snapshot_seq: manifest.keymap_epoch, @@ -1916,12 +1919,12 @@ impl VectorStore { // they track LIVE keys, not historical inserts — without this // they grow monotonically under key churn (~1GB / 24M deletes). // A re-insert of the same key repopulates all three maps. - Arc::make_mut(&mut idx.key_hash_to_key).remove(&key_hash); - Arc::make_mut(&mut idx.key_hash_to_global_id).remove(&key_hash); + idx.key_hash_to_key.remove(&key_hash); + idx.key_hash_to_global_id.remove(&key_hash); // B2 (durability): keep the checksum map in lockstep so it never // drifts from key_hash_to_key (a stale entry would be a silent // false-"unchanged" in the B3 dedup rescan). - Arc::make_mut(&mut idx.key_hash_to_vec_checksum).remove(&key_hash); + idx.key_hash_to_vec_checksum.remove(&key_hash); true } @@ -2292,7 +2295,7 @@ impl VectorStore { .load() .mutable .append(key_hash, vector, insert_lsn); - Arc::make_mut(&mut idx.key_hash_to_key).insert(key_hash, key); + idx.key_hash_to_key.insert(key_hash, key); Ok(()) } From c8f35b5ad35742d27901062bb0771abbc0626e51 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 19:30:04 +0700 Subject: [PATCH 3/8] =?UTF-8?q?perf(vector):=20coalescing=20SnapshotPool?= =?UTF-8?q?=20=E2=80=94=20bound=20the=20durability=20write=20queue?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SnapshotPool used an unbounded flume::unbounded::() queue drained by a single worker. If compact/merge cadence outpaced the worker, queued jobs pinned every submitter's Arcs (now BucketedKeyMap buckets) indefinitely, growing unboundedly under sustained load. Replaced the channel with a coalescing slot: Mutex> + Condvar, keyed by the job's idx_dir (already unique per index and present on every SnapshotJob, so no new key type is needed). submit() inserts/overwrites the pending entry for that index and notifies one worker; a resubmit for an index whose prior job is still pending (not yet dequeued) replaces it in place, dropping the stale job's data immediately, so only the newest state per index is ever persisted. A job already dequeued and running is unaffected by a concurrent resubmit — the worker loop only removes an entry from the map at the moment it starts running that job. submit() never blocks the caller (map insert + notify, no waiting on capacity). SnapshotJob's three fields are now BucketedKeyMap (no Arc wrapper) to match VectorIndex's field types from the previous commit. run_snapshot_job's body is unchanged since it only calls .iter()/.get() on the map. 3 new tests: same-index submits under a slow worker execute far fewer than the number submitted (final persisted state is the newest job); jobs for distinct indexes all execute; an in-flight job survives a resubmit for the same index (the running job completes normally, the resubmit becomes the new pending entry). Also fixes 2 doc_lazy_continuation clippy lints in the new SnapshotPool doc comment. author: Tin Dang --- src/vector/persistence/manifest.rs | 333 +++++++++++++++++++++++++---- 1 file changed, 293 insertions(+), 40 deletions(-) diff --git a/src/vector/persistence/manifest.rs b/src/vector/persistence/manifest.rs index cb08edfe..3a13691e 100644 --- a/src/vector/persistence/manifest.rs +++ b/src/vector/persistence/manifest.rs @@ -32,7 +32,7 @@ //! can never overwrite a newer, already-committed manifest — see that //! function's doc comment for the full argument. -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::fs; use std::io; use std::path::{Path, PathBuf}; @@ -40,10 +40,11 @@ use std::sync::Arc; use std::sync::OnceLock; use bytes::Bytes; -use parking_lot::Mutex; +use parking_lot::{Condvar, Mutex}; use serde::{Deserialize, Serialize}; use crate::persistence::fsync::{fsync_directory, fsync_file}; +use crate::vector::keymap::BucketedKeyMap; /// Current on-disk manifest format version. pub const MANIFEST_FORMAT_VERSION: u32 = 1; @@ -414,9 +415,9 @@ pub struct SnapshotJob { /// The manifest to commit if this job is not stale. `keymap_epoch` MUST /// equal `seq`. pub manifest: IndexManifest, - pub key_hash_to_key: Arc>, - pub key_hash_to_global_id: Arc>, - pub key_hash_to_vec_checksum: Arc>, + pub key_hash_to_key: BucketedKeyMap, + pub key_hash_to_global_id: BucketedKeyMap, + pub key_hash_to_vec_checksum: BucketedKeyMap, } /// Run one snapshot job to completion. @@ -534,48 +535,126 @@ pub fn run_snapshot_job(job: SnapshotJob) { /// `background_compact::BackgroundCompactor` (different job shape — this is /// small, fsync-bound I/O, not CPU-bound HNSW build work) but the same /// lazy-init-singleton shape. +/// +/// ## Coalescing (RSS/CPU wave 4, defect 2) +/// +/// A [`SnapshotJob`] is a FULL-STATE snapshot of one index's keymap + +/// manifest — a newer job for the same index strictly supersedes an older +/// one that hasn't started yet. The original design used an *unbounded* +/// `flume` channel: if compact/merge installs outpaced the single worker, +/// every queued job kept its three `BucketedKeyMap` clones pinned at a +/// refcount above one, so EVERY subsequent write on the shard thread paid +/// the bucket-clone CoW cost until the backlog drained — unbounded, and in +/// the worst case (a burst of installs) effectively never. +/// +/// Instead, pending jobs live in a `Mutex>` +/// keyed by the index's persist directory (unique per index). `submit` +/// inserts under that key: a job already pending for the same index is +/// REPLACED (the old job's `BucketedKeyMap`s are dropped immediately, +/// releasing any pinned Arcs) rather than queued behind it. A single worker +/// thread drains the map via a `Condvar`; an in-flight (already dequeued, +/// currently executing) job is never affected by a later `submit` for the +/// same index — the next submit simply becomes the new pending entry. pub struct SnapshotPool { - job_tx: flume::Sender, + /// Pending jobs, keyed by index persist directory. At most one entry per + /// index — the latest submitted, not-yet-started job for that index. + pending: Arc>>, + /// Signalled on every `submit` so an idle worker wakes immediately. + not_empty: Arc, + /// Set on `Drop` so worker threads exit their wait loop instead of + /// blocking forever (matters for test-local pools; the process-wide + /// singleton is never dropped). + shutdown: Arc, _workers: Vec>, } impl SnapshotPool { fn new(num_workers: usize) -> Self { - let (job_tx, job_rx) = flume::unbounded::(); + Self::new_with_runner(num_workers, run_snapshot_job) + } + + /// Same as [`Self::new`] but with an injectable per-job runner — the + /// production path always passes [`run_snapshot_job`]; tests inject a + /// slow/instrumented stub to observe coalescing without touching disk. + fn new_with_runner(num_workers: usize, runner: F) -> Self + where + F: Fn(SnapshotJob) + Send + Sync + 'static, + { + let pending: Arc>> = + Arc::new(Mutex::new(HashMap::new())); + let not_empty = Arc::new(Condvar::new()); + let shutdown = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let runner = Arc::new(runner); let workers: Vec<_> = (0..num_workers.max(1)) .map(|i| { - let rx = job_rx.clone(); + let pending = Arc::clone(&pending); + let not_empty = Arc::clone(¬_empty); + let shutdown = Arc::clone(&shutdown); + let runner = Arc::clone(&runner); // Startup-time OS thread spawn failure is unrecoverable // (matches the existing `BackgroundCompactor::new` convention). #[allow(clippy::unwrap_used)] - let handle = std::thread::Builder::new() + std::thread::Builder::new() .name(format!("moon-vec-snapshot-{i}")) .spawn(move || { - while let Ok(job) = rx.recv() { - run_snapshot_job(job); + loop { + let next_job = { + let mut guard = pending.lock(); + loop { + if let Some(key) = guard.keys().next().cloned() { + break guard.remove(&key); + } + if shutdown.load(std::sync::atomic::Ordering::Acquire) { + break None; + } + not_empty.wait(&mut guard); + } + }; + match next_job { + Some(job) => runner(job), + None => return, // shutdown, queue drained + } } }) - .expect("failed to spawn vector snapshot worker thread"); - handle + .expect("failed to spawn vector snapshot worker thread") }) .collect(); Self { - job_tx, + pending, + not_empty, + shutdown, _workers: workers, } } - /// Enqueue a snapshot job. Best-effort: send only fails if every worker - /// has exited (e.g. all panicked), in which case the job is dropped with - /// a warning — per the crash-safety invariant this only leaves the - /// on-disk state a bit more stale (understating), never wrong. + /// Enqueue a snapshot job. Never blocks the caller (the shard event + /// loop): this only takes a `parking_lot::Mutex` for the duration of one + /// `HashMap` insert, never waits on I/O or on the worker. + /// + /// Latest-wins coalescing: if a job for the same index (`idx_dir`) is + /// already pending (submitted but not yet dequeued by the worker), it is + /// replaced — the old job, and the `Arc`s it held into the keymap + /// buckets, are dropped right here. A job the worker has already + /// dequeued and is currently executing is unaffected; it runs to + /// completion (its own staleness is still checked against the shared + /// watermark inside `run_snapshot_job`, exactly as before). pub fn submit(&self, job: SnapshotJob) { - if self.job_tx.send(job).is_err() { - tracing::warn!( - "vector snapshot pool: all workers gone — snapshot job dropped \ - (index falls behind on durability, no correctness impact)" - ); - } + let mut pending = self.pending.lock(); + pending.insert(job.idx_dir.clone(), job); + drop(pending); + self.not_empty.notify_one(); + } +} + +impl Drop for SnapshotPool { + fn drop(&mut self) { + // Only meaningful for test-local pools (the process-wide singleton + // in `global_snapshot_pool` lives in a `OnceLock` and is never + // dropped) — lets worker threads exit their wait loop cleanly + // instead of blocking forever once `pending` has no other owners. + self.shutdown + .store(true, std::sync::atomic::Ordering::Release); + self.not_empty.notify_all(); } } @@ -816,24 +895,26 @@ mod tests { assert!(dir.join("manifest.json").exists()); } - fn checksum_map(entries: &[KeymapEntry]) -> Arc> { - Arc::new( - entries - .iter() - .map(|e| (e.key_hash, e.vec_checksum)) - .collect(), - ) + fn checksum_map(entries: &[KeymapEntry]) -> BucketedKeyMap { + let mut m = BucketedKeyMap::new(); + for e in entries { + m.insert(e.key_hash, e.vec_checksum); + } + m } - fn global_id_map(entries: &[KeymapEntry]) -> Arc> { - Arc::new(entries.iter().map(|e| (e.key_hash, e.global_id)).collect()) + fn global_id_map(entries: &[KeymapEntry]) -> BucketedKeyMap { + let mut m = BucketedKeyMap::new(); + for e in entries { + m.insert(e.key_hash, e.global_id); + } + m } - fn key_map(entries: &[KeymapEntry]) -> Arc> { - Arc::new( - entries - .iter() - .map(|e| (e.key_hash, e.key.clone())) - .collect(), - ) + fn key_map(entries: &[KeymapEntry]) -> BucketedKeyMap { + let mut m = BucketedKeyMap::new(); + for e in entries { + m.insert(e.key_hash, e.key.clone()); + } + m } #[test] @@ -946,4 +1027,176 @@ mod tests { assert!(a.to_string_lossy().contains("idx-")); assert_eq!(index_name_hex(b"myidx").len(), 16); } + + // ── SnapshotPool coalescing (RSS/CPU wave 4, defect 2) ────────────────── + + /// Minimal `SnapshotJob` for pool tests — the pool's coalescing logic + /// only inspects `idx_dir` and `seq`; keymap contents/manifest shape are + /// irrelevant here (unlike `test_run_snapshot_job_*`, which exercises the + /// real I/O in `run_snapshot_job`). + fn sample_job(idx_dir: &Path, seq: u64, watermark: Arc>) -> SnapshotJob { + SnapshotJob { + idx_dir: idx_dir.to_path_buf(), + seq, + watermark, + manifest: sample_manifest("stub", vec![], seq), + key_hash_to_key: BucketedKeyMap::new(), + key_hash_to_global_id: BucketedKeyMap::new(), + key_hash_to_vec_checksum: BucketedKeyMap::new(), + } + } + + /// THE key coalescing property: 100 rapid-fire submits for the SAME + /// index against a deliberately slow worker must NOT queue (and + /// eventually run) all 100 — coalescing must collapse them down to a + /// handful of executions, and the final persisted state must be the + /// NEWEST submitted job (seq 100), never an older one. + #[test] + fn test_snapshot_pool_coalesces_same_index_submits_under_slow_worker() { + let executed_seqs: Arc>> = Arc::new(Mutex::new(Vec::new())); + let executed_seqs_cl = Arc::clone(&executed_seqs); + let same_dir = PathBuf::from("/fake/idx-same"); + + let pool = SnapshotPool::new_with_runner(1, move |job: SnapshotJob| { + // Slow-worker stub: models a worker whose fsync-bound I/O is + // slower than the shard thread's submit rate — the exact + // condition that used to pin every queued job's BucketedKeyMap + // Arcs under the old unbounded-channel design. + std::thread::sleep(std::time::Duration::from_millis(5)); + executed_seqs_cl.lock().push(job.seq); + }); + + let submit_start = std::time::Instant::now(); + for seq in 1..=100u64 { + pool.submit(sample_job(&same_dir, seq, Arc::new(Mutex::new(0)))); + } + let submit_elapsed = submit_start.elapsed(); + // The submitter must never block on the (slow) worker: 100 serial + // 5ms-per-job executions would take ~500ms; a non-blocking submit + // (an uncontended Mutex + HashMap insert) finishes in well under + // that even with generous CI scheduling slack. + assert!( + submit_elapsed < std::time::Duration::from_millis(200), + "submit() must not block on the worker — took {submit_elapsed:?} for 100 submits" + ); + + // Wait for the queue to fully drain (bounded poll, never hangs the + // test suite even if this regresses). + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + let drained = + pool.pending.lock().is_empty() && executed_seqs.lock().last().copied() == Some(100); + if drained { + break; + } + assert!( + std::time::Instant::now() < deadline, + "snapshot pool did not drain within 5s" + ); + std::thread::sleep(std::time::Duration::from_millis(5)); + } + + let seqs = executed_seqs.lock(); + assert!( + seqs.len() < 100, + "coalescing must skip most of the 100 same-index submits — executed {} jobs: {seqs:?}", + seqs.len() + ); + assert_eq!( + *seqs.last().unwrap(), + 100, + "the final executed job for a coalesced index must be the NEWEST submitted state" + ); + } + + /// Jobs for DIFFERENT indexes must never coalesce with each other — each + /// gets its own coalescing slot (keyed by `idx_dir`), so every distinct + /// index's job executes exactly once. + #[test] + fn test_snapshot_pool_executes_all_distinct_index_jobs() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let executed = Arc::new(AtomicUsize::new(0)); + let executed_cl = Arc::clone(&executed); + let pool = SnapshotPool::new_with_runner(1, move |_job: SnapshotJob| { + executed_cl.fetch_add(1, Ordering::SeqCst); + }); + + const N: usize = 10; + for i in 0..N { + let dir = PathBuf::from(format!("/fake/idx-distinct-{i}")); + pool.submit(sample_job(&dir, 1, Arc::new(Mutex::new(0)))); + } + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while executed.load(Ordering::SeqCst) < N as usize { + assert!( + std::time::Instant::now() < deadline, + "distinct-index jobs did not all execute within 5s (got {})", + executed.load(Ordering::SeqCst) + ); + std::thread::sleep(std::time::Duration::from_millis(5)); + } + assert_eq!( + executed.load(Ordering::SeqCst), + N, + "every distinct-index job must execute exactly once — no cross-index coalescing" + ); + } + + /// A job the worker has already dequeued (in-flight) must run to + /// completion even if a NEW job for the same index is submitted while it + /// executes — the new submit must land as a fresh pending entry (which + /// then also runs), never silently dropped and never merged into the + /// in-flight job. + #[test] + fn test_snapshot_pool_in_flight_job_survives_resubmit_for_same_index() { + let executed_seqs: Arc>> = Arc::new(Mutex::new(Vec::new())); + let executed_cl = Arc::clone(&executed_seqs); + let (started_tx, started_rx) = std::sync::mpsc::channel::<()>(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let release_rx = Arc::new(Mutex::new(release_rx)); + + let pool = SnapshotPool::new_with_runner(1, move |job: SnapshotJob| { + if job.seq == 1 { + // Signal "now in-flight", then block until the test says go + // — simulates a slow fsync-bound write while job 1 executes. + let _ = started_tx.send(()); + let _ = release_rx.lock().recv(); + } + executed_cl.lock().push(job.seq); + }); + + let dir = PathBuf::from("/fake/idx-inflight"); + pool.submit(sample_job(&dir, 1, Arc::new(Mutex::new(0)))); + + // Block until job 1 is confirmed in-flight (dequeued, running). + started_rx + .recv_timeout(std::time::Duration::from_secs(5)) + .expect("job 1 never started"); + + // Submit job 2 for the SAME index WHILE job 1 is still executing — + // must NOT be lost, and must NOT be merged into the in-flight job 1. + pool.submit(sample_job(&dir, 2, Arc::new(Mutex::new(0)))); + + // Let job 1 finish. + release_tx.send(()).expect("worker thread gone"); + + // Both must eventually execute, in order: job 1 (in-flight, + // unaffected by the resubmit) then job 2 (queued fresh once job 1's + // slot was empty again). + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + let seqs = executed_seqs.lock().clone(); + if seqs == vec![1, 2] { + break; + } + assert!( + std::time::Instant::now() < deadline, + "expected in-flight job 1 then resubmitted job 2, got {seqs:?}" + ); + drop(seqs); + std::thread::sleep(std::time::Duration::from_millis(5)); + } + } } From cb8b2f2c61b1e01bf8f735ab449f7d7c206a7d87 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 19:30:16 +0700 Subject: [PATCH 4/8] docs(changelog): vector keymap CoW amplification + bounded snapshot queue fix Records the RSS/CPU wave 4 defects fixed in the previous three commits under [Unreleased]: BucketedKeyMap replacing the single Arc keymap fields, and the coalescing SnapshotPool replacing the unbounded flume queue. author: Tin Dang --- CHANGELOG.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1312096f..8d0b6fc2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -46,6 +46,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 trade-off as the diskfull guard: `DEL`/`UNLINK`/`EXPIRE`/`FLUSHALL` are write-flagged and blocked too while paused — no allowlist. +### Fixed — vector keymap CoW amplification + bounded snapshot queue (PR #TBD) + +- `VectorIndex`'s three `key_hash -> V` maps (`key_hash_to_key`, `key_hash_to_global_id`, + `key_hash_to_vec_checksum`) were each a single `Arc>`. Under + `ft-search-off-eventloop`, a live FT.SEARCH snapshot holds an extra `Arc` clone of the whole + map while the event loop keeps processing writes; the next write's `Arc::make_mut` then sees + refcount > 1 and full-clones the entire map synchronously on the shard event loop (~47MB + @1M vectors). Replaced with `BucketedKeyMap` (`src/vector/keymap.rs`): 256 fixed buckets, + each independently `Arc>`, bucket selected by `(key_hash >> 56)`. A concurrent + snapshot now pins at most 1/256th of the map per write instead of the whole thing. Same + treatment applied to `SnapshotJob`'s mirrored fields and `SearchSnapshot.key_hash_to_key`. + On-disk `keymap.bin` format is unchanged (read-compatible). +- `SnapshotPool` (`src/vector/persistence/manifest.rs`) used an unbounded `flume` queue with a + single worker; if compact/merge cadence outpaced the worker, queued jobs pinned every + submitter's `Arc`s indefinitely. Replaced with a coalescing slot keyed by index directory: a + new submit for an index with a job still pending (not yet dequeued) replaces it in place, + dropping the stale job's `Arc`s immediately; an in-flight (already-dequeued) job is unaffected. + Submit never blocks the caller. + ### Fixed — zombie/CPU hardening wave 1 (PR #TBD) - **Busy-poll spin idle-disengages** (`--io-busy-poll-us` / vendored monoio From 714d3bd6247fdd25839c62a7e52086cd96c0eb7d Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 21:37:53 +0700 Subject: [PATCH 5/8] =?UTF-8?q?fix(vector):=20kill-9=20doc=20loss=20?= =?UTF-8?q?=E2=80=94=20FT.COMPACT=20residue=20drain=20+=20B3=20segment-mem?= =?UTF-8?q?bership=20gate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two pre-existing, compounding bugs (both reproduce on main; surfaced while validating this branch against crash_recovery_vector_durability — the "wave-4 regression" was actually machine-load timing): 1. force_compact early-return left mutable residue unfrozen. When an explicit FT.COMPACT found a background auto-compaction in flight, it drained that build, installed it, persisted, and RETURNED — without compacting the docs inserted while the background build ran (the clone_suffix(frozen_len) mutable tail). Those docs stayed mutable-only with no durable segment until some future compact, breaking force_compact's full-drain contract (frozen == mutable on reply). Observed stuck durable state: segments live=1961, keymap entries=2000, converging never. force_compact now falls through to the inline drain loop after installing the in-flight build (a no-op when the tail is empty). 2. B3 recovery "verified" keymap entries with no backing segment doc. The durable keymap-.bin covers EVERY indexed key (mutable + immutable) at snapshot-submit time, while the paired manifest.json lists only installed immutable segments — so after a kill -9 the on-disk keymap can be a strict SUPERSET of the on-disk segments (bug 1 makes that state persistent; even with bug 1 fixed, a crash between a segment install and its async snapshot commit still produces it transiently). The B3 dedup rescan treated a keymap checksum match as "verified unchanged" and never re-indexed those keys: their documents silently vanished from search (post-crash num_docs 1950/2000, "2000 key(s) verified unchanged, 0 re-indexed", wrong KNN top-1 — decoded from a preserved failing run's artifacts). load_segments_and_keymap now builds the union of live key_hashes across loaded segments (new ImmutableSegment::live_key_hashes, honoring install-time delete_lsn and steady-state tombstones) and drops any keymap entry not in it, so the rescan re-indexes those keys from the AOF — never wrong, only occasionally slower. A loud info line reports the dropped-phantom count. Tests (each red/green verified by toggling its fix): - unit: test_force_compact_drains_tail_inserted_during_inflight_bg_build stages bug 1 deterministically (in-flight bg build + 30 tail inserts, then force_compact; red: 30 docs left mutable, green: 0 and all 130 segment-resident). - unit: recover_reindexes_keymap_entries_not_backed_by_any_segment stages the window deterministically (rewrites the durable keymap with a checksum-matching phantom entry) and asserts the phantom is dropped at load, re-indexed on rescan, and searchable end-to-end. - integration S6 (crash_recovery_vector_durability): kills inside the async-snapshot window (manifest >= 2 segments gate, no full-durability wait) and asserts only the crash contract: num_docs == N and every key answers its own exact-match query. - S1/S2 flake fix: their re_indexed == 0 assertion silently assumed full durability at kill time; a new wait_for_durable_docs pre-kill gate (manifest'd segment live-counts AND keymap entries both == N) makes the precondition explicit — it is what exposed bug 1's never-converging state. Server stdout/stderr logs now open in append mode so the pre-crash generation's log survives restart for post-mortem diagnosis. author: Tin Dang --- CHANGELOG.md | 36 ++++ src/vector/persistence/recover_v2.rs | 178 ++++++++++++++++++++ src/vector/segment/immutable.rs | 14 ++ src/vector/store.rs | 89 ++++++++-- tests/crash_recovery_vector_durability.rs | 190 +++++++++++++++++++++- 5 files changed, 486 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d0b6fc2..180d4ab4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,42 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 dropping the stale job's `Arc`s immediately; an in-flight (already-dequeued) job is unaffected. Submit never blocks the caller. +### Fixed — FT.COMPACT left mutable residue behind when draining a background build (PR #TBD) + +- **Pre-existing** (`main`, not introduced by this branch): when an explicit + `FT.COMPACT`/`force_compact` found a background auto-compaction already in flight, it drained + that build, installed it, persisted, and returned — WITHOUT compacting the documents inserted + while the background build was running (the `clone_suffix(frozen_len)` mutable tail). Those + docs stayed mutable-only (no durable segment) until some future compact fired, breaking + force_compact's documented full-drain contract (frozen == mutable on reply) and — combined + with the B3 phantom-keymap hole below — silently losing them on a kill -9 (observed stuck + durable state: segments live=1961, keymap=2000, converging never). Load-dependent: only + reproduces when insert cadence keeps a background build in flight at FT.COMPACT time (this is + what made the crash suite's S1/S2 flake by machine load, masquerading as a branch regression). + `force_compact` now falls through to the inline drain loop after installing the in-flight + build (a no-op when the tail is empty). + +### Fixed — B3 vector recovery: phantom keymap entries silently dropped docs (PR #TBD) + +- **Pre-existing silent data loss on kill -9** (found while validating this branch against the + `crash_recovery_vector_durability` suite; reproduces on `main`): the durable + `keymap-.bin` covers EVERY indexed key (mutable + immutable) at snapshot-submit time, + while the paired `manifest.json` lists only installed immutable segments. A crash landing after + one snapshot commits but before the next (covering a just-frozen segment) leaves the on-disk + keymap a strict superset of the on-disk segments. The B3 dedup rescan then "verified" those + uncovered keys as unchanged (keymap checksum matches the AOF blob — exactly the crash-window + signature) and never re-indexed them: their documents vanished from search with + `num_docs` under-reporting (observed: 1950/2000) while recovery logged + `verified unchanged` for all keys. Fix: `recover_v2::load_segments_and_keymap` now drops any + keymap entry whose `key_hash` is not LIVE in a loaded segment + (`ImmutableSegment::live_key_hashes`), so the rescan re-indexes those keys from the AOF; a + loud log line reports the dropped-phantom count. New deterministic unit regression + (`recover_reindexes_keymap_entries_not_backed_by_any_segment`, red/green verified) plus + integration scenario S6 (kill inside the async-snapshot window; asserts only the crash + contract: `num_docs == N`, every key answers its own exact-match query). S1/S2's + `re_indexed == 0` fast-path determinism is restored by a new full-durability pre-kill gate + (`wait_for_durable_docs`: manifest'd segment live-counts AND keymap entries both equal N). + ### Fixed — zombie/CPU hardening wave 1 (PR #TBD) - **Busy-poll spin idle-disengages** (`--io-busy-poll-us` / vendored monoio diff --git a/src/vector/persistence/recover_v2.rs b/src/vector/persistence/recover_v2.rs index 75246b08..8895c888 100644 --- a/src/vector/persistence/recover_v2.rs +++ b/src/vector/persistence/recover_v2.rs @@ -413,19 +413,50 @@ fn load_segments_and_keymap( } } + // Segment-membership gate: the durable keymap covers EVERY indexed key + // (mutable + immutable) at snapshot-submit time, so after a crash it can + // be a strict SUPERSET of the durable segments — any key that was still + // mutable-resident when the last snapshot committed (or whose freshly + // installed segment's snapshot job never ran before the kill) has a + // keymap entry with a perfectly matching checksum but NO doc in any + // loaded segment. Loading such a phantom entry would make the B3 dedup + // rescan "verify" the key as unchanged and silently drop its document. + // Only keys live in a successfully loaded segment may enter the + // recovered maps; everything else stays unknown → full re-index from + // the AOF rescan. + let mut segment_resident: HashSet = HashSet::new(); + for seg in &immutable { + segment_resident.extend(seg.live_key_hashes()); + } + let entries = manifest::read_keymap_tolerant(idx_dir, manifest.keymap_epoch).unwrap_or_default(); let mut key_hash_to_key = BucketedKeyMap::new(); let mut key_hash_to_global_id = BucketedKeyMap::new(); let mut key_hash_to_vec_checksum = BucketedKeyMap::new(); + let mut phantom_entries = 0usize; for e in entries { if dropped_key_hashes.contains(&e.key_hash) { continue; } + if !segment_resident.contains(&e.key_hash) { + phantom_entries += 1; + continue; + } key_hash_to_key.insert(e.key_hash, e.key); key_hash_to_global_id.insert(e.key_hash, e.global_id); key_hash_to_vec_checksum.insert(e.key_hash, e.vec_checksum); } + if phantom_entries > 0 { + info!( + "vector index {}: {} keymap entr{} not backed by any loaded segment \ + (crash inside the async-snapshot window) — dropped; the rescan will \ + re-index those keys from the AOF", + String::from_utf8_lossy(name), + phantom_entries, + if phantom_entries == 1 { "y" } else { "ies" } + ); + } let loaded_segments = immutable.len(); let idx = vector_store.get_index_mut(name)?; @@ -799,6 +830,153 @@ mod tests { ); } + /// Deterministic regression test for the "phantom keymap entry" + /// silent-loss hole (found by `tests/crash_recovery_vector_durability.rs` + /// S1 flaking: post-crash `num_docs` 1950/2000 with `verified_unchanged + /// == 2000, re_indexed == 0`). + /// + /// The durable keymap covers EVERY indexed key (mutable + immutable) at + /// snapshot-submit time; a kill -9 landing before the NEXT snapshot + /// (covering a just-frozen segment) leaves on-disk keymap ⊃ on-disk + /// segments. This test stages that state directly: persist n docs + /// normally, then rewrite the durable keymap with one EXTRA entry whose + /// checksum matches its AOF blob but whose doc is in NO segment. The + /// dedup rescan must RE-INDEX that key (making it searchable), never + /// "verify" it unchanged (which drops the doc silently). + #[test] + fn recover_reindexes_keymap_entries_not_backed_by_any_segment() { + use crate::protocol::Frame; + + let tmp = tempfile::tempdir().unwrap(); + let dim = 8usize; + + // ---- Build + persist n real docs (same flow as the round-trip test) ---- + let mut store = VectorStore::new(); + store.set_persist_dir(tmp.path().to_path_buf()); + let meta = make_meta("idx", dim as u32, "doc:"); + store.create_index(meta.clone()).unwrap(); + + let n = 8usize; + for i in 0..n { + let key = format!("doc:{i}"); + let blob = f32_blob(dim, i as u32 + 1); + let args = vec![ + Frame::BulkString(Bytes::from(key.clone())), + Frame::BulkString(Bytes::from_static(b"vec")), + Frame::BulkString(blob), + ]; + let _ = crate::shard::spsc_handler::auto_index_hset_public( + &mut store, + &mut TextStore::new(), + key.as_bytes(), + &args, + ); + } + { + let idx = store.get_index_mut(b"idx").unwrap(); + idx.force_compact(); + } + let idx_dir = manifest::index_persist_dir(tmp.path(), b"idx"); + let loaded_manifest = wait_for_manifest(&idx_dir) + .expect("manifest must land within the timeout via global_snapshot_pool()"); + + // ---- Stage the crash window: durable keymap ⊃ durable segments ---- + // The phantom's checksum MATCHES the blob the rescan will replay + // (that is the crash-window signature: the key was mutable-resident + // and fully checksummed when this keymap snapshot committed, but its + // segment never reached disk). + let phantom_key = b"doc:phantom".to_vec(); + let phantom_hash = xxhash_rust::xxh64::xxh64(&phantom_key, 0); + let phantom_blob = f32_blob(dim, 4242); + let max_gid = { + let idx = store.get_index(b"idx").unwrap(); + idx.key_hash_to_global_id.values().copied().max().unwrap() + }; + let mut entries = manifest::read_keymap_tolerant(&idx_dir, loaded_manifest.keymap_epoch) + .expect("keymap must be readable"); + assert_eq!(entries.len(), n, "sanity: durable keymap covers all n docs"); + entries.push(manifest::KeymapEntry { + key_hash: phantom_hash, + global_id: max_gid + 1, + vec_checksum: xxhash_rust::xxh64::xxh64(&phantom_blob, 0), + key: Bytes::from(phantom_key.clone()), + }); + manifest::write_keymap_atomic(&idx_dir, loaded_manifest.keymap_epoch, &entries) + .expect("rewrite keymap with phantom entry"); + + // ---- Recover into a FRESH store ---- + let mut fresh = VectorStore::new(); + fresh.set_persist_dir(tmp.path().to_path_buf()); + let mut state = RecoveryState::new(); + state.create_index(&mut fresh, tmp.path(), &meta); + assert!(state.recovered_names.contains(&Bytes::from_static(b"idx"))); + + // The phantom must NOT have been loaded into the recovered maps — + // its doc exists in no loaded segment. + assert!( + fresh + .get_index(b"idx") + .unwrap() + .key_hash_to_key + .get(&phantom_hash) + .is_none(), + "keymap entry with no backing segment doc must be dropped at load" + ); + + // Rescan: replay all n real keys plus the phantom (its HSET is in + // the AOF — the key genuinely exists in the keyspace). + let mut text_store = TextStore::new(); + for i in 0..n { + let key = format!("doc:{i}"); + let blob = f32_blob(dim, i as u32 + 1); + let args = vec![ + Frame::BulkString(Bytes::from(key.clone())), + Frame::BulkString(Bytes::from_static(b"vec")), + Frame::BulkString(blob), + ]; + state.reconcile_key(&mut fresh, &mut text_store, key.as_bytes(), &args); + } + let phantom_args = vec![ + Frame::BulkString(Bytes::from(phantom_key.clone())), + Frame::BulkString(Bytes::from_static(b"vec")), + Frame::BulkString(phantom_blob.clone()), + ]; + state.reconcile_key(&mut fresh, &mut text_store, &phantom_key, &phantom_args); + + let counters = *state.counters.get(&Bytes::from_static(b"idx")).unwrap(); + assert_eq!( + counters.verified_unchanged, n, + "the n segment-backed keys dedup as unchanged" + ); + assert_eq!( + counters.re_indexed, 1, + "the phantom key must take the full re-index path (checksum match \ + without a backing segment doc must never count as unchanged)" + ); + + state.finish(&mut fresh, tmp.path()); + + // End-to-end: the phantom's document must actually be searchable — + // exact-match query returns it as top-1 (before the fix the doc was + // silently absent). + let phantom_gid = *fresh + .get_index(b"idx") + .unwrap() + .key_hash_to_global_id + .get(&phantom_hash) + .expect("phantom key must be indexed after the rescan"); + let query_f32: Vec = phantom_blob + .chunks_exact(4) + .map(|c| f32::from_le_bytes(c.try_into().unwrap())) + .collect(); + let results = fresh.search_index(b"idx", &query_f32, 1, 64).unwrap(); + assert_eq!( + results.first().copied(), + Some(phantom_gid), + "phantom key's doc must be searchable post-recovery (was silently lost)" + ); + } + /// A segment whose directory is entirely missing (simulating a crash /// that lost the whole segment, headers included) must not panic /// recovery — falls back to abandoning this index's recovered state diff --git a/src/vector/segment/immutable.rs b/src/vector/segment/immutable.rs index b1a0b738..794c12ef 100644 --- a/src/vector/segment/immutable.rs +++ b/src/vector/segment/immutable.rs @@ -810,6 +810,20 @@ impl ImmutableSegment { }) } + /// Iterate the `key_hash`es of all live entries — both install-time + /// (`delete_lsn`) and steady-state (`tombstoned_keys`) tombstones are + /// respected, mirroring [`Self::is_live_bfs`]. B3 recovery uses this to + /// gate the dedup rescan on actual segment membership: a durable keymap + /// entry whose doc is in NO loaded segment must be re-indexed from the + /// AOF, never "verified unchanged" (its keymap checksum still matching + /// is exactly the crash-window signature — the key was mutable-resident + /// when that keymap snapshot committed). + pub fn live_key_hashes(&self) -> impl Iterator + '_ { + (0..self.mvcc.len() as u32) + .filter(|&pos| self.is_live_bfs(pos)) + .map(|pos| self.mvcc[pos as usize].key_hash) + } + /// Map a BFS-reordered position to the globally unique key_hash. /// Used for building search results that are comparable across segments. #[inline] diff --git a/src/vector/store.rs b/src/vector/store.rs index cf9ccb85..3c5017e6 100644 --- a/src/vector/store.rs +++ b/src/vector/store.rs @@ -463,22 +463,17 @@ impl VectorIndex { // submit time, if configured) — commit the keymap/manifest // snapshot now that install is durable in memory. self.persist_hook_after_install(); - // After draining we're done — the data is compacted. - // Compact additional fields inline (no in-flight for those). - for (_, fs) in &mut self.field_segments { - let dim = fs.collection.dimension; - let mut unused_id = 0u64; - Self::compact_segments( - &mut fs.segments, - &mut fs.scratch, - &fs.collection, - dim, - 0, - None, - &mut unused_id, - ); - } - return; + // Do NOT return here: inserts that landed WHILE the + // background build was in flight are still in the mutable + // tail (`clone_suffix(frozen_len)` above). An early return + // breaks force_compact's full-drain contract (FT.COMPACT: + // frozen == mutable on reply) and leaves those docs with no + // durable segment until some future compact fires — a kill + // -9 in that window relied on them being "verified + // unchanged" from a keymap that covers them while no + // segment does (the B3 phantom-entry hole). Fall through to + // the inline compact below, which is a no-op when the tail + // is empty and otherwise drains + persists it. } // Worker failed or dropped — fall through to inline compact below. } @@ -3363,6 +3358,68 @@ mod bg_compact_tests { ); } + /// Regression (red/green verified): `force_compact` draining an IN-FLIGHT + /// background build must ALSO compact the docs inserted while that build + /// was running (the `clone_suffix(frozen_len)` mutable tail). The old + /// early return left the tail mutable-only — no durable segment — until + /// some future compact fired, breaking FT.COMPACT's full-drain contract + /// (frozen == mutable on reply) and, combined with the B3 phantom-keymap + /// hole, silently losing those docs on a kill -9 (see + /// `crash_recovery_vector_durability` S1/S6). Deterministic regardless of + /// worker speed: `bg_compact_inflight` stays `Some` until installed, and + /// force_compact's drain blocks on the reply channel. + #[test] + fn test_force_compact_drains_tail_inserted_during_inflight_bg_build() { + distance::init(); + let dim = 16u32; + let compactor = BackgroundCompactor::new(1); + let mut store = VectorStore::new(); + let mut meta = make_idx(dim); + meta.compact_threshold = 100; + store.create_index(meta).unwrap(); + + // 100 docs -> due-gated background build freezes all of them. + for i in 0..100u64 { + insert( + &mut store, + format!("doc:{i}").as_bytes(), + random_vec(dim as usize, i), + ); + } + store.begin_background_compactions(&compactor); + { + let idx = store.indexes.get_mut(b"idx".as_ref()).unwrap(); + assert!( + idx.bg_compact_inflight.is_some(), + "sanity: a background build must be in flight" + ); + } + + // Tail: 30 more docs land while the background build is in flight. + for i in 100..130u64 { + insert( + &mut store, + format!("doc:{i}").as_bytes(), + random_vec(dim as usize, i), + ); + } + + store.force_compact_index(b"idx").unwrap(); + + let idx = store.indexes.get_mut(b"idx".as_ref()).unwrap(); + let snap = idx.segments.load_full(); + assert_eq!( + snap.mutable.len(), + 0, + "force_compact must drain the tail inserted during the in-flight background build" + ); + let live: u32 = snap.immutable.iter().map(|s| s.live_count()).sum(); + assert_eq!( + live, 130, + "all docs must be segment-resident after force_compact" + ); + } + /// Like [`make_idx`] but with a caller-chosen index name (and a matching /// key prefix), so a test can create several independent indexes. fn make_idx_named(name: Bytes, dim: u32) -> IndexMeta { diff --git a/tests/crash_recovery_vector_durability.rs b/tests/crash_recovery_vector_durability.rs index f71e747b..326c147c 100644 --- a/tests/crash_recovery_vector_durability.rs +++ b/tests/crash_recovery_vector_durability.rs @@ -5,12 +5,14 @@ //! persist-on-compact (B1/B2), and startup recovery + dedup rescan + //! deletion probe + orphan sweep (B3, `src/vector/persistence/recover_v2.rs`). //! -//! Five scenarios, one `#[test]` each, sharing the harness below: +//! Six scenarios, one `#[test]` each, sharing the harness below: //! S1 unchanged-keys fast path (dedup rescan skips re-encoding) //! S2 updates+deletes across a crash (reconcile + deletion probe) //! S3 orphan sweep (stray staging/segment/keymap files removed on boot) //! S4 collection_id pin survives a post-recovery compact+GraphUnion merge //! S5 no-persist-dir regression guard (`--appendonly no`: no idx-* dirs) +//! S6 crash inside the async-snapshot window (durable keymap ⊃ durable +//! segments): uncovered keys re-index from the AOF, nothing silently lost //! //! Every wait is a bounded, condition-based poll (port accept / manifest //! file / log line / FT.INFO field) — never a fixed sleep as @@ -222,8 +224,20 @@ fn spawn_moon_aof(port: u16, dir: &Path) -> ServerGuard { ]) .arg(dir) .env("RUST_LOG", "moon=info") - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .stdout( + std::fs::File::options() + .append(true) + .create(true) + .open(dir.join("moon.stdout.log")) + .expect("stdout log"), + ) + .stderr( + std::fs::File::options() + .append(true) + .create(true) + .open(dir.join("moon.stderr.log")) + .expect("stderr log"), + ) .spawn() .expect("spawn moon (run `cargo build --release` first, or set MOON_BIN)"); ServerGuard::new(child, dir) @@ -249,8 +263,20 @@ fn spawn_moon_no_persist(port: u16, dir: &Path) -> ServerGuard { ]) .arg(dir) .env("RUST_LOG", "moon=info") - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .stdout( + std::fs::File::options() + .append(true) + .create(true) + .open(dir.join("moon.stdout.log")) + .expect("stdout log"), + ) + .stderr( + std::fs::File::options() + .append(true) + .create(true) + .open(dir.join("moon.stderr.log")) + .expect("stderr log"), + ) .spawn() .expect("spawn moon (run `cargo build --release` first, or set MOON_BIN)"); ServerGuard::new(child, dir) @@ -726,6 +752,55 @@ fn wait_for_manifest_exact_segments( } } +/// Bounded poll for FULL durability of an insert-only index: the manifest's +/// segments must hold exactly `n` live docs AND the paired keymap epoch must +/// hold exactly `n` entries. `wait_for_manifest_min_segments` is NOT enough +/// as a pre-SIGKILL gate when a compact emits more than one snapshot job +/// (e.g. an auto-compact install followed by the explicit FT.COMPACT's +/// residue freeze): the min-N manifest can be a stale intermediate whose +/// keymap covers keys its segments don't — killing there lands inside the +/// async-snapshot window and the dedup rescan must RE-INDEX the uncovered +/// keys (S6 asserts that contract), which breaks S1/S2's `re_indexed == 0` +/// fast-path determinism. Insert-only: live == `delete_lsn == 0`. +fn wait_for_durable_docs(idx_dir: &Path, n: usize, timeout: Duration) -> IndexManifest { + use moon::vector::persistence::segment_io; + let observe = |m: &IndexManifest| -> (usize, usize) { + let seg_live: usize = m + .segment_ids + .iter() + .map(|&id| { + segment_io::read_mvcc_headers_only(idx_dir, id) + .map(|hs| hs.iter().filter(|h| h.delete_lsn == 0).count()) + .unwrap_or(0) + }) + .sum(); + let keymap_n = manifest::read_keymap_tolerant(idx_dir, m.keymap_epoch) + .map(|e| e.len()) + .unwrap_or(0); + (seg_live, keymap_n) + }; + let deadline = Instant::now() + timeout; + loop { + if let Some(m) = manifest::read_manifest_tolerant(idx_dir) { + let (seg_live, keymap_n) = observe(&m); + if seg_live == n && keymap_n == n { + return m; + } + if Instant::now() >= deadline { + panic!( + "durable state at {idx_dir:?} did not reach {n} docs within {timeout:?} \ + (segments live={seg_live}, keymap entries={keymap_n}, \ + segment_ids={:?}, keymap_epoch={})", + m.segment_ids, m.keymap_epoch + ); + } + } else if Instant::now() >= deadline { + panic!("no manifest at {idx_dir:?} within {timeout:?}"); + } + std::thread::sleep(Duration::from_millis(20)); + } +} + /// Extract the 4 integer counters from a B3 recovery acceptance log line: /// "vector index {name}: B3 recovery — loaded {N} segment(s), {N} key(s) /// verified unchanged, {N} re-indexed, {N} tombstoned" (exact format string @@ -845,6 +920,11 @@ fn build_two_segment_snapshot(c: &mut Client, idx: &str, idx_dir: &Path, n: usiz hset_batch(c, &format!("{idx}:"), &batch2, simple_vec_bytes); ft_compact(c, idx); wait_for_manifest_min_segments(idx_dir, 2, Duration::from_secs(10)); + // Full-durability gate: min-2-segments alone can observe a stale + // intermediate snapshot (see `wait_for_durable_docs` doc) — S1/S2's + // `re_indexed == 0` determinism needs every doc segment-resident AND + // keymap-covered on disk before the SIGKILL. + wait_for_durable_docs(idx_dir, n, Duration::from_secs(10)); } #[test] @@ -1429,3 +1509,103 @@ fn s5_no_persist_dir_regression_guard() { drop(guard2); let _ = std::fs::remove_dir_all(&dir); } + +// --------------------------------------------------------------------------- +// S6: crash inside the async-snapshot window — AOF is the recovery authority +// --------------------------------------------------------------------------- + +/// Regression test for the B3 "phantom keymap entry" silent-loss hole. +/// +/// The durable `keymap-.bin` covers EVERY indexed key (mutable + +/// immutable) at snapshot-submit time, while the paired `manifest.json` only +/// lists installed immutable segments. A kill -9 landing after one snapshot +/// commits but before the NEXT one (covering a just-frozen segment) leaves +/// the on-disk keymap a strict SUPERSET of the on-disk segments. Recovery +/// must treat those uncovered keys as "not durable" and re-index them from +/// the AOF — a checksum match against the keymap alone must never count as +/// "verified unchanged" (observed failure before the fix: `num_docs` 1950/ +/// 2000 with 50 docs unsearchable, `verified_unchanged == N`, `re_indexed +/// == 0`). +/// +/// Unlike S1, this test deliberately does NOT wait for full durability +/// before the kill: it uses the same manifest >= 2 segment gate S1 +/// historically used (which lands inside the window whenever the final +/// compact produced more than one snapshot job) and then asserts only the +/// crash-safety contract: every key answers its own exact-match query and +/// `num_docs == N`, with the verified/re-indexed split left free. +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn s6_crash_during_snapshot_window_reindexes_uncovered_keys() { + const N: usize = 2000; + const IDX: &str = "s6"; + + let port = unique_port(); + let dir = unique_dir("s6"); + let vdir = vector_persist_dir(&dir); + let idx_dir = manifest::index_persist_dir(&vdir, IDX.as_bytes()); + + let guard = spawn_moon_aof(port, &dir); + let mut c = wait_ready(port); + + ft_create(&mut c, IDX, DIM, 100, None); + + let batch = N / 2; + let batch1: Vec = (0..batch as u32).collect(); + let batch2: Vec = (batch as u32..N as u32).collect(); + + hset_batch(&mut c, &format!("{IDX}:"), &batch1, simple_vec_bytes); + ft_compact(&mut c, IDX); + wait_for_manifest_min_segments(&idx_dir, 1, Duration::from_secs(10)); + + hset_batch(&mut c, &format!("{IDX}:"), &batch2, simple_vec_bytes); + ft_compact(&mut c, IDX); + // Kill as soon as ANY manifest with >= 2 segments is durable — if the + // final compact emitted more than one snapshot job, this is inside the + // window where the durable keymap covers keys the durable segments + // don't. + wait_for_manifest_min_segments(&idx_dir, 2, Duration::from_secs(10)); + + let pre_num_docs = ft_info_num_docs(&mut c, IDX); + assert_eq!(pre_num_docs, N as i64, "pre-crash num_docs must equal N"); + + drop(c); + let mut guard = guard; + sigkill(&mut guard.child); + wait_for_port_down(port); + + // -- Restart -------------------------------------------------------- + let guard2 = start_moon_alive(spawn_moon_aof, port, &dir); + let mut c2 = wait_ready(port); + + // The B3 line must exist (a manifest was durable), but the + // verified/re-indexed split is timing-dependent — only the total is + // contractual: every key was observed, none tombstoned. + let (_loaded, verified_unchanged, re_indexed, tombstoned) = + wait_for_recovery_counters(&dir, IDX, Duration::from_secs(10)); + assert_eq!( + verified_unchanged + re_indexed, + N, + "every key must be either verified unchanged or re-indexed" + ); + assert_eq!(tombstoned, 0, "no key was deleted before the crash"); + + // The crash-safety contract: nothing is silently lost, regardless of + // which snapshot prefix made it to disk. + let post_num_docs = ft_info_num_docs(&mut c2, IDX); + assert_eq!(post_num_docs, N as i64, "post-crash num_docs must equal N"); + + for i in (0..N as u32).step_by(25) { + let blob = simple_vec_bytes(i); + let results = search_keys(&mut c2, IDX, 1, &blob); + let want_key = format!("{IDX}:{i}"); + assert_eq!( + results.first().map(String::as_str), + Some(want_key.as_str()), + "exact-match query for {want_key} must return itself as top-1 after crash recovery" + ); + } + + drop(c2); + drop(guard2); + let _ = std::fs::remove_dir_all(&dir); +} From c27d80e795cb33d4ba2f2f990124f5cc583df286 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 22:07:48 +0700 Subject: [PATCH 6/8] test(ci): deterministic OOM case E escalation + Windows compile gate for quickwins_red_api MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Main's post-merge CI runs for PR #217/#218/#219 are all red on two issues this branch's PR (#228) also inherits; both are test-side. 1. test_case_e_cross_db_copy_oom: 0/300 OOM on BOTH CI platforms while passing locally (76/300; earlier deflake N/10 -> N/30 saw 28/300). The single-shot statistical assert races GAP-1's elastic budget: one round of 300x4KB COPYs (~300KB/shard) fits inside the slack the elastic budget plus its 100ms-stale usage snapshots can grant, so the OOM share is runner-speed-dependent all the way down to zero. Rewritten as bounded escalation: up to 8 rounds of COPYs into fresh destination keys, stopping early once OOMs appear. The destination db's per-shard usage (~2.4MB after all rounds) provably exceeds the ABSOLUTE per-shard budget ceiling (compute_elastic_budget returns at most base + surplus <= maxmemory = 2MB, and the gate compares db.estimated_memory() against it under noeviction), so gate-checked COPYs MUST OOM on any runner speed. The RED floor stays exact and was re-verified: with the Gap A gate block neutered, 0 OOM through all 8 rounds (2400 COPYs) — the assert still discriminates. 2. tests/quickwins_red_api.rs broke the Windows CI leg at COMPILE time: qw1_accepted_socket_has_nodelay calls moon::server::socket_opts::apply_client_socket_opts, which is #[cfg(unix)] (takes AsFd). The test and its TcpListener/TcpStream imports are now unix-gated to match the helper. Validation: oom_bypass_closure 8/8 green on monoio AND tokio+jemalloc (MOON_NO_URING=1) locally; case E red/green toggled via the gate block. author: Tin Dang --- CHANGELOG.md | 13 ++++++ tests/oom_bypass_closure.rs | 85 ++++++++++++++++++++++++------------- tests/quickwins_red_api.rs | 5 +++ 3 files changed, 73 insertions(+), 30 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 180d4ab4..050717d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 dropping the stale job's `Arc`s immediately; an in-flight (already-dequeued) job is unaffected. Submit never blocks the caller. +### Fixed — CI: de-flake OOM case E + Windows test-compile gate (PR #TBD) + +- **`test_case_e_cross_db_copy_oom` (was red on main's post-merge CI for PR #217/#218/#219, + 0/300 OOM on both platforms while passing locally):** the single-shot statistical assert sat + inside the slack GAP-1's elastic budget + its 100ms-stale usage snapshots can grant. Rewritten + as bounded escalation: up to 8 rounds of COPYs into fresh destination keys until the + destination db's per-shard usage (~2.4MB) provably exceeds the ABSOLUTE budget ceiling + (`compute_elastic_budget ≤ maxmemory` = 2MB) — timing-independent on any runner speed, and the + RED floor stays exact (gate stashed = 0 OOM through all rounds; re-verified red/green). +- **`tests/quickwins_red_api.rs` broke the Windows CI leg at COMPILE time** (also red on main): + `qw1_accepted_socket_has_nodelay` calls `apply_client_socket_opts`, which is `#[cfg(unix)]` + (takes `AsFd`). The test (and its imports) are now unix-gated to match. + ### Fixed — FT.COMPACT left mutable residue behind when draining a background build (PR #TBD) - **Pre-existing** (`main`, not introduced by this branch): when an explicit diff --git a/tests/oom_bypass_closure.rs b/tests/oom_bypass_closure.rs index d9789a79..c658f77f 100644 --- a/tests/oom_bypass_closure.rs +++ b/tests/oom_bypass_closure.rs @@ -560,40 +560,65 @@ fn test_case_e_cross_db_copy_oom() { "SELECT 0 (back for phase 2) failed" ); - // Phase 2: COPY each key to a DISTINCT dst key (src != dst — required, - // Redis rejects same-key COPY regardless of db) — doubles the value's - // footprint, pushing per-shard usage past the cap. - let copy_cmds: Vec>> = (0..N) - .map(|i| { - let src = format!("e:{i}").into_bytes(); - let dst = format!("e:{i}:c").into_bytes(); - vec![b"COPY".to_vec(), src, dst, b"DB".to_vec(), b"1".to_vec()] - }) - .collect(); - let copy_replies = c.pipeline(©_cmds); - assert_eq!(copy_replies.len(), N); - - let oom_count = copy_replies.iter().filter(|r| r.is_oom_error()).count(); - let ok_count = copy_replies - .iter() - .filter(|r| matches!(r, V::Int(1))) - .count(); - let other_count = N - oom_count - ok_count; + // Phase 2: escalating rounds of COPYs, each round to FRESH dst keys + // (src != dst — required, Redis rejects same-key COPY regardless of db), + // until the destination db's per-shard footprint provably exceeds ANY + // possible budget. + // + // Why rounds instead of the original single-shot statistical assert: + // one round of 300×4KB (~300KB/shard) sits inside the slack that GAP-1's + // elastic budget + its 100ms-stale usage snapshots can grant, so the + // single-shot OOM share was timing-sensitive — 76/300 locally, 28/300 on + // slow GitHub runners after one deflake (N/10 → N/30), and finally + // 0/300 EXACTLY on both CI platforms (post-merge main runs of PR + // #217/#218/#219 all red). Escalation removes the timing dependence via + // an absolute ceiling: `compute_elastic_budget` can never grant a shard + // more than `base + surplus ≤ maxmemory` (2MB here), and the gate is + // per-DATABASE (`db.estimated_memory()` vs that budget) under + // `noeviction` — so once db 1's per-shard usage passes 2MB, every + // further gate-checked COPY into it MUST OOM, on any runner speed. 8 + // rounds × 300 × 4KB ≈ 9.6MB into db 1 (~2.4MB/shard, margin over the + // 2MB ceiling even for the luckiest shard of the hash spread). + // + // The RED floor is unchanged and still exact: with the Gap A gate block + // stashed, COPY never consults the eviction gate at ANY pressure, so + // the cumulative OOM count stays 0 through all rounds (re-verified + // red/green methodology in the Gap A commit body). + const ROUNDS: usize = 8; + let mut oom_count = 0usize; + let mut ok_count = 0usize; + let mut other_count = 0usize; + for round in 0..ROUNDS { + let copy_cmds: Vec>> = (0..N) + .map(|i| { + let src = format!("e:{i}").into_bytes(); + let dst = format!("e:{i}:c{round}").into_bytes(); + vec![b"COPY".to_vec(), src, dst, b"DB".to_vec(), b"1".to_vec()] + }) + .collect(); + let copy_replies = c.pipeline(©_cmds); + assert_eq!(copy_replies.len(), N); + oom_count += copy_replies.iter().filter(|r| r.is_oom_error()).count(); + ok_count += copy_replies + .iter() + .filter(|r| matches!(r, V::Int(1))) + .count(); + other_count = (round + 1) * N - oom_count - ok_count; + if oom_count >= N / 30 { + break; // gate demonstrably firing — no need to keep escalating + } + } - // Threshold picked from observed data, same methodology as case B: - // gate-block-disabled (routing intact, see the Gap A commit body) run = - // 0/300 OOM EXACTLY; fully fixed = 76/300 locally but as low as 28/300 - // on slow CI runners (GAP-1's elastic budget redistributes on a 100ms - // tick, so the OOM share is timing-sensitive — N/10 flaked at 28<30 on - // GitHub Actions). Since the RED floor is exactly 0, any clearly-nonzero - // threshold discriminates; N/30 (10) keeps margin both ways. assert!( oom_count >= N / 30, - "cross-shard COPY bypass: expected a substantial share of {N} COPYs \ - routed via cross-shard SPSC to OOM once their shard's budget is \ - exhausted, got only {oom_count} OOM / {ok_count} OK / \ + "cross-shard COPY bypass: expected COPYs routed via cross-shard SPSC \ + to OOM once the destination db's usage exceeds every possible \ + budget (≈{}KB/shard vs the {}KB absolute budget ceiling after \ + {ROUNDS} rounds), got only {oom_count} OOM / {ok_count} OK / \ {other_count} other — COPY is not hitting the generic eviction \ - gate on the cross-shard write path" + gate on the cross-shard write path", + (N / 4 * VALUE_SIZE * (ROUNDS + 1)) / 1024, + MAXMEMORY / 1024, ); } diff --git a/tests/quickwins_red_api.rs b/tests/quickwins_red_api.rs index f3007c55..d8cb1dae 100644 --- a/tests/quickwins_red_api.rs +++ b/tests/quickwins_red_api.rs @@ -7,6 +7,7 @@ //! cargo test --test quickwins_red # runs (1 red, 4 pins) //! cargo test --test quickwins_red_api # compile error = red, by design +#[cfg(unix)] use std::net::{TcpListener, TcpStream}; // --------------------------------------------------------------------------- @@ -15,6 +16,10 @@ use std::net::{TcpListener, TcpStream}; /// The accept paths (tokio + uring register) funnel socket options through one /// helper; applying it to an accepted fd must enable TCP_NODELAY. +/// Unix-only like the helper itself (`apply_client_socket_opts` is +/// `#[cfg(unix)]` — it takes `AsFd`); without this gate the whole test +/// binary fails to COMPILE on the Windows CI leg (main has been red there). +#[cfg(unix)] #[test] fn qw1_accepted_socket_has_nodelay() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); From deb9d530b175c3cf2debcb729f0acf9badeb5066 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 22:24:03 +0700 Subject: [PATCH 7/8] test(ci): merge racing RECL_SEGMENT_STALL_ACTIVE tests into one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit test_recl_segment_stall_active_atomic_exists and test_is_segment_stall_active_helper both store/load the PROCESS-GLOBAL RECL_SEGMENT_STALL_ACTIVE atomic while the test harness runs them on parallel threads within the same binary — the atomic-exists test's `store(1); load()` observed the helper test's final `store(0)` interleaving (CI macOS: `left: 0, right: 1` at the read-back assert; PR #228 round-2 Check (macOS) leg). Pre-existing since MA1 (wave 1); purely a test-parallelism race, not a stall-gate defect. All mutation of the global now lives in ONE merged test (test_recl_segment_stall_active_atomic_and_helper) so there is nothing left to race with; the default-0 assert is also only valid under that single-writer arrangement. Ran 5x locally green. author: Tin Dang --- tests/write_stall_segment_backlog.rs | 42 ++++++++++++++-------------- 1 file changed, 21 insertions(+), 21 deletions(-) diff --git a/tests/write_stall_segment_backlog.rs b/tests/write_stall_segment_backlog.rs index eaee505e..e31406c1 100644 --- a/tests/write_stall_segment_backlog.rs +++ b/tests/write_stall_segment_backlog.rs @@ -23,40 +23,40 @@ fn test_vector_store_total_immutable_count_method_exists() { ); } -/// Unit-level test: RECL_SEGMENT_STALL_ACTIVE is updated by the segment-stall -/// check and OR-merged into RECL_WRITE_STALL_ACTIVE semantics. +/// Unit-level test: RECL_SEGMENT_STALL_ACTIVE exists, defaults to 0, is +/// writable, and drives `is_segment_stall_active()`. /// -/// RED: fails because RECL_SEGMENT_STALL_ACTIVE does not exist. +/// ONE test (not two) on purpose: the atomic is a PROCESS-GLOBAL and the +/// test harness runs `#[test]`s in this binary on parallel threads — two +/// tests store/load-ing the same global raced (observed on CI macOS: +/// `store(1)` here read back `0` because the sibling helper test's final +/// `store(0)` interleaved). All mutation of the global lives in this single +/// test so there is nothing to race with. +/// +/// RED: fails because RECL_SEGMENT_STALL_ACTIVE / is_segment_stall_active +/// do not exist. #[test] -fn test_recl_segment_stall_active_atomic_exists() { +fn test_recl_segment_stall_active_atomic_and_helper() { use moon::command::info_reclamation::RECL_SEGMENT_STALL_ACTIVE; - // Must be zero by default. + use moon::shard::segment_stall::is_segment_stall_active; + + // Must be zero by default (no other test in this binary touches it). assert_eq!( RECL_SEGMENT_STALL_ACTIVE.load(Ordering::Relaxed), 0, "RECL_SEGMENT_STALL_ACTIVE must default to 0" ); - // Must be writable. - RECL_SEGMENT_STALL_ACTIVE.store(1, Ordering::Relaxed); - assert_eq!(RECL_SEGMENT_STALL_ACTIVE.load(Ordering::Relaxed), 1); - RECL_SEGMENT_STALL_ACTIVE.store(0, Ordering::Relaxed); -} - -/// Unit-level test: `is_segment_stall_active()` returns true when -/// RECL_SEGMENT_STALL_ACTIVE is non-zero, false otherwise. -/// -/// RED: fails because `is_segment_stall_active` does not exist. -#[test] -fn test_is_segment_stall_active_helper() { - use moon::command::info_reclamation::RECL_SEGMENT_STALL_ACTIVE; - use moon::shard::segment_stall::is_segment_stall_active; - - RECL_SEGMENT_STALL_ACTIVE.store(0, Ordering::Relaxed); assert!(!is_segment_stall_active(), "must be false when atomic is 0"); + // Must be writable, and the helper must observe the change. RECL_SEGMENT_STALL_ACTIVE.store(1, Ordering::Relaxed); + assert_eq!(RECL_SEGMENT_STALL_ACTIVE.load(Ordering::Relaxed), 1); assert!(is_segment_stall_active(), "must be true when atomic is 1"); // Restore RECL_SEGMENT_STALL_ACTIVE.store(0, Ordering::Relaxed); + assert!( + !is_segment_stall_active(), + "must be false again after restore" + ); } From fd3d63043711362dcd7580ab837cfcac62b43ea0 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 22:24:37 +0700 Subject: [PATCH 8/8] docs(changelog): note the stall-atomic test-race merge in the CI-fix entry author: Tin Dang --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 050717d6..720e2dbc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -77,6 +77,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **`tests/quickwins_red_api.rs` broke the Windows CI leg at COMPILE time** (also red on main): `qw1_accepted_socket_has_nodelay` calls `apply_client_socket_opts`, which is `#[cfg(unix)]` (takes `AsFd`). The test (and its imports) are now unix-gated to match. +- **`write_stall_segment_backlog`: two tests raced on the process-global + `RECL_SEGMENT_STALL_ACTIVE` atomic** (test harness runs same-binary tests on parallel + threads; observed on CI macOS as `store(1)` reading back `0`). All mutation of the global now + lives in one merged test. ### Fixed — FT.COMPACT left mutable residue behind when draining a background build (PR #TBD)