[no merge] conservative 1.0.6

core/src/update.rs

@@ -160,9 +160,16 @@ pub fn build_trie<H: NodeHasher>(
     }
 
     let common_after_prefix = |k1: &KeyPath, k2: &KeyPath| {
-        let x = &k1.view_bits::<Msb0>()[skip..];
-        let y = &k2.view_bits::<Msb0>()[skip..];
-        shared_bits(x, y)
+        let first_byte = skip / 8;
+        let shared = k1[first_byte..]
+            .iter()
+            .zip(&k2[first_byte..])
+            .position(|(a, b)| a != b)
+            .map_or(256, |i| {
+                (first_byte + i) * 8
+                    + (k1[first_byte + i] ^ k2[first_byte + i]).leading_zeros() as usize
+            });
+        shared - skip
     };
 
     while let Some((this_key, this_val)) = b {

nomt/src/beatree/iterator.rs

@@ -235,6 +235,27 @@ impl LeafIterator {
     }
 
     fn next<'a>(&'a mut self) -> Option<IterOutput<'a>> {
+        // Most steps remain within the same leaf. Advance its cursor in place instead
+        // of moving the entire state, including its branch and leaf handles, twice.
+        let within_leaf = match &self.state {
+            LeafIteratorState::Proceeding { current, .. } => {
+                let next = current.index + 1;
+                next < current.leaf.n()
+                    && self
+                        .end
+                        .as_ref()
+                        .map_or(true, |end| current.leaf.key(next) < *end)
+            }
+            _ => false,
+        };
+        if within_leaf {
+            let LeafIteratorState::Proceeding { current, .. } = &mut self.state else {
+                unreachable!();
+            };
+            current.index += 1;
+            return Some(current.last_output());
+        }
+
         // This is a so-called "Streaming Iterator", where the items borrow the lifetime
         // of the callsite rather than consuming items from the object - these are somewhat awkward
         // because we first have to advance the state of the iterator and then return a borrow

nomt/src/beatree/leaf_cache.rs

@@ -50,6 +50,11 @@ impl LeafCache {
         shard.cache.put(page_number, node);
     }
 
+    /// Remove a superseded page. Existing readers can still reload it from disk.
+    pub(super) fn remove(&self, page_number: PageNumber) {
+        self.inner.shard_for(page_number).cache.pop(&page_number);
+    }
+
     /// Evict all excess items from the cache.
     pub fn evict(&self) {
         for shard in &self.inner.shards {

nomt/src/beatree/ops/update/leaf_stage.rs

@@ -57,6 +57,9 @@ impl PostIoWork {
     pub(super) fn run(self, leaf_cache: &LeafCache) {
         for worker_output in self.worker_changes.into_iter() {
             for (_, leaf_entry) in worker_output.leaves_tracker.inner {
+                if let Some(pn) = leaf_entry.deleted {
+                    leaf_cache.remove(pn);
+                }
                 if let Some((leaf, pn)) = leaf_entry.inserted {
                     leaf_cache.insert(pn, leaf);
                 }

nomt/src/merkle/page_walker.rs

@@ -952,38 +952,24 @@ impl<H: NodeHasher> PageWalker<H> {
 /// Count the number of leaves present *only* in the provided page,
 /// without jumping into child pages.
 fn count_leaves<H: NodeHasher>(page: &PageMut) -> u64 {
-    // A simpler linear scan cannot be done because the page could contain some garbage.
+    // Follow only reachable nodes: unused slots can contain garbage. A page has
+    // DEPTH levels, so depth-first traversal needs at most DEPTH + 1 pending nodes.
+    let mut pending = [0usize; DEPTH + 1];
+    pending[1] = 1;
+    let mut len = 2;
     let mut counter = 0;
-
-    // We just need the node indexes within a page,
-    // so we treat this as the root page starting from the root position.
-    let mut pos = TriePosition::new();
-    let initial_depth = pos.depth();
-    pos.down(false);
-
-    loop {
-        let node = page.node(pos.node_index());
-        // Continue to traverse the left child if the current node is internal,
-        // stop if we reach the end of the page.
-        if trie::is_internal::<H>(&node) && pos.depth_in_page() != DEPTH {
-            pos.down(false);
-            continue;
-        }
-
-        if trie::is_leaf::<H>(&node) {
-            counter += 1;
-        }
-
-        // Going up until I reach a leaf sibling or the node I started from.
-        while pos.depth() != initial_depth && pos.peek_last_bit() {
-            pos.up(1);
-        }
-
-        if pos.depth() == initial_depth {
-            break;
+    while len != 0 {
+        len -= 1;
+        let index = pending[len];
+        match NodeKind::of::<H>(&page.node(index)) {
+            NodeKind::Leaf => counter += 1,
+            NodeKind::Internal if index < (1 << DEPTH) - 2 => {
+                pending[len] = index * 2 + 2;
+                pending[len + 1] = index * 2 + 3;
+                len += 2;
+            }
+            _ => {}
         }
-
-        pos.sibling();
     }
     counter
 }