diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1312096f..720e2dbc 100644
--- a/CHANGELOG.md
+++ b/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
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<std::collections::HashMap<u64, Bytes>>,
+        key_hash_to_key: BucketedKeyMap<Bytes>,
     },
     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<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,
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<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}")));
         }
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<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());
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<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() {
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<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());
         }
 
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/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<HashMap>` 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<HashMap<u64, V>>`. 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<HashMap<u64, V>>`
+//! 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<V> {
+    buckets: Box<[Arc<HashMap<u64, V>>]>,
+}
+
+impl<V> BucketedKeyMap<V> {
+    /// 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<Arc<HashMap<u64, V>>> =
+            (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<V>
+    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<V>
+    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<Item = (&u64, &V)> {
+        self.buckets.iter().flat_map(|b| b.iter())
+    }
+
+    /// Iterate all key hashes across every bucket.
+    pub fn keys(&self) -> impl Iterator<Item = &u64> {
+        self.iter().map(|(k, _)| k)
+    }
+
+    /// Iterate all values across every bucket.
+    pub fn values(&self) -> impl Iterator<Item = &V> {
+        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<V> Default for BucketedKeyMap<V> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+// Manual `Clone` impl (rather than `#[derive(Clone)]`) so cloning a
+// `BucketedKeyMap<V>` never requires `V: Clone` — only `Arc<HashMap<u64,
+// V>>: 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<V> Clone for BucketedKeyMap<V> {
+    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<u32> = 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<u32> = 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<u32> = 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<u64, u64> = StdHashMap::new();
+        let mut bucketed: BucketedKeyMap<u64> = 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<u32> = 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<u32> = 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<NotClone> = BucketedKeyMap::new();
+        let cloned = m.clone(); // must compile without `V: Clone`
+        assert_eq!(cloned.len(), 0);
+    }
+
+    #[test]
+    fn default_is_empty() {
+        let m: BucketedKeyMap<u32> = 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;
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<std::collections::HashMap<u64, Bytes>>,
-    pub key_hash_to_global_id: Arc<std::collections::HashMap<u64, u32>>,
-    pub key_hash_to_vec_checksum: Arc<std::collections::HashMap<u64, u64>>,
+    pub key_hash_to_key: BucketedKeyMap<Bytes>,
+    pub key_hash_to_global_id: BucketedKeyMap<u32>,
+    pub key_hash_to_vec_checksum: BucketedKeyMap<u64>,
 }
 
 /// 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<HashMap<idx_dir, SnapshotJob>>`
+/// 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<SnapshotJob>,
+    /// 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<Mutex<HashMap<PathBuf, SnapshotJob>>>,
+    /// Signalled on every `submit` so an idle worker wakes immediately.
+    not_empty: Arc<Condvar>,
+    /// 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<std::sync::atomic::AtomicBool>,
     _workers: Vec<std::thread::JoinHandle<()>>,
 }
 
 impl SnapshotPool {
     fn new(num_workers: usize) -> Self {
-        let (job_tx, job_rx) = flume::unbounded::<SnapshotJob>();
+        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<F>(num_workers: usize, runner: F) -> Self
+    where
+        F: Fn(SnapshotJob) + Send + Sync + 'static,
+    {
+        let pending: Arc<Mutex<HashMap<PathBuf, SnapshotJob>>> =
+            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(&not_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<std::collections::HashMap<u64, u64>> {
-        Arc::new(
-            entries
-                .iter()
-                .map(|e| (e.key_hash, e.vec_checksum))
-                .collect(),
-        )
+    fn checksum_map(entries: &[KeymapEntry]) -> BucketedKeyMap<u64> {
+        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<std::collections::HashMap<u64, u32>> {
-        Arc::new(entries.iter().map(|e| (e.key_hash, e.global_id)).collect())
+    fn global_id_map(entries: &[KeymapEntry]) -> BucketedKeyMap<u32> {
+        let mut m = BucketedKeyMap::new();
+        for e in entries {
+            m.insert(e.key_hash, e.global_id);
+        }
+        m
     }
-    fn key_map(entries: &[KeymapEntry]) -> Arc<std::collections::HashMap<u64, Bytes>> {
-        Arc::new(
-            entries
-                .iter()
-                .map(|e| (e.key_hash, e.key.clone()))
-                .collect(),
-        )
+    fn key_map(entries: &[KeymapEntry]) -> BucketedKeyMap<Bytes> {
+        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<Mutex<u64>>) -> 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<Mutex<Vec<u64>>> = 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<Mutex<Vec<u64>>> = 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));
+        }
+    }
 }
diff --git a/src/vector/persistence/recover_v2.rs b/src/vector/persistence/recover_v2.rs
index a855804a..8895c888 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;
@@ -412,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<u64> = 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 = 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();
+    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)?;
@@ -441,9 +473,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,
@@ -798,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<f32> = 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/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<std::collections::HashMap<u64, bytes::Bytes>>,
+    /// 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<bytes::Bytes>,
     /// 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/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<Item = u64> + '_ {
+        (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 e671ea56..3c5017e6 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:<internal_id>` 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<std::collections::HashMap<u64, Bytes>>,
+    /// Bucketed CoW (RSS/CPU wave 4, defect 1): 256 independent
+    /// `Arc<HashMap>` 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<Bytes>,
     /// 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<std::collections::HashMap<u64, u32>>,
+    /// 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<u32>,
     /// 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<std::collections::HashMap<u64, u64>>,
+    pub key_hash_to_vec_checksum: BucketedKeyMap<u64>,
     /// Shard-relative directory this index persists segments/manifest/keymap
     /// under, i.e. `<vector_persist_dir>/idx-<hex(xxh64(name))>/` (B2).
     /// `None` when the store has no `persist_dir` configured (disk-offload
@@ -460,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.
         }
@@ -1572,9 +1570,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 +1698,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 +1914,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 +2290,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(())
     }
 
@@ -3360,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-<epoch>.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<u32> = (0..batch as u32).collect();
+    let batch2: Vec<u32> = (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);
+}
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<Vec<Vec<u8>>> = (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(&copy_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<Vec<Vec<u8>>> = (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(&copy_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();
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"
+    );
 }