jthomson04 · Oct 13, 2026 · Oct 12, 2026 · Oct 12, 2026 · Oct 13, 2026
@@ -17,8 +17,9 @@ before changing locking, versioning, split, remove, or lookup-repair behavior.
 - Do not replace lazy lookup repair with eager/global repair without
   benchmarking. Direction-aware and batched repair are intentional.
 - `find_matches` may undercount during races, but must never overcount past a
-  valid reachable prefix, except through the documented equal-size skip (see
-  `README.md`). Do not add new overcount paths.
+  valid reachable prefix, except through a fingerprint collision in the
+  equal-set skip (about 2^-64 per comparison; see `README.md`). Do not add new
+  overcount paths.
 - Any hot-path change to locking, versioning, lookup repair, split/remove, child
   insertion, or read traversal must include before/after benchmark numbers in
   the PR.

@@ -238,23 +238,24 @@ exclusive shape gate.
 `find_matches` is best-effort during concurrent shape changes. It reads node
 state and child pointers without taking `shape_gate` on the hot step, so it may
 observe adjacent tree shapes during a split and undercount. It must not panic,
-and apart from the equal-size skip below it must not return a match past a
-valid reachable prefix.
+and apart from a fingerprint collision in the equal-set skip below it must not
+return a match past a valid reachable prefix.
 
-### Equal-size skip
+### Equal-set skip
 
 After the first node, the walk intersects its active workers with each node's
-full-edge coverage. When the two sets have the same size, it skips that
-intersection and treats them as equal. This is an accepted approximation.
-Removal does not cascade to children, so after a worker's head blocks are
-evicted a child can still list it, and an equal-sized coverage set can then hold
-a different worker than the walk carries. The walk credits the carried worker
-with the child's depth, past its cached prefix.
-
-The skip saves a pass over up to all workers at most hops. On the Mooncake
-replay with 128 workers it fired about 4.3M times and was wrong in 0 of them at a
-keep-up load and 2 of about 741K when overloaded, while always intersecting
-raised lookup service p50 by 34%.
+full-edge coverage. When the two sets are equal it skips that intersection,
+which saves a pass over up to all workers at most hops: on the Mooncake replay
+with 128 workers, always intersecting raised lookup service p50 by 34%.
+
+Size alone does not prove the sets equal. Removal does not cascade to
+children, so after a worker's head blocks are evicted a child can still list
+it, and an equal-sized coverage set can then hold a different worker than the
+walk carries. Both sets are therefore `WorkerSet`s, which keep the XOR of a
+64-bit mix of each member, and the walk skips only when the sizes and these
+fingerprints match. The check is probabilistic: unequal sets collide with
+probability about 2^-64 per comparison, and a collision overcounts exactly as
+the size-only check did. That is the one remaining overcount path.
 
 ## Wire Compatibility
 

@@ -85,8 +85,7 @@ impl ConcurrentRadixTreeCompressed {
         >,
         mut kv_transfer_chain: Option<&mut Vec<ExternalSequenceBlockHash>>,
     ) -> MatchWalkResult {
-        let mut active: FxHashSet<WorkerWithDpRank> = FxHashSet::default();
-        let mut active_count: usize = 0;
+        let mut active = WorkerSet::default();
         let mut matched_depth: u32 = 0;
         let mut seq_pos: usize = 0;
         let mut first_node = true;
@@ -111,14 +110,13 @@ impl ConcurrentRadixTreeCompressed {
                 prev_depth: matched_depth,
                 prev_edge_last_hash,
                 active: &mut active,
-                active_count,
                 scores,
                 last_matched_hashes: last_matched_hashes.as_deref_mut(),
                 kv_transfer_chain: kv_transfer_chain.as_deref_mut(),
             });
             let edge_len = outcome.edge_len;
             let edge_match_len = outcome.edge_match_len;
-            active_count = outcome.active_count;
+            let active_count = outcome.active_count;
             next_child = outcome.next_child;
             prev_edge_last_hash = outcome.prev_edge_last_hash;
             if first_node {

@@ -28,9 +28,11 @@ mod lane_lookup;
 mod node;
 mod state;
 mod types;
+mod worker_set;
 use lane_lookup::LaneLookup;
 use node::*;
 use types::*;
+use worker_set::WorkerSet;
 
 mod dump;
 mod matches;

@@ -6,11 +6,12 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::{Arc, Weak};
 
 use parking_lot::{RwLock, RwLockUpgradableReadGuard};
-use rustc_hash::{FxHashMap, FxHashSet};
+use rustc_hash::FxHashMap;
 
 use super::children::{ChildInsertResult, NodeChildren};
 use super::state::CrtcNodeState;
 use super::types::*;
+use super::worker_set::WorkerSet;
 use crate::protocols::*;
 
 fn record_last_matched_hash(
@@ -269,7 +270,7 @@ impl Node {
         }
     }
 
-    fn has_full_coverage_only_matching(&self, workers: &FxHashSet<WorkerWithDpRank>) -> bool {
+    fn has_full_coverage_only_matching(&self, workers: &WorkerSet) -> bool {
         let state = self.state.read();
         state.worker_cutoffs.is_empty()
             && state.full_edge_workers == *workers
@@ -696,9 +697,9 @@ impl Node {
                         false
                     }
                 });
-            } else if state.full_edge_workers.len() != input.active_count {
-                // Equal sizes are treated as equal sets, skipping the intersection. This
-                // accepts an overcount after head-first eviction; see README "Equal-size skip".
+            } else if !input.active.same_members(&state.full_edge_workers) {
+                // Matching size and fingerprint skip the intersection; see README
+                // "Equal-set skip".
                 input.active.retain(|worker| {
                     if state.full_edge_workers.contains(worker) {
                         true
@@ -710,6 +711,8 @@ impl Node {
                         false
                     }
                 });
+            } else {
+                debug_assert_eq!(**input.active, *state.full_edge_workers);
             }
         }
 

@@ -3,9 +3,10 @@
 
 //! Edge and coverage state of one node, kept behind the node's state lock.
 
-use rustc_hash::{FxBuildHasher, FxHashMap, FxHashSet};
+use rustc_hash::{FxBuildHasher, FxHashMap};
 
 use super::edge_index::EdgeIndex;
+use super::worker_set::WorkerSet;
 use crate::protocols::*;
 
 pub(super) struct RemoveOutcome {
@@ -25,7 +26,7 @@ pub(super) struct CrtcNodeState {
     /// has cached `edge[0..k]`, where `0 < k < edge.len()`.
     pub(super) worker_cutoffs: FxHashMap<WorkerWithDpRank, usize>,
     /// Workers with full edge coverage.
-    pub(super) full_edge_workers: FxHashSet<WorkerWithDpRank>,
+    pub(super) full_edge_workers: WorkerSet,
 }
 
 impl CrtcNodeState {
@@ -35,7 +36,7 @@ impl CrtcNodeState {
             edge_index: EdgeIndex::build(&edge),
             edge,
             worker_cutoffs: FxHashMap::default(),
-            full_edge_workers: FxHashSet::default(),
+            full_edge_workers: WorkerSet::default(),
         }
     }
 
@@ -307,6 +308,8 @@ impl CrtcNodeState {
 mod tests {
     use std::collections::HashMap;
 
+    use rustc_hash::FxHashSet;
+
     use super::super::edge_index::SCAN_MAX_LEN;
     use super::*;
 
@@ -445,15 +448,15 @@ mod tests {
 
         assert_eq!(state.edge.len(), 3);
         assert_eq!(
-            state.full_edge_workers,
+            *state.full_edge_workers,
             FxHashSet::from_iter([full, at_split, past_split])
         );
         assert_eq!(state.worker_cutoffs, FxHashMap::from_iter([(short, 1)]));
         assert_eq!(state.position(ExternalSequenceBlockHash(3)), Some(2));
         assert_eq!(state.position(ExternalSequenceBlockHash(4)), None);
 
         assert_eq!(suffix.edge.len(), 3);
-        assert_eq!(suffix.full_edge_workers, FxHashSet::from_iter([full]));
+        assert_eq!(*suffix.full_edge_workers, FxHashSet::from_iter([full]));
         assert_eq!(
             suffix.worker_cutoffs,
             FxHashMap::from_iter([(past_split, 2)])

@@ -4,8 +4,9 @@
 use super::*;
 use crate::indexer::{KvIndexerInterface, ThreadPoolIndexer};
 use crate::test_utils::{
-    assert_score, flush_and_settle, make_clear_event_with_dp_rank, make_remove_event_with_parent,
-    make_store_event, make_store_event_with_dp_rank, make_store_event_with_parent, remove_event,
+    assert_score, flush_and_settle, make_clear_event_with_dp_rank, make_remove_event,
+    make_remove_event_with_parent, make_store_event, make_store_event_full,
+    make_store_event_with_dp_rank, make_store_event_with_parent, remove_event, router_event,
     snapshot_events, snapshot_tree,
 };
 use std::sync::{Arc, Barrier};
@@ -1047,9 +1048,8 @@ mod remove_tests {
     /// score must never exceed its held prefix of the query. Worker 0 removes
     /// one block or a held run in either order, leaving mid-chain holes that
     /// later stores partially refill. The other workers share and split its
-    /// edges but evict only leaves, so the documented equal-size skip in
-    /// `find_matches` cannot credit worker 0 past its prefix. Workers are spread
-    /// over two event lanes, so splits also reach worker 0 through lazy repair.
+    /// edges but evict only leaves. Workers are spread over two event lanes, so
+    /// splits also reach worker 0 through lazy repair.
     #[test]
     fn randomized_removes_never_overcount_held_prefix() {
         const SEEDS: u64 = 32;
@@ -1175,6 +1175,164 @@ mod remove_tests {
             }
         }
     }
+
+    /// Worker 1's branch splits `[1, 2]` into `[1]` with children `[2]` and `[3]`. After
+    /// worker 0 evicts its head block, the child `[2]` still lists it. Worker 1 then holds
+    /// `[1]` alone and worker 0 the child alone, so both sets have one member but name
+    /// different workers. The walk must intersect them rather than carry worker 1 into the
+    /// child on the size match.
+    #[test]
+    fn equal_sized_coverage_of_another_worker_does_not_extend_match() {
+        let index = ConcurrentRadixTreeCompressed::new();
+        let mut lookup = direct_lookup();
+        apply_direct(&index, &mut lookup, make_store_event(0, &[1, 2]));
+        apply_direct(&index, &mut lookup, make_store_event(1, &[1, 3]));
+        apply_direct(&index, &mut lookup, make_remove_event(0, &[1]));
+
+        let scores = index.find_matches_impl(&local_hashes(&[1, 2]), false);
+        assert_eq!(scores.scores.get(&worker(1)), Some(&1));
+        assert_eq!(scores.scores.get(&worker(0)), None);
+    }
+
+    /// Randomized differential test against each rank's set of cached blocks. Every rank
+    /// removes tail-first, head-first or shuffled runs that may leave holes, stores restore
+    /// parts of evicted chains, and ranks are cleared one at a time, so unlinked subtrees
+    /// keep stale coverage and lookup entries. The tree may undercount there, but it must
+    /// never score a rank past the leading blocks the rank holds. `CRTC_REMOVE_FUZZ_SEEDS`
+    /// sets the number of seeds.
+    #[test]
+    fn random_hole_streams_never_overcount() {
+        let seeds = std::env::var("CRTC_REMOVE_FUZZ_SEEDS")
+            .ok()
+            .and_then(|seeds| seeds.parse().ok())
+            .unwrap_or(40u64);
+        // Lanes split ranks by worker id, as `ThreadPoolIndexer` does with two threads.
+        let ranks = [
+            WorkerWithDpRank::new(0, 0),
+            WorkerWithDpRank::new(0, 1),
+            WorkerWithDpRank::new(1, 0),
+            WorkerWithDpRank::new(2, 0),
+        ];
+        let mut overcounts = Vec::new();
+
+        for seed in 0..seeds {
+            let mut rng = fastrand::Rng::with_seed(seed);
+            // A base chain plus branches off prefixes of earlier chains.
+            let mut docs: Vec<Vec<u64>> = vec![(1..=24).collect()];
+            for branch in 1..10 {
+                let base = &docs[rng.usize(..docs.len())];
+                let mut doc = base[..rng.usize(1..base.len())].to_vec();
+                doc.extend((0..rng.u64(1..=12)).map(|i| 1_000 * branch + i));
+                docs.push(doc);
+            }
+            let seqs: Vec<Vec<u64>> = docs
+                .iter()
+                .map(|doc| compute_seq_hash_for_block(&local_hashes(doc)))
+                .collect();
+
+            let index = ConcurrentRadixTreeCompressed::new();
+            let mut lanes = [direct_lookup(), direct_lookup()];
+            let mut held = vec![FxHashSet::<u64>::default(); ranks.len()];
+            for step in 0..1_000 {
+                let r = rng.usize(..ranks.len());
+                let rank = ranks[r];
+                let d = rng.usize(..docs.len());
+                let (doc, seq) = (&docs[d], &seqs[d]);
+                let data = match rng.u32(..100) {
+                    0..=46 => {
+                        // Any cached parent can take a store, so evicted chains come back
+                        // in parts.
+                        let starts: Vec<usize> = (0..doc.len())
+                            .filter(|&start| start == 0 || held[r].contains(&seq[start - 1]))
+                            .collect();
+                        let start = starts[rng.usize(..starts.len())];
+                        let end = rng.usize(start + 1..=doc.len());
+                        held[r].extend(&seq[start..end]);
+                        let mut op = stored_data(make_store_event_full(
+                            rank.worker_id,
+                            &doc[..end],
+                            rank.dp_rank,
+                            None,
+                            None,
+                        ));
+                        op.parent_hash = start
+                            .checked_sub(1)
+                            .map(|parent| ExternalSequenceBlockHash(seq[parent]));
+                        op.blocks.drain(..start);
+                        KvCacheEventData::Stored(op)
+                    }
+                    47..=96 => {
+                        let cached: Vec<u64> = seq
+                            .iter()
+                            .copied()
+                            .filter(|hash| held[r].contains(hash))
+                            .collect();
+                        if cached.is_empty() {
+                            continue;
+                        }
+                        // Cached blocks from some position on, skipping any holes.
+                        let from = rng.usize(..cached.len());
+                        let len = rng.usize(1..=cached.len() - from);
+                        let mut evicted = cached[from..from + len].to_vec();
+                        match rng.u32(..3) {
+                            0 => evicted.reverse(),
+                            1 => rng.shuffle(&mut evicted),
+                            _ => {}
+                        }
+                        for hash in &evicted {
+                            held[r].remove(hash);
+                        }
+                        KvCacheEventData::Removed(KvCacheRemoveData {
+                            block_hashes: evicted
+                                .into_iter()
+                                .map(ExternalSequenceBlockHash)
+                                .collect(),
+                        })
+                    }
+                    _ => {
+                        held[r].clear();
+                        KvCacheEventData::Cleared
+                    }
+                };
+
+                let lookup = &mut lanes[(rank.worker_id % 2) as usize];
+                // A store under a parent the tree no longer holds is rejected, which
+                // only undercounts.
+                let _ = index.apply_event(
+                    lookup,
+                    router_event(rank.worker_id, step, rank.dp_rank, data),
+                    None,
+                );
+                lookup.assert_invariants();
+
+                for (doc, seq) in docs.iter().zip(&seqs) {
+                    let scores = index.find_matches_impl(&local_hashes(doc), false).scores;
+                    for (rank, held) in ranks.iter().zip(&held) {
+                        let prefix = seq.iter().take_while(|hash| held.contains(hash)).count();
+                        let score = scores.get(rank).map_or(0, |&score| score as usize);
+                        if score > prefix {
+                            overcounts.push((
+                                seed,
+                                format!(
+                                    "seed={seed} step={step} rank={rank:?} doc={doc:?} \
+                                     score={score} held_prefix={prefix}"
+                                ),
+                            ));
+                        }
+                    }
+                }
+            }
+        }
+
+        let failed_seeds: FxHashSet<_> = overcounts.iter().map(|&(seed, _)| seed).collect();
+        assert!(
+            overcounts.is_empty(),
+            "{} overcounting lookups in {} of {seeds} seeds, first: {}",
+            overcounts.len(),
+            failed_seeds.len(),
+            overcounts[0].1
+        );
+    }
 }
 
 mod structural_tests {