perf(vector): bucket-scoped CoW keymaps + coalescing snapshot queue; fix pre-existing kill-9 doc loss

CHANGELOG.md

@@ -46,6 +46,78 @@ 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<HashMap<u64, V>>`. 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<V>` (`src/vector/keymap.rs`): 256 fixed buckets,
+  each independently `Arc<HashMap<u64, V>>`, 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 — 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.
+- **`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)
+
+- **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-<epoch>.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

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<std::collections::HashMap<u64, Bytes>>,
+        key_hash_to_key: BucketedKeyMap<Bytes>,
     },
     Error(Frame),
 }

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<u64, Bytes>,
+    key_hash_to_key: &BucketedKeyMap<Bytes>,
     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<u64, Bytes>,
+    key_hash_to_key: &BucketedKeyMap<Bytes>,
     dense_count: usize,
     sparse_count: usize,
     offset: usize,

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<SearchResult>,
-        std::sync::Arc<std::collections::HashMap<u64, Bytes>>,
-    ),
-    Frame,
-> {
+) -> Result<(Vec<SearchResult>, BucketedKeyMap<Bytes>), 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<u64, Bytes>,
+    vector_key_hash_to_key: &BucketedKeyMap<Bytes>,
     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<u64, Bytes>,
+    vector_map: &BucketedKeyMap<Bytes>,
     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:<id> fallback.
-        let key_map = std::collections::HashMap::new();
+        let key_map: BucketedKeyMap<Bytes> = 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}")));
         }

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<u64, Bytes>,
+    vector_key_hash_to_key: &BucketedKeyMap<Bytes>,
 ) -> Frame {
     let total_items = 3 + 4 * (bm25.len() + dense.len() + sparse.len());
     let mut out: Vec<Frame> = 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<u64, Bytes> = std::collections::HashMap::new();
+        let mut vec_map: BucketedKeyMap<Bytes> = 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<u64, Bytes> = std::collections::HashMap::new();
+        let vec_map: BucketedKeyMap<Bytes> = 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());

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<Bytes, f64>,
-    key_hash_to_key: &HashMap<u64, Bytes>,
+    key_hash_to_key: &BucketedKeyMap<Bytes>,
 ) -> 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<u64, Bytes>,
+    key_hash_to_key: &BucketedKeyMap<Bytes>,
     timestamp: f64,
 ) {
     if results.is_empty() {

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<Bytes> = 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<Bytes> = 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<Bytes, f64> = HashMap::new();
-    let mut key_hash_to_key = HashMap::new();
+    let mut key_hash_to_key: BucketedKeyMap<Bytes> = 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<Bytes, f64> = 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<Bytes> = 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<Bytes> = 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());
         }