fix(kv-router): reclaim CRTC dead leaves as they die

lib/kv-router/src/cleanup.rs

@@ -4,12 +4,10 @@
 //! Shared stale-child cleanup machinery for rooted tree structures.
 //!
 //! Provides a throttled, one-in-flight sweep that unlinks empty child nodes
-//! from their parent. It is used by
-//! [`ConcurrentRadixTreeCompressed`](super::concurrent_radix_tree_compressed)
-//! and the sequence-side
+//! from their parent. It is used by the sequence-side
 //! [`PromptMembershipTrie`](super::sequences::prompt_membership_trie::PromptMembershipTrie),
-//! each of which embeds a [`CleanupState`] and implements [`CleanableNode`]
-//! for its node type.
+//! which embeds a [`CleanupState`] and implements [`CleanableNode`] for its
+//! node type.
 //!
 //! # Sweep semantics
 //!
@@ -80,10 +78,6 @@ impl CleanupState {
             .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
             .is_ok()
     }
-
-    pub fn cancel(&self) {
-        self.scheduled.store(false, Ordering::Release);
-    }
 }
 
 impl Default for CleanupState {

lib/kv-router/src/indexer/concurrent_radix_tree_compressed/README.md

@@ -77,6 +77,10 @@ Each node contains:
   means the worker has cached `edge[0..k]`, with `0 < k < edge.len()`.
 - `children`: child nodes keyed by the first `LocalBlockHash` of the child
   edge.
+- `parent_tail`: the hash of the block before `edge[0]`, or 0 under the root.
+  A node's head never moves, so however often the tree splits, its parent is
+  the node whose edge ends with this block. See
+  [Dead Leaf Reclamation](#dead-leaf-reclamation).
 - `shape_gate` and `shape_version`: a per-node shape guard used to validate
   plans across lock gaps.
 - `internal`: a sticky marker that becomes true once the node has had children.
@@ -174,6 +178,44 @@ After the coverage update, removal may clear children only when no full-edge
 workers remain. Because `internal` is sticky, clearing those children does not
 make the node eligible for future leaf extension.
 
+A removal that takes a node's last holder leaves it linked as a dead leaf; the
+lane queues it for [reclamation](#dead-leaf-reclamation).
+
+## Dead Leaf Reclamation
+
+Removal changes coverage only, so a leaf whose last holder evicts it stays
+linked under its parent until something unlinks it. Under tail-first eviction
+almost every dead node is such a leaf, and they can hold most of the tree's
+memory. Unlinking one is a structural change on the parent: it takes the
+parent's exclusive shape gate and bumps `shape_version`. Neither reclaimer
+waits for that gate.
+
+- **Per-lane drain.** When a removal takes a leaf's last holder, the lane finds
+  the leaf's parent and queues the pair. The parent is the root, or the node
+  whose edge ends with the leaf's `parent_tail`: an anchor, or the node the
+  evicting worker's lookup names for that block. The parent is verified by
+  identity, and a stale lookup entry is not repaired on this path. The lane
+  unlinks queued leaves whenever no task is waiting, at most 256 at a time,
+  and all of them before a `Flush` completes. A queue that reaches 4,096
+  drains a quarter at once. Each parent's leaves go under one try-locked gate
+  and one version bump. A split that races an eviction on another lane can
+  produce a suffix that is dead from birth, which no removal will kill; the
+  splitting lane queues it with the prefix as its parent.
+- **Background sweep.** `ThreadPoolIndexer` runs a `kv-maint` thread that
+  sweeps once 1,024 nodes have died since the last sweep or 30 s have passed,
+  checking every 50 ms. The walk takes no gates: it records holder-less
+  children under the root and every anchor, descending only into nodes with
+  children. Each parent's candidates are re-checked and unlinked under one
+  try-locked gate, deepest parents first. The sweep catches what the drain
+  missed: leaves whose parent was busy or not in the evicting worker's lookup,
+  and leaves something still referenced.
+
+Both unlink a child only while it is still attached, holder-less and
+childless, its own shape gate is free, and nothing references it but the
+parent's map and the unlinker (`Arc::strong_count == 2`). A lane lookup that
+still names the leaf, or a reader mid-walk, keeps it for a later sweep. A store
+that reaches a dead leaf before it is unlinked re-covers it.
+
 ## Lookup Repair
 
 Cross-thread splits can make a worker lookup entry stale: the lookup still points

lib/kv-router/src/indexer/concurrent_radix_tree_compressed/children.rs

@@ -104,6 +104,24 @@ impl NodeChildren {
         }
     }
 
+    /// Visits each child by reference, without cloning its `Arc`.
+    pub(super) fn for_each_child(&self, mut visit: impl FnMut(LocalBlockHash, &SharedNode)) {
+        match &**self.state.load() {
+            ChildrenState::Empty => {}
+            ChildrenState::Singleton(entry) => visit(entry.hash, &entry.node),
+            ChildrenState::Small(children) => {
+                for entry in children.iter() {
+                    visit(entry.hash, &entry.node);
+                }
+            }
+            ChildrenState::Sharded(children) => {
+                for entry in children.iter() {
+                    visit(*entry.key(), entry.value());
+                }
+            }
+        }
+    }
+
     pub(super) fn values_snapshot(&self) -> Vec<SharedNode> {
         match &**self.state.load() {
             ChildrenState::Empty => Vec::new(),
@@ -117,6 +135,7 @@ impl NodeChildren {
         }
     }
 
+    #[cfg(test)]
     pub(super) fn entries_snapshot(&self) -> Vec<(LocalBlockHash, SharedNode)> {
         match &**self.state.load() {
             ChildrenState::Empty => Vec::new(),

lib/kv-router/src/indexer/concurrent_radix_tree_compressed/lane_lookup.rs

@@ -20,6 +20,7 @@ use std::sync::Arc;
 use rustc_hash::FxHashMap;
 
 use super::block_lookup::BlockLookup;
+use super::node::CleanupEdge;
 use super::types::SharedNode;
 use crate::protocols::{ExternalSequenceBlockHash, WorkerWithDpRank};
 
@@ -139,6 +140,9 @@ impl NodeHolds {
 pub(super) struct LaneLookup {
     workers: FxHashMap<WorkerWithDpRank, BlockLookup<NodeId>>,
     nodes: NodeHolds,
+    /// Leaves this lane's removals left dead, with their parents, awaiting a batched
+    /// unlink when the lane goes idle.
+    pub(super) pending_unlinks: Vec<CleanupEdge>,
 }
 
 impl LaneLookup {
@@ -215,7 +219,7 @@ impl LaneLookup {
     where
         I: Iterator<Item = ExternalSequenceBlockHash> + Clone,
     {
-        let Self { workers, nodes } = self;
+        let Self { workers, nodes, .. } = self;
         let id = nodes.hold(node);
         let changed = workers
             .entry(worker)

lib/kv-router/src/indexer/concurrent_radix_tree_compressed/mod.rs

@@ -11,14 +11,13 @@ use std::sync::Arc;
 use dashmap::DashMap;
 use rustc_hash::{FxBuildHasher, FxHashMap, FxHashSet};
 use std::collections::VecDeque;
-#[cfg(any(test, feature = "bench"))]
 use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
 
 use super::{
     AnchorRef, AnchorTask, EventKind, EventWarningKind, KvIndexerMetrics, KvRouterError,
     MatchDetails, PreBoundEventCounters, SyncIndexer, WorkerLookupStats, WorkerTask,
 };
-use crate::cleanup::{CleanupGuard, CleanupState};
 use crate::protocols::*;
 
 mod block_lookup;
@@ -50,12 +49,34 @@ pub struct ConcurrentRadixTreeCompressed {
     root: SharedNode,
 
     anchor_nodes: DashMap<ExternalSequenceBlockHash, SharedNode, FxBuildHasher>,
-    cleanup: CleanupState,
     lifecycle: super::HashLifecycle,
+    sweep_schedule: SweepSchedule,
+    /// Nodes that lost their last holder since the last stale-leaf sweep began.
+    deaths_since_sweep: AtomicU64,
+    /// When the last stale-leaf sweep began.
+    last_sweep: parking_lot::Mutex<Instant>,
     #[cfg(any(test, feature = "bench"))]
     bench_metrics: CrtcBenchMetrics,
 }
 
+/// How often the background maintenance thread checks whether a stale-leaf sweep is due.
+const SWEEP_TICK: Duration = Duration::from_millis(50);
+/// A sweep is due once this many nodes have died since the last one began...
+const SWEEP_AFTER_DEATHS: u64 = 1024;
+/// ...or once this long has passed, for nodes no death was counted for.
+const SWEEP_MAX_INTERVAL: Duration = Duration::from_secs(30);
+
+/// Most queued dead leaves an idle event lane unlinks before it checks for new events.
+const IDLE_UNLINK_LIMIT: usize = 256;
+
+/// When the background maintenance thread sweeps stale leaves.
+struct SweepSchedule {
+    /// How often the thread checks, or `None` for no maintenance thread.
+    tick: Option<Duration>,
+    after_deaths: u64,
+    max_interval: Duration,
+}
+
 #[cfg(any(test, feature = "bench"))]
 struct CrtcBenchMetrics {
     node_splits: AtomicU64,
@@ -121,13 +142,57 @@ impl ConcurrentRadixTreeCompressed {
         Self {
             root: Arc::new(Node::new()),
             anchor_nodes: DashMap::with_hasher(FxBuildHasher),
-            cleanup: CleanupState::new(),
             lifecycle: super::HashLifecycle::default(),
+            sweep_schedule: SweepSchedule {
+                tick: Some(SWEEP_TICK),
+                after_deaths: SWEEP_AFTER_DEATHS,
+                max_interval: SWEEP_MAX_INTERVAL,
+            },
+            deaths_since_sweep: AtomicU64::new(0),
+            last_sweep: parking_lot::Mutex::new(Instant::now()),
             #[cfg(any(test, feature = "bench"))]
             bench_metrics: CrtcBenchMetrics::new(),
         }
     }
 
+    /// Overrides when stale leaves are swept; `tick: None` runs no maintenance thread.
+    #[cfg(test)]
+    pub(crate) fn with_sweep_schedule(
+        mut self,
+        tick: Option<Duration>,
+        after_deaths: u64,
+        max_interval: Duration,
+    ) -> Self {
+        self.sweep_schedule = SweepSchedule {
+            tick,
+            after_deaths,
+            max_interval,
+        };
+        self
+    }
+
+    /// Counts a node that lost its last holder toward the next stale-leaf sweep. Call
+    /// only after dropping every reference to it, so a sweep that sees the count can
+    /// unlink it.
+    fn note_death(&self) {
+        self.deaths_since_sweep.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// One maintenance tick: sweeps stale leaves once enough nodes have died or the
+    /// schedule's longest interval has passed.
+    fn sweep_if_due(&self) {
+        let schedule = &self.sweep_schedule;
+        let due = self.deaths_since_sweep.load(Ordering::Relaxed) >= schedule.after_deaths
+            || self.last_sweep.lock().elapsed() >= schedule.max_interval;
+        if !due {
+            return;
+        }
+        // Deaths during the sweep count toward the next one.
+        self.deaths_since_sweep.store(0, Ordering::Relaxed);
+        *self.last_sweep.lock() = Instant::now();
+        self.sweep_stale_children();
+    }
+
     #[cfg(test)]
     pub(crate) fn raw_child_edge_count(&self) -> usize {
         let mut queue = VecDeque::from([self.root.clone()]);
@@ -234,6 +299,10 @@ impl ConcurrentRadixTreeCompressed {
         self.bench_metrics
             .node_splits
             .fetch_add(1, Ordering::Relaxed);
+        // A split racing an eviction on another lane can leave the suffix with no holder
+        // and no children from birth. No removal will kill it, so queue it here, where its
+        // parent is known.
+        self.queue_split_born_dead_leaf(lookup, prefix, &split.suffix);
         if !lookup.names(prefix) {
             return;
         }