From 5f565e1f9e1233d2cc396c4ee951b4061e895f2e Mon Sep 17 00:00:00 2001
From: Codex <codex@openai.com>
Date: Thu, 8 Oct 2026 18:36:08 +0200
Subject: [PATCH 01/17] Evict superseded leaf cache pages after write
 completion

---
 nomt/src/beatree/leaf_cache.rs            | 5 +++++
 nomt/src/beatree/ops/update/leaf_stage.rs | 8 ++++++++
 2 files changed, 13 insertions(+)

diff --git a/nomt/src/beatree/leaf_cache.rs b/nomt/src/beatree/leaf_cache.rs
index da18639d86..7915f95655 100644
--- a/nomt/src/beatree/leaf_cache.rs
+++ b/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 {
diff --git a/nomt/src/beatree/ops/update/leaf_stage.rs b/nomt/src/beatree/ops/update/leaf_stage.rs
index 8ddb37fbab..0fa075020f 100644
--- a/nomt/src/beatree/ops/update/leaf_stage.rs
+++ b/nomt/src/beatree/ops/update/leaf_stage.rs
@@ -55,6 +55,14 @@ pub struct PostIoWork {
 
 impl PostIoWork {
     pub(super) fn run(self, leaf_cache: &LeafCache) {
+        // Evict old pages before inserting any replacements.
+        for worker_output in &self.worker_changes {
+            for leaf_entry in worker_output.leaves_tracker.inner.values() {
+                if let Some(pn) = leaf_entry.deleted {
+                    leaf_cache.remove(pn);
+                }
+            }
+        }
         for worker_output in self.worker_changes.into_iter() {
             for (_, leaf_entry) in worker_output.leaves_tracker.inner {
                 if let Some((leaf, pn)) = leaf_entry.inserted {

From 710e510f353de4e814497663ba236bf9cc60845d Mon Sep 17 00:00:00 2001
From: Nikolai Golub <nikolay.v.golub@gmail.com>
Date: Tue, 13 Oct 2026 10:52:03 +0200
Subject: [PATCH 02/17] clean up same thing

---
 nomt/src/beatree/ops/update/leaf_stage.rs | 9 ++-------
 1 file changed, 2 insertions(+), 7 deletions(-)

diff --git a/nomt/src/beatree/ops/update/leaf_stage.rs b/nomt/src/beatree/ops/update/leaf_stage.rs
index 0fa075020f..8eed7fe8ad 100644
--- a/nomt/src/beatree/ops/update/leaf_stage.rs
+++ b/nomt/src/beatree/ops/update/leaf_stage.rs
@@ -55,16 +55,11 @@ pub struct PostIoWork {
 
 impl PostIoWork {
     pub(super) fn run(self, leaf_cache: &LeafCache) {
-        // Evict old pages before inserting any replacements.
-        for worker_output in &self.worker_changes {
-            for leaf_entry in worker_output.leaves_tracker.inner.values() {
+        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);
                 }
-            }
-        }
-        for worker_output in self.worker_changes.into_iter() {
-            for (_, leaf_entry) in worker_output.leaves_tracker.inner {
                 if let Some((leaf, pn)) = leaf_entry.inserted {
                     leaf_cache.insert(pn, leaf);
                 }

From 46fdaecd1e3040ea33cfdb2985241136cd7aa5f3 Mon Sep 17 00:00:00 2001
From: Codex <codex@openai.com>
Date: Thu, 8 Oct 2026 17:17:40 +0200
Subject: [PATCH 03/17] perf: advance B-tree leaf cursors in place

---
 nomt/src/beatree/iterator.rs | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/nomt/src/beatree/iterator.rs b/nomt/src/beatree/iterator.rs
index 6f23963edd..b5a74fd979 100644
--- a/nomt/src/beatree/iterator.rs
+++ 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

From 6b14ce4aab3248695e39ea63ec9228e9eaa7836c Mon Sep 17 00:00:00 2001
From: Codex <codex@openai.com>
Date: Thu, 8 Oct 2026 17:19:28 +0200
Subject: [PATCH 04/17] perf: count page leaves with a compact index stack

---
 nomt/src/merkle/page_walker.rs | 46 ++++++++++++----------------------
 1 file changed, 16 insertions(+), 30 deletions(-)

diff --git a/nomt/src/merkle/page_walker.rs b/nomt/src/merkle/page_walker.rs
index 93044a8fc0..a924b191bb 100644
--- a/nomt/src/merkle/page_walker.rs
+++ b/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
 }

From a4452035ad75ef88a50630b26070b38346a4beed Mon Sep 17 00:00:00 2001
From: Codex <codex@openai.com>
Date: Thu, 8 Oct 2026 17:39:37 +0200
Subject: [PATCH 05/17] Compare trie key prefixes by bytes

---
 core/src/update.rs | 13 ++++++++++---
 1 file changed, 10 insertions(+), 3 deletions(-)

diff --git a/core/src/update.rs b/core/src/update.rs
index 5aa857a38b..f86f3a3ae6 100644
--- a/core/src/update.rs
+++ b/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 {

From 8104051ea51b4d090f49131c5e2cecabecfbd464 Mon Sep 17 00:00:00 2001
From: Nikolai Golub <nikolay.v.golub@gmail.com>
Date: Tue, 13 Oct 2026 14:45:22 +0200
Subject: [PATCH 06/17] Move overlay changes into the commit instead of cloning
 them

Overlay::commit cloned every page and value change before writing them
out. Validate the parent marker and root under the write lock, mark the
overlay committed, then move the changes out. Data is only cloned if a
live overlay still holds it.

Validating before marking also stops a rejected commit from leaving the
overlay marked as committed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 nomt/src/lib.rs     | 60 ++++++++++++---------------------------------
 nomt/src/overlay.rs | 31 +++++++++++++++++++++++
 2 files changed, 47 insertions(+), 44 deletions(-)

diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs
index bab4adad00..35dc0b0d30 100644
--- a/nomt/src/lib.rs
+++ b/nomt/src/lib.rs
@@ -767,30 +767,14 @@ impl Overlay {
     /// This will return an error if I/O fails or if the changeset is no longer valid, or if the
     /// overlay has an uncommitted parent. An overlay may be invalidated by a competing commit or
     /// rollback.
-    pub fn commit<T: HashAlgorithm>(self, nomt: &Nomt<T>) -> anyhow::Result<()> {
-        if !self.parent_matches_marker(nomt.shared.lock().last_commit_marker.as_ref()) {
-            anyhow::bail!("Overlay parent not committed");
-        }
-
-        let root = self.root();
-        let page_changes: Vec<_> = self
-            .page_changes()
-            .into_iter()
-            .map(|(page_id, dirty_page)| (page_id.clone(), dirty_page.clone()))
-            .collect();
-        let values: Vec<_> = self
-            .value_changes()
-            .iter()
-            .map(|(k, v)| (k.clone(), v.clone()))
-            .collect();
-        let rollback_delta = self.rollback_delta().map(|delta| delta.clone());
-
+    pub fn commit<T: HashAlgorithm>(mut self, nomt: &Nomt<T>) -> anyhow::Result<()> {
         let _write_guard = nomt.access_lock.write();
 
-        let marker = self.mark_committed();
-
         {
             let mut shared = nomt.shared.lock();
+            if !self.parent_matches_marker(shared.last_commit_marker.as_ref()) {
+                anyhow::bail!("Overlay parent not committed");
+            }
             if shared.root != self.prev_root() {
                 anyhow::bail!(
                     "Changeset no longer valid (expected previous root {:?}, got {:?})",
@@ -798,10 +782,12 @@ impl Overlay {
                     shared.root
                 );
             }
-            shared.root = root;
-            shared.last_commit_marker = Some(marker);
+            shared.root = self.root();
+            shared.last_commit_marker = Some(self.mark_committed());
         }
 
+        let (page_changes, values, rollback_delta) = self.take_committed_changes();
+
         if let Some(rollback_delta) = rollback_delta {
             // UNWRAP: if rollback_delta is `Some`, then rollback must be also `Some`.
             let rollback = nomt.store.rollback().unwrap();
@@ -820,35 +806,19 @@ impl Overlay {
     /// overlay has an uncommitted parent. An overlay may be invalidated by a competing commit or
     /// rollback.
     pub fn try_commit_nonblocking<T: HashAlgorithm>(
-        self,
+        mut self,
         nomt: &Nomt<T>,
     ) -> anyhow::Result<Option<Self>> {
-        if !self.parent_matches_marker(nomt.shared.lock().last_commit_marker.as_ref()) {
-            anyhow::bail!("Overlay parent not committed");
-        }
-
-        let root = self.root();
-        let page_changes: Vec<_> = self
-            .page_changes()
-            .into_iter()
-            .map(|(page_id, dirty_page)| (page_id.clone(), dirty_page.clone()))
-            .collect();
-        let values: Vec<_> = self
-            .value_changes()
-            .iter()
-            .map(|(k, v)| (k.clone(), v.clone()))
-            .collect();
-        let rollback_delta = self.rollback_delta().map(|delta| delta.clone());
-
         let write_guard = nomt.access_lock.try_write();
         if write_guard.is_none() {
             return Ok(Some(self));
         }
 
-        let marker = self.mark_committed();
-
         {
             let mut shared = nomt.shared.lock();
+            if !self.parent_matches_marker(shared.last_commit_marker.as_ref()) {
+                anyhow::bail!("Overlay parent not committed");
+            }
             if shared.root != self.prev_root() {
                 anyhow::bail!(
                     "Changeset no longer valid (expected previous root {:?}, got {:?})",
@@ -856,10 +826,12 @@ impl Overlay {
                     shared.root
                 );
             }
-            shared.root = root;
-            shared.last_commit_marker = Some(marker);
+            shared.root = self.root();
+            shared.last_commit_marker = Some(self.mark_committed());
         }
 
+        let (page_changes, values, rollback_delta) = self.take_committed_changes();
+
         if let Some(rollback_delta) = rollback_delta {
             // UNWRAP: if rollback_delta is `Some`, then rollback must be also `Some`.
             let rollback = nomt.store.rollback().unwrap();
diff --git a/nomt/src/overlay.rs b/nomt/src/overlay.rs
index b6b4a28873..3b1a00e922 100644
--- a/nomt/src/overlay.rs
+++ b/nomt/src/overlay.rs
@@ -70,6 +70,36 @@ impl Overlay {
         self.inner.rollback_delta.as_ref()
     }
 
+    /// Move the changes out of the overlay, cloning only what is still shared.
+    ///
+    /// Must be called after marking the overlay committed: if the data has to be detached, the
+    /// remaining copy is dropped later and must not mark the overlay as dropped.
+    pub(super) fn take_committed_changes(
+        &mut self,
+    ) -> (
+        HashMap<PageId, DirtyPage>,
+        HashMap<KeyPath, ValueChange>,
+        Option<crate::rollback::Delta>,
+    ) {
+        match Arc::get_mut(&mut self.inner) {
+            Some(inner) => {
+                // Clones only if a live overlay still holds the data. Weak references from
+                // descendants are detached; the data is no longer needed once committed.
+                let data = Arc::make_mut(&mut inner.data);
+                (
+                    std::mem::take(&mut data.pages),
+                    std::mem::take(&mut data.values),
+                    inner.rollback_delta.take(),
+                )
+            }
+            None => (
+                self.page_changes().clone(),
+                self.value_changes().clone(),
+                self.rollback_delta().cloned(),
+            ),
+        }
+    }
+
     /// Mark the overlay as committed and return a marker.
     pub(super) fn mark_committed(&self) -> OverlayMarker {
         let status = self.inner.data.status.clone();
@@ -210,6 +240,7 @@ impl Index {
 }
 
 /// Data associated with a single overlay.
+#[derive(Clone)]
 struct Data {
     pages: HashMap<PageId, DirtyPage>,
     values: HashMap<KeyPath, ValueChange>,

From 6ef4f50891d82a8579d96a62cbd5d08fe4342bd3 Mon Sep 17 00:00:00 2001
From: Nikolai Golub <nikolay.v.golub@gmail.com>
Date: Tue, 13 Oct 2026 14:46:25 +0200
Subject: [PATCH 07/17] Overlap rollback persistence and post-meta cleanup with
 tree work

The rollback delta used to be appended and fsynced before the store
commit started. Pass it into Store::commit instead and append it while
bitbox and beatree prepare their writes. All three are joined before
the manifest is written, so the delta is still durable before the
manifest references it.

After the manifest, finish the beatree sync on its thread pool while
bitbox writes the hash-table pages.

FinishedSession::try_commit_nonblocking no longer needs
Rollback::commit_nonblocking: the rollback log is only touched under
the store's sync lock now, so it is removed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 nomt/src/beatree/mod.rs  | 23 ++++++++++++++------
 nomt/src/lib.rs          | 47 +++++++++++++---------------------------
 nomt/src/rollback/mod.rs | 21 ------------------
 nomt/src/store/mod.rs    |  2 ++
 nomt/src/store/sync.rs   | 31 ++++++++++++++++++++------
 5 files changed, 57 insertions(+), 67 deletions(-)

diff --git a/nomt/src/beatree/mod.rs b/nomt/src/beatree/mod.rs
index 547a763e34..60838b733a 100644
--- a/nomt/src/beatree/mod.rs
+++ b/nomt/src/beatree/mod.rs
@@ -483,13 +483,22 @@ impl SyncController {
 
     /// Finishes sync.
     ///
-    /// Has to be called after the manifest is updated. Must be invoked by the sync
-    /// thread. Blocking.
-    pub fn post_meta(&mut self) {
-        let pre_swap_rx = self.inner.pre_swap_rx.lock().take().unwrap();
-        join_task(&pre_swap_rx);
-        let bbn_index = self.inner.bbn_index.lock().take().unwrap();
-        Tree::finish_sync(&self.inner.shared, bbn_index);
+    /// Has to be called after the manifest is updated. Non-blocking: the returned task must be
+    /// joined before the commit completes.
+    pub fn post_meta(&mut self) -> Receiver<TaskResult<()>> {
+        let inner = self.inner.clone();
+        let (tx, rx) = crossbeam_channel::bounded(1);
+        spawn_task(
+            &self.inner.sync.tp,
+            move || {
+                let pre_swap_rx = inner.pre_swap_rx.lock().take().unwrap();
+                join_task(&pre_swap_rx);
+                let bbn_index = inner.bbn_index.lock().take().unwrap();
+                Tree::finish_sync(&inner.shared, bbn_index);
+            },
+            tx,
+        );
+        rx
     }
 }
 
diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs
index 35dc0b0d30..d480c7f2d9 100644
--- a/nomt/src/lib.rs
+++ b/nomt/src/lib.rs
@@ -691,18 +691,13 @@ impl FinishedSession {
             shared.last_commit_marker = None;
         }
 
-        if let Some(rollback_delta) = self.rollback_delta {
-            // UNWRAP: if rollback_delta is `Some`, then rollback must be also `Some`.
-            let rollback = nomt.store.rollback().unwrap();
-            rollback.commit(rollback_delta)?;
-        }
-
         nomt.store.commit(
             self.value_transaction.into_iter(),
             nomt.page_cache.clone(),
             self.merkle_output
                 .updated_pages
                 .into_frozen_iter(/* into_overlay */ false),
+            self.rollback_delta,
         )
     }
 
@@ -714,7 +709,7 @@ impl FinishedSession {
     /// The changeset may be invalidated if another competing session, overlay, or rollback was
     /// committed.
     pub fn try_commit_nonblocking<T: HashAlgorithm>(
-        mut self,
+        self,
         nomt: &Nomt<T>,
     ) -> Result<Option<Self>, anyhow::Error> {
         let write_guard = self
@@ -725,15 +720,6 @@ impl FinishedSession {
             return Ok(Some(self));
         }
 
-        if let Some(rollback_delta) = self.rollback_delta {
-            // UNWRAP: if rollback_delta is `Some`, then rollback must be also `Some`.
-            let rollback = nomt.store.rollback().unwrap();
-            if let Some(delta) = rollback.commit_nonblocking(rollback_delta)? {
-                self.rollback_delta = Some(delta);
-                return Ok(Some(self));
-            }
-        }
-
         {
             let mut shared = nomt.shared.lock();
             if shared.root != self.prev_root {
@@ -753,6 +739,7 @@ impl FinishedSession {
             self.merkle_output
                 .updated_pages
                 .into_frozen_iter(/* into_overlay */ false),
+            self.rollback_delta,
         )?;
 
         Ok(None)
@@ -788,14 +775,12 @@ impl Overlay {
 
         let (page_changes, values, rollback_delta) = self.take_committed_changes();
 
-        if let Some(rollback_delta) = rollback_delta {
-            // UNWRAP: if rollback_delta is `Some`, then rollback must be also `Some`.
-            let rollback = nomt.store.rollback().unwrap();
-            rollback.commit(rollback_delta)?;
-        }
-
-        nomt.store
-            .commit(values, nomt.page_cache.clone(), page_changes)
+        nomt.store.commit(
+            values,
+            nomt.page_cache.clone(),
+            page_changes,
+            rollback_delta,
+        )
     }
 
     /// Commit the changes from this overlay to the underlying database without blocking.
@@ -832,14 +817,12 @@ impl Overlay {
 
         let (page_changes, values, rollback_delta) = self.take_committed_changes();
 
-        if let Some(rollback_delta) = rollback_delta {
-            // UNWRAP: if rollback_delta is `Some`, then rollback must be also `Some`.
-            let rollback = nomt.store.rollback().unwrap();
-            rollback.commit(rollback_delta)?;
-        }
-
-        nomt.store
-            .commit(values, nomt.page_cache.clone(), page_changes)?;
+        nomt.store.commit(
+            values,
+            nomt.page_cache.clone(),
+            page_changes,
+            rollback_delta,
+        )?;
 
         Ok(None)
     }
diff --git a/nomt/src/rollback/mod.rs b/nomt/src/rollback/mod.rs
index b2d0f176a6..3deaa65b0b 100644
--- a/nomt/src/rollback/mod.rs
+++ b/nomt/src/rollback/mod.rs
@@ -173,27 +173,6 @@ impl Rollback {
         Ok(())
     }
 
-    /// Saves the delta into the log, unless the locks are already held by another thread.
-    ///
-    /// If commit is blocked, it returns the delta back to the caller.
-    pub fn commit_nonblocking(&self, delta: Delta) -> anyhow::Result<Option<Delta>> {
-        let delta_bytes = delta.encode();
-
-        // Try to lock the in-memory log and the seglog.
-        let mut in_memory = match self.shared.in_memory.try_lock() {
-            Some(lock) => lock,
-            None => return Ok(Some(delta)), // Another thread is holding the lock.
-        };
-        let mut seglog = match self.shared.seglog.try_lock() {
-            Some(lock) => lock,
-            None => return Ok(Some(delta)), // Another thread is holding the lock.
-        };
-
-        let record_id = seglog.append(&delta_bytes)?;
-        in_memory.push_recent(record_id, delta);
-        Ok(None)
-    }
-
     /// Truncates the rollback log by removing the last `n` deltas.
     ///
     /// This function returns the keys and values that we should apply to the database to restore
diff --git a/nomt/src/store/mod.rs b/nomt/src/store/mod.rs
index 7cb81a6886..a3f986baae 100644
--- a/nomt/src/store/mod.rs
+++ b/nomt/src/store/mod.rs
@@ -278,6 +278,7 @@ impl Store {
         value_tx: impl IntoIterator<Item = (beatree::Key, beatree::ValueChange)> + Send + 'static,
         page_cache: PageCache,
         updated_pages: impl IntoIterator<Item = (PageId, DirtyPage)> + Send + 'static,
+        rollback_delta: Option<crate::rollback::Delta>,
     ) -> anyhow::Result<()> {
         let mut sync = self.sync.lock();
 
@@ -295,6 +296,7 @@ impl Store {
             self.shared.pages.clone(),
             self.shared.values.clone(),
             self.shared.rollback.clone(),
+            rollback_delta,
             page_cache,
             updated_pages,
         ) {
diff --git a/nomt/src/store/sync.rs b/nomt/src/store/sync.rs
index 0ac076351a..12b53b01d5 100644
--- a/nomt/src/store/sync.rs
+++ b/nomt/src/store/sync.rs
@@ -4,7 +4,9 @@ use super::{
     meta::{self, Meta},
     DirtyPage, Shared,
 };
-use crate::{beatree, bitbox, options::PanicOnSyncMode, page_cache::PageCache, rollback};
+use crate::{
+    beatree, bitbox, options::PanicOnSyncMode, page_cache::PageCache, rollback, task::join_task,
+};
 
 pub struct Sync {
     pub(crate) sync_seqn: u32,
@@ -35,6 +37,7 @@ impl Sync {
         bitbox: bitbox::DB,
         beatree: beatree::Tree,
         rollback: Option<rollback::Rollback>,
+        rollback_delta: Option<rollback::Delta>,
         page_cache: PageCache,
         updated_pages: impl IntoIterator<Item = (PageId, DirtyPage)> + Send + 'static,
     ) -> anyhow::Result<()> {
@@ -42,18 +45,29 @@ impl Sync {
 
         let mut bitbox_sync = bitbox.sync();
         let mut beatree_sync = beatree.sync();
-        let mut rollback_sync = rollback.map(|rollback| rollback.sync());
+        let mut rollback_sync = rollback.as_ref().map(|rollback| rollback.sync());
 
         bitbox_sync.begin_sync(sync_seqn, page_cache, updated_pages);
         beatree_sync.begin_sync(value_tx);
+
+        // Persist the rollback delta while the trees are being prepared. All of them must
+        // succeed before the manifest references their new state.
+        let rollback_result = match rollback_delta {
+            // UNWRAP: a rollback delta is only produced when rollback is enabled.
+            Some(delta) => rollback.as_ref().unwrap().commit(delta),
+            None => Ok(()),
+        };
+        let bitbox_result = bitbox_sync.wait_pre_meta();
+        let beatree_result = beatree_sync.wait_pre_meta();
+        rollback_result?;
+        bitbox_result?;
+        let beatree_meta_wd = beatree_result?;
+
         let (rollback_start_live, rollback_end_live) = match rollback_sync {
             Some(ref mut rollback) => rollback.begin_sync(),
             None => (0, 0),
         };
 
-        bitbox_sync.wait_pre_meta()?;
-        let beatree_meta_wd = beatree_sync.wait_pre_meta()?;
-
         if let Some(PanicOnSyncMode::PostWal) = self.panic_on_sync {
             panic!("panic_on_sync is true (post-wal)")
         }
@@ -82,8 +96,11 @@ impl Sync {
             rollback.post_meta();
         }
 
-        bitbox_sync.post_meta(shared.io_pool.make_handle())?;
-        beatree_sync.post_meta();
+        // Finish the beatree sync while the hash-table pages are written out.
+        let beatree_post_meta = beatree_sync.post_meta();
+        let bitbox_result = bitbox_sync.post_meta(shared.io_pool.make_handle());
+        join_task(&beatree_post_meta);
+        bitbox_result?;
 
         if let Some(ref rollback) = rollback_sync {
             rollback.wait_post_meta()?;

From 2179e77e33f475b52190283e02244773fdbb618c Mon Sep 17 00:00:00 2001
From: Nikolai Golub <nikolai@sovlabs.io>
Date: Tue, 20 Oct 2026 14:22:24 +0200
Subject: [PATCH 08/17] rename and interface change

---
 nomt/src/lib.rs     | 8 ++++----
 nomt/src/overlay.rs | 4 ++--
 2 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs
index 35dc0b0d30..1ef815c193 100644
--- a/nomt/src/lib.rs
+++ b/nomt/src/lib.rs
@@ -767,7 +767,7 @@ impl Overlay {
     /// This will return an error if I/O fails or if the changeset is no longer valid, or if the
     /// overlay has an uncommitted parent. An overlay may be invalidated by a competing commit or
     /// rollback.
-    pub fn commit<T: HashAlgorithm>(mut self, nomt: &Nomt<T>) -> anyhow::Result<()> {
+    pub fn commit<T: HashAlgorithm>(self, nomt: &Nomt<T>) -> anyhow::Result<()> {
         let _write_guard = nomt.access_lock.write();
 
         {
@@ -786,7 +786,7 @@ impl Overlay {
             shared.last_commit_marker = Some(self.mark_committed());
         }
 
-        let (page_changes, values, rollback_delta) = self.take_committed_changes();
+        let (page_changes, values, rollback_delta) = self.into_committed_changes();
 
         if let Some(rollback_delta) = rollback_delta {
             // UNWRAP: if rollback_delta is `Some`, then rollback must be also `Some`.
@@ -806,7 +806,7 @@ impl Overlay {
     /// overlay has an uncommitted parent. An overlay may be invalidated by a competing commit or
     /// rollback.
     pub fn try_commit_nonblocking<T: HashAlgorithm>(
-        mut self,
+        self,
         nomt: &Nomt<T>,
     ) -> anyhow::Result<Option<Self>> {
         let write_guard = nomt.access_lock.try_write();
@@ -830,7 +830,7 @@ impl Overlay {
             shared.last_commit_marker = Some(self.mark_committed());
         }
 
-        let (page_changes, values, rollback_delta) = self.take_committed_changes();
+        let (page_changes, values, rollback_delta) = self.into_committed_changes();
 
         if let Some(rollback_delta) = rollback_delta {
             // UNWRAP: if rollback_delta is `Some`, then rollback must be also `Some`.
diff --git a/nomt/src/overlay.rs b/nomt/src/overlay.rs
index 3b1a00e922..37d6712b8c 100644
--- a/nomt/src/overlay.rs
+++ b/nomt/src/overlay.rs
@@ -74,8 +74,8 @@ impl Overlay {
     ///
     /// Must be called after marking the overlay committed: if the data has to be detached, the
     /// remaining copy is dropped later and must not mark the overlay as dropped.
-    pub(super) fn take_committed_changes(
-        &mut self,
+    pub(super) fn into_committed_changes(
+        mut self,
     ) -> (
         HashMap<PageId, DirtyPage>,
         HashMap<KeyPath, ValueChange>,

From 4263c8e7d733f3ce84f06f58be292814f42f4b1e Mon Sep 17 00:00:00 2001
From: Nikolai Golub <nikolai@sovlabs.io>
Date: Tue, 20 Oct 2026 15:10:25 +0200
Subject: [PATCH 09/17] explicit clone

---
 nomt/src/overlay.rs | 136 ++++++++++++++++++++++++++++++++------------
 1 file changed, 101 insertions(+), 35 deletions(-)

diff --git a/nomt/src/overlay.rs b/nomt/src/overlay.rs
index 37d6712b8c..8cd9c453b9 100644
--- a/nomt/src/overlay.rs
+++ b/nomt/src/overlay.rs
@@ -55,49 +55,29 @@ impl Overlay {
         }
     }
 
-    /// Get the merkle page changes associated uniquely with this overlay.
-    pub(super) fn page_changes(&self) -> &HashMap<PageId, DirtyPage> {
-        &self.inner.data.pages
-    }
-
-    /// Get the merkle page changes associated uniquely with this overlay.
-    pub(super) fn value_changes(&self) -> &HashMap<KeyPath, ValueChange> {
-        &self.inner.data.values
-    }
-
-    /// Get the rollback delta associated uniquely with this overlay.
-    pub(super) fn rollback_delta(&self) -> Option<&crate::rollback::Delta> {
-        self.inner.rollback_delta.as_ref()
-    }
-
     /// Move the changes out of the overlay, cloning only what is still shared.
     ///
-    /// Must be called after marking the overlay committed: if the data has to be detached, the
-    /// remaining copy is dropped later and must not mark the overlay as dropped.
+    /// Must be called after marking the overlay committed so dropping the extracted `Data`
+    /// does not mark the overlay as dropped.
     pub(super) fn into_committed_changes(
-        mut self,
+        self,
     ) -> (
         HashMap<PageId, DirtyPage>,
         HashMap<KeyPath, ValueChange>,
         Option<crate::rollback::Delta>,
     ) {
-        match Arc::get_mut(&mut self.inner) {
-            Some(inner) => {
-                // Clones only if a live overlay still holds the data. Weak references from
-                // descendants are detached; the data is no longer needed once committed.
-                let data = Arc::make_mut(&mut inner.data);
-                (
-                    std::mem::take(&mut data.pages),
-                    std::mem::take(&mut data.values),
-                    inner.rollback_delta.take(),
-                )
-            }
-            None => (
-                self.page_changes().clone(),
-                self.value_changes().clone(),
-                self.rollback_delta().cloned(),
+        let (data, rollback_delta) = match Arc::try_unwrap(self.inner) {
+            Ok(inner) => (inner.data, inner.rollback_delta),
+            Err(inner) => (Arc::clone(&inner.data), inner.rollback_delta.clone()),
+        };
+        let (pages, values) = match Arc::try_unwrap(data) {
+            Ok(mut data) => (
+                std::mem::take(&mut data.pages),
+                std::mem::take(&mut data.values),
             ),
-        }
+            Err(data) => (data.pages.clone(), data.values.clone()),
+        };
+        (pages, values, rollback_delta)
     }
 
     /// Mark the overlay as committed and return a marker.
@@ -240,7 +220,6 @@ impl Index {
 }
 
 /// Data associated with a single overlay.
-#[derive(Clone)]
 struct Data {
     pages: HashMap<PageId, DirtyPage>,
     values: HashMap<KeyPath, ValueChange>,
@@ -485,6 +464,93 @@ mod tests {
         assert!(LiveOverlay::new(None).is_ok());
     }
 
+    fn overlay_with_changes() -> super::Overlay {
+        LiveOverlay::new(None).unwrap().finish(
+            [0; 32],
+            [1; 32],
+            [(
+                ROOT_PAGE_ID,
+                dummy_page(ROOT_PAGE_ID, 1, BucketInfo::FreshWithNoDependents),
+            )]
+            .into_iter()
+            .collect(),
+            [([1; 32], ValueChange::Insert(vec![1, 2, 3]))]
+                .into_iter()
+                .collect(),
+            Some(crate::rollback::Delta {
+                priors: [([1; 32], Some(vec![4, 5, 6]))].into_iter().collect(),
+            }),
+        )
+    }
+
+    #[test]
+    fn committed_changes_move_with_weak_descendants() {
+        let overlay = overlay_with_changes();
+        let child = LiveOverlay::new([&overlay]).unwrap().finish(
+            [1; 32],
+            [2; 32],
+            HashMap::new(),
+            HashMap::new(),
+            None,
+        );
+        let value_ptr = overlay.inner.data.values[&[1; 32]]
+            .as_option()
+            .unwrap()
+            .as_ptr();
+        let prior_ptr = overlay.inner.rollback_delta.as_ref().unwrap().priors[&[1; 32]]
+            .as_ref()
+            .unwrap()
+            .as_ptr();
+        let marker = overlay.mark_committed();
+
+        let (pages, values, delta) = overlay.into_committed_changes();
+
+        assert_eq!(pages[&ROOT_PAGE_ID].page.node(0), [1; 32]);
+        assert_eq!(values[&[1; 32]], ValueChange::Insert(vec![1, 2, 3]));
+        assert_eq!(values[&[1; 32]].as_option().unwrap().as_ptr(), value_ptr);
+        let prior = delta.as_ref().unwrap().priors[&[1; 32]].as_ref().unwrap();
+        assert_eq!(prior, &[4, 5, 6]);
+        assert_eq!(prior.as_ptr(), prior_ptr);
+        assert!(child.parent_matches_marker(Some(&marker)));
+        assert!(LiveOverlay::new([&child]).is_ok());
+        assert!(marker.0.is_committed());
+    }
+
+    #[test]
+    fn committed_changes_preserve_live_parent_and_ancestor_data() {
+        for retain_parent in [true, false] {
+            let overlay = overlay_with_changes();
+            let child = LiveOverlay::new([&overlay]).unwrap().finish(
+                [1; 32],
+                [2; 32],
+                HashMap::new(),
+                HashMap::new(),
+                None,
+            );
+            let live = if retain_parent {
+                LiveOverlay::new([&overlay]).unwrap()
+            } else {
+                LiveOverlay::new([&child, &overlay]).unwrap()
+            };
+            let marker = overlay.mark_committed();
+
+            let (pages, values, delta) = overlay.into_committed_changes();
+
+            assert_eq!(pages[&ROOT_PAGE_ID].page.node(0), [1; 32]);
+            assert_eq!(values[&[1; 32]], ValueChange::Insert(vec![1, 2, 3]));
+            assert_eq!(delta.unwrap().priors[&[1; 32]], Some(vec![4, 5, 6]));
+            assert_eq!(live.page(&ROOT_PAGE_ID).unwrap().page.node(0), [1; 32]);
+            assert_eq!(
+                live.value(&[1; 32]),
+                Some(ValueChange::Insert(vec![1, 2, 3]))
+            );
+            assert!(child.parent_matches_marker(Some(&marker)));
+            assert!(LiveOverlay::new([&child]).is_ok());
+            drop(live);
+            assert!(marker.0.is_committed());
+        }
+    }
+
     #[test]
     fn not_ancestor() {
         let a = LiveOverlay::new(None).unwrap().finish(

From 2d970a6ade4128115bc00cfcdfea0702427de64a Mon Sep 17 00:00:00 2001
From: Nikolai Golub <nikolay.v.golub@gmail.com>
Date: Mon, 12 Oct 2026 13:44:30 +0200
Subject: [PATCH 10/17] Add Options::prepopulate_leaf_cache to fill the leaf
 cache on startup

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 nomt/src/beatree/index.rs            |  5 ++++
 nomt/src/beatree/leaf_cache.rs       |  5 ++++
 nomt/src/beatree/mod.rs              | 31 ++++++++++++++++++++
 nomt/src/lib.rs                      |  3 ++
 nomt/src/options.rs                  | 11 ++++++++
 nomt/src/store/mod.rs                |  5 ++++
 nomt/tests/prepopulate_leaf_cache.rs | 42 ++++++++++++++++++++++++++++
 7 files changed, 102 insertions(+)
 create mode 100644 nomt/tests/prepopulate_leaf_cache.rs

diff --git a/nomt/src/beatree/index.rs b/nomt/src/beatree/index.rs
index 6537c77541..6cf7362215 100644
--- a/nomt/src/beatree/index.rs
+++ b/nomt/src/beatree/index.rs
@@ -43,6 +43,11 @@ impl Index {
         self.first_key_map.insert(separator, branch)
     }
 
+    /// Iterate all bottom-level branch nodes.
+    pub fn branches(&self) -> impl Iterator<Item = &Arc<BranchNode>> {
+        self.first_key_map.values()
+    }
+
     #[cfg(test)]
     pub fn into_iter(self) -> impl Iterator<Item = (Key, Arc<BranchNode>)> {
         self.first_key_map.into_iter()
diff --git a/nomt/src/beatree/leaf_cache.rs b/nomt/src/beatree/leaf_cache.rs
index da18639d86..5c83057611 100644
--- a/nomt/src/beatree/leaf_cache.rs
+++ b/nomt/src/beatree/leaf_cache.rs
@@ -43,6 +43,11 @@ impl LeafCache {
         shard.cache.get(&page_number).map(|x| x.clone())
     }
 
+    /// The maximum number of leaves the cache holds.
+    pub fn capacity(&self) -> usize {
+        self.inner.shards.iter().map(|shard| shard.lock().max_items).sum()
+    }
+
     /// Insert a cache entry. This does not evict anything.
     pub fn insert(&self, page_number: PageNumber, node: Arc<LeafNode>) {
         let mut shard = self.inner.shard_for(page_number);
diff --git a/nomt/src/beatree/mod.rs b/nomt/src/beatree/mod.rs
index 547a763e34..3aa15f8d2d 100644
--- a/nomt/src/beatree/mod.rs
+++ b/nomt/src/beatree/mod.rs
@@ -141,6 +141,37 @@ impl Tree {
         })
     }
 
+    /// Fill the leaf cache with leaves, up to its capacity, reading them on `threads` threads.
+    pub fn prepopulate_leaf_cache(&self, threads: usize) {
+        let shared = self.shared.read();
+        let leaf_cache = shared.leaf_cache.clone();
+        let leaf_store = shared.leaf_store_rd.clone();
+        let leaves: Vec<PageNumber> = shared
+            .bbn_index
+            .branches()
+            .flat_map(|branch| {
+                (0..branch.n() as usize).map(|i| PageNumber(branch.node_pointer(i)))
+            })
+            .take(leaf_cache.capacity())
+            .collect();
+        drop(shared);
+
+        let chunk = leaves.len().div_ceil(threads.max(1)).max(1);
+        std::thread::scope(|scope| {
+            for chunk in leaves.chunks(chunk) {
+                let (leaf_cache, leaf_store) = (&leaf_cache, &leaf_store);
+                scope.spawn(move || {
+                    for &pn in chunk {
+                        let leaf = Arc::new(leaf::node::LeafNode {
+                            inner: leaf_store.query(pn),
+                        });
+                        leaf_cache.insert(pn, leaf);
+                    }
+                });
+            }
+        });
+    }
+
     /// Lookup a key in the btree. This blocks the current thread.
     pub fn lookup(&self, key: Key) -> Option<Vec<u8>> {
         let shared = self.shared.read();
diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs
index bab4adad00..e945615070 100644
--- a/nomt/src/lib.rs
+++ b/nomt/src/lib.rs
@@ -234,6 +234,9 @@ impl<T: HashAlgorithm> Nomt<T> {
             let io_handle = store.io_pool().make_handle();
             merkle::prepopulate_cache(io_handle, &page_cache, &store, o.page_cache_upper_levels)?;
         }
+        if o.prepopulate_leaf_cache {
+            store.prepopulate_leaf_cache(o.commit_concurrency);
+        }
 
         Ok(Self {
             merkle_update_pool: UpdatePool::new(o.commit_concurrency, o.warm_up),
diff --git a/nomt/src/options.rs b/nomt/src/options.rs
index a44652b733..bbbf0c2fa4 100644
--- a/nomt/src/options.rs
+++ b/nomt/src/options.rs
@@ -29,6 +29,7 @@ pub struct Options {
     /// Whether to prepopulate the upper layers of the page cache on startup.
     /// This incurs some I/O on startup but leads to predictable worst-case performance.
     pub(crate) prepopulate_page_cache: bool,
+    pub(crate) prepopulate_leaf_cache: bool,
     pub(crate) page_cache_upper_levels: usize,
 }
 
@@ -52,6 +53,7 @@ impl Options {
             page_cache_size: 256,
             leaf_cache_size: 256,
             prepopulate_page_cache: false,
+            prepopulate_leaf_cache: false,
             page_cache_upper_levels: 2,
         }
     }
@@ -170,6 +172,15 @@ impl Options {
         self.prepopulate_page_cache = prepopulate;
     }
 
+    /// Sets whether to fill the leaf cache with beatree leaves on startup, up to its capacity.
+    ///
+    /// This reads up to [`Self::leaf_cache_size`] of leaves on startup, so that sessions after a
+    /// restart do not pay for cold leaf reads.
+    /// Default: false
+    pub fn prepopulate_leaf_cache(&mut self, prepopulate: bool) {
+        self.prepopulate_leaf_cache = prepopulate;
+    }
+
     /// Sets the number of upper levels of the page tree to keep permanently
     /// cached.
     ///
diff --git a/nomt/src/store/mod.rs b/nomt/src/store/mod.rs
index 7cb81a6886..eb195104bd 100644
--- a/nomt/src/store/mod.rs
+++ b/nomt/src/store/mod.rs
@@ -243,6 +243,11 @@ impl Store {
         }
     }
 
+    /// Fill the beatree leaf cache on `threads` threads, up to its capacity.
+    pub fn prepopulate_leaf_cache(&self, threads: usize) {
+        self.shared.values.prepopulate_leaf_cache(threads);
+    }
+
     /// Creates a new [`beatree::ReadTransaction`]. `sync` will be blocked until this is dropped.
     pub fn read_transaction(&self) -> beatree::ReadTransaction {
         self.shared.values.read_transaction()
diff --git a/nomt/tests/prepopulate_leaf_cache.rs b/nomt/tests/prepopulate_leaf_cache.rs
new file mode 100644
index 0000000000..0d2e8b39e3
--- /dev/null
+++ b/nomt/tests/prepopulate_leaf_cache.rs
@@ -0,0 +1,42 @@
+use nomt::{hasher::Blake3Hasher, KeyReadWrite, Nomt, Options, SessionParams};
+use nomt_test_utils::account_path;
+use std::path::PathBuf;
+
+fn open(path: &PathBuf, prepopulate: bool) -> Nomt<Blake3Hasher> {
+    let mut o = Options::new();
+    o.path(path);
+    o.bitbox_seed([0; 16]);
+    o.commit_concurrency(2);
+    o.io_workers(1);
+    o.leaf_cache_size(1);
+    o.prepopulate_leaf_cache(prepopulate);
+    Nomt::open(o).unwrap()
+}
+
+fn value(id: u64) -> Vec<u8> {
+    vec![id as u8; 400]
+}
+
+#[test]
+fn prepopulate_leaf_cache_reads_back_values() {
+    let path = PathBuf::from("test/prepopulate_leaf_cache");
+    let _ = std::fs::remove_dir_all(&path);
+    // ~1.6 MiB of values: more leaves than a 1 MiB leaf cache holds.
+    let n = 4000;
+
+    {
+        let nomt = open(&path, false);
+        let session = nomt.begin_session(SessionParams::default());
+        let mut actuals: Vec<_> = (0..n)
+            .map(|id| (account_path(id), KeyReadWrite::Write(Some(value(id)))))
+            .collect();
+        actuals.sort_by_key(|(k, _)| *k);
+        session.finish(actuals).unwrap().commit(&nomt).unwrap();
+    }
+
+    let nomt = open(&path, true);
+    let session = nomt.begin_session(SessionParams::default());
+    for id in 0..n {
+        assert_eq!(session.read(account_path(id)).unwrap(), Some(value(id)));
+    }
+}

From 76a99d94e07c0ed8fc3c12e8f9650b8d12065c34 Mon Sep 17 00:00:00 2001
From: Nikolai Golub <nikolai@sovlabs.io>
Date: Wed, 21 Oct 2026 10:46:01 +0200
Subject: [PATCH 11/17] Hash session actuals for the merkle update on multiple
 threads

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 nomt/src/lib.rs | 35 +++++++++++++++++++++++++++++------
 1 file changed, 29 insertions(+), 6 deletions(-)

diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs
index bab4adad00..1477b333e3 100644
--- a/nomt/src/lib.rs
+++ b/nomt/src/lib.rs
@@ -585,14 +585,9 @@ impl<T: HashAlgorithm> Session<T> {
             .take()
             .map(|delta_builder| delta_builder.finalize(&actuals));
 
-        let mut compact_actuals = Vec::with_capacity(actuals.len());
-        for (path, read_write) in &actuals {
-            compact_actuals.push((path.clone(), read_write.to_compact::<T>()));
-        }
-
         let merkle_update_handle = self
             .merkle_updater
-            .update_and_prove::<T>(compact_actuals, self.witness_mode.0)?;
+            .update_and_prove::<T>(compact_actuals::<T>(&actuals), self.witness_mode.0)?;
 
         let mut tx = self.store.new_value_tx();
         for (path, read_write) in actuals {
@@ -613,6 +608,34 @@ impl<T: HashAlgorithm> Session<T> {
     }
 }
 
+/// Hashes the values of `actuals` for the merkle update, splitting large inputs across scoped threads.
+fn compact_actuals<T: HashAlgorithm>(
+    actuals: &[(KeyPath, KeyReadWrite)],
+) -> Vec<(KeyPath, crate::merkle::KeyReadWrite)> {
+    const MIN_CHUNK: usize = 2048;
+    let compact = |chunk: &[(KeyPath, KeyReadWrite)]| {
+        chunk
+            .iter()
+            .map(|(path, read_write)| (*path, read_write.to_compact::<T>()))
+            .collect::<Vec<_>>()
+    };
+    let threads = std::thread::available_parallelism().map_or(1, |n| n.get().min(16));
+    let chunk = actuals.len().div_ceil(threads).max(MIN_CHUNK);
+    if actuals.len() <= chunk {
+        return compact(actuals);
+    }
+    std::thread::scope(|scope| {
+        let handles: Vec<_> = actuals
+            .chunks(chunk)
+            .map(|chunk| scope.spawn(move || compact(chunk)))
+            .collect();
+        handles
+            .into_iter()
+            .flat_map(|handle| handle.join().expect("compacting thread panicked"))
+            .collect()
+    })
+}
+
 /// A finished session.
 ///
 /// This is the result of completing a session and computing the merkle root and merkle DB changes,

From fc91b25d2088c3c2ac17c0f94d585cd785862596 Mon Sep 17 00:00:00 2001
From: Nikolai Golub <nikolai@sovlabs.io>
Date: Wed, 21 Oct 2026 10:46:01 +0200
Subject: [PATCH 12/17] Build the rollback delta while the merkle update runs

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 nomt/src/lib.rs | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs
index bab4adad00..699d91c6ee 100644
--- a/nomt/src/lib.rs
+++ b/nomt/src/lib.rs
@@ -580,11 +580,6 @@ impl<T: HashAlgorithm> Session<T> {
                 );
             }
         }
-        let rollback_delta = self
-            .rollback_delta
-            .take()
-            .map(|delta_builder| delta_builder.finalize(&actuals));
-
         let mut compact_actuals = Vec::with_capacity(actuals.len());
         for (path, read_write) in &actuals {
             compact_actuals.push((path.clone(), read_write.to_compact::<T>()));
@@ -594,6 +589,13 @@ impl<T: HashAlgorithm> Session<T> {
             .merkle_updater
             .update_and_prove::<T>(compact_actuals, self.witness_mode.0)?;
 
+        // Built while the merkle update runs: both only read the store, which is not modified
+        // until the finished session is committed.
+        let rollback_delta = self
+            .rollback_delta
+            .take()
+            .map(|delta_builder| delta_builder.finalize(&actuals));
+
         let mut tx = self.store.new_value_tx();
         for (path, read_write) in actuals {
             if let KeyReadWrite::Write(value) | KeyReadWrite::ReadThenWrite(_, value) = read_write {

From acd216e9afe56b580f92e82c30bee824f243cd6e Mon Sep 17 00:00:00 2001
From: Nikolai Golub <nikolay.v.golub@gmail.com>
Date: Mon, 12 Oct 2026 12:35:26 +0200
Subject: [PATCH 13/17] Resolve a seek's own leaf from its read value hash
 without a beatree fetch

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 nomt/src/lib.rs           | 11 ++++++---
 nomt/src/merkle/mod.rs    | 29 ++++++++++++++--------
 nomt/src/merkle/seek.rs   | 51 ++++++++++++++++++++++++++++++++++++---
 nomt/src/merkle/worker.rs |  9 ++++---
 4 files changed, 78 insertions(+), 22 deletions(-)

diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs
index bab4adad00..e5a84f50b9 100644
--- a/nomt/src/lib.rs
+++ b/nomt/src/lib.rs
@@ -124,14 +124,17 @@ impl KeyReadWrite {
         }
     }
 
+    /// Hashes the values. The read value hash lets the merkle update recognize the key's own leaf
+    /// without fetching it.
     fn to_compact<T: HashAlgorithm>(&self) -> crate::merkle::KeyReadWrite {
         let hash = |v: &Value| T::hash_value(v);
         match self {
-            KeyReadWrite::Read(_) => crate::merkle::KeyReadWrite::Read,
+            KeyReadWrite::Read(read) => crate::merkle::KeyReadWrite::Read(read.as_ref().map(hash)),
             KeyReadWrite::Write(val) => crate::merkle::KeyReadWrite::Write(val.as_ref().map(hash)),
-            KeyReadWrite::ReadThenWrite(_, val) => {
-                crate::merkle::KeyReadWrite::ReadThenWrite(val.as_ref().map(hash))
-            }
+            KeyReadWrite::ReadThenWrite(read, val) => crate::merkle::KeyReadWrite::ReadThenWrite(
+                read.as_ref().map(hash),
+                val.as_ref().map(hash),
+            ),
         }
     }
 }
diff --git a/nomt/src/merkle/mod.rs b/nomt/src/merkle/mod.rs
index 809f2f8a11..12391914e0 100644
--- a/nomt/src/merkle/mod.rs
+++ b/nomt/src/merkle/mod.rs
@@ -142,33 +142,42 @@ pub enum BucketInfo {
 /// Whether a key was read or written.
 #[derive(Debug, Clone)]
 pub enum KeyReadWrite {
-    /// The key was read.
-    Read,
+    /// The key was read. Contains the read value hash or `None` if absent.
+    Read(Option<ValueHash>),
     /// The key was written. Contains the written value or `None` if deleted.
     Write(Option<ValueHash>),
-    /// The key was both read and written. Contains the written value of `None` if deleted.
-    ReadThenWrite(Option<ValueHash>),
+    /// The key was both read and written. Contains the read value hash (or `None` if absent) and
+    /// the written value hash (or `None` if deleted).
+    ReadThenWrite(Option<ValueHash>, Option<ValueHash>),
 }
 
 impl KeyReadWrite {
     fn is_write(&self) -> bool {
         match self {
-            KeyReadWrite::Read => false,
-            KeyReadWrite::Write(_) | KeyReadWrite::ReadThenWrite(_) => true,
+            KeyReadWrite::Read(_) => false,
+            KeyReadWrite::Write(_) | KeyReadWrite::ReadThenWrite(_, _) => true,
         }
     }
 
     fn is_read(&self) -> bool {
         match self {
-            KeyReadWrite::Read | KeyReadWrite::ReadThenWrite(_) => true,
+            KeyReadWrite::Read(_) | KeyReadWrite::ReadThenWrite(_, _) => true,
             KeyReadWrite::Write(_) => false,
         }
     }
 
+    /// The hash of the value the key was read with, if it was read and present.
+    fn read_value_hash(&self) -> Option<ValueHash> {
+        match self {
+            KeyReadWrite::Read(v) | KeyReadWrite::ReadThenWrite(v, _) => *v,
+            KeyReadWrite::Write(_) => None,
+        }
+    }
+
     fn written_value(&self) -> Option<Option<ValueHash>> {
         match self {
-            KeyReadWrite::Read => None,
-            KeyReadWrite::Write(v) | KeyReadWrite::ReadThenWrite(v) => {
+            KeyReadWrite::Read(_) => None,
+            KeyReadWrite::Write(v) | KeyReadWrite::ReadThenWrite(_, v) => {
                 Some(v.as_ref().map(|vh| *vh))
             }
         }
@@ -339,7 +348,7 @@ impl Updater {
             true, // collect witness
         );
 
-        seeker.push(key_path);
+        seeker.push(key_path, None);
         let mut page_set = PageSet::new(self.page_pool.clone(), None);
 
         // Blocking I/O loop.
diff --git a/nomt/src/merkle/seek.rs b/nomt/src/merkle/seek.rs
index c46dca66c4..d9257acbf1 100644
--- a/nomt/src/merkle/seek.rs
+++ b/nomt/src/merkle/seek.rs
@@ -38,6 +38,9 @@ const MAX_INFLIGHT: usize = 1024;
 
 struct SeekRequest {
     key: KeyPath,
+    /// Hash of the value the caller read for `key`, if it exists. A terminal leaf whose hash
+    /// matches `(key, known_value_hash)` is this key's leaf, so it needs no beatree fetch.
+    known_value_hash: Option<ValueHash>,
     position: TriePosition,
     page_id: Option<PageId>,
     siblings: Vec<Node>,
@@ -50,18 +53,27 @@ impl SeekRequest {
         read_transaction: &BeatreeReadTx,
         overlay: &LiveOverlay,
         key: KeyPath,
+        known_value_hash: Option<ValueHash>,
         root: Node,
     ) -> SeekRequest {
         let state = if trie::is_terminator::<H>(&root) {
             RequestState::Completed(None)
         } else if trie::is_leaf::<H>(&root) {
-            RequestState::begin_leaf_fetch::<H>(read_transaction, overlay, &TriePosition::new())
+            RequestState::resolve_leaf::<H>(
+                read_transaction,
+                overlay,
+                &TriePosition::new(),
+                &root,
+                key,
+                known_value_hash,
+            )
         } else {
             RequestState::Seeking
         };
 
         let mut request = SeekRequest {
             key,
+            known_value_hash,
             position: TriePosition::new(),
             page_id: None,
             siblings: Vec::new(),
@@ -144,8 +156,14 @@ impl SeekRequest {
             }
 
             if trie::is_leaf::<H>(&cur_node) {
-                self.state =
-                    RequestState::begin_leaf_fetch::<H>(read_transaction, overlay, &self.position);
+                self.state = RequestState::resolve_leaf::<H>(
+                    read_transaction,
+                    overlay,
+                    &self.position,
+                    &cur_node,
+                    self.key,
+                    self.known_value_hash,
+                );
                 if let RequestState::FetchingLeaf { .. } = self.state {
                     self.continue_leaf_fetch::<H>(None);
                 }
@@ -412,6 +430,28 @@ enum RequestState {
 }
 
 impl RequestState {
+    /// Resolves the leaf node `leaf_node` at `pos`: without I/O if it is the leaf of `key` with the
+    /// known value hash, otherwise by fetching the leaf data from the beatree.
+    fn resolve_leaf<H: HashAlgorithm>(
+        read_transaction: &BeatreeReadTx,
+        overlay: &LiveOverlay,
+        pos: &TriePosition,
+        leaf_node: &Node,
+        key: KeyPath,
+        known_value_hash: Option<ValueHash>,
+    ) -> Self {
+        if let Some(value_hash) = known_value_hash {
+            let leaf = trie::LeafData {
+                key_path: key,
+                value_hash,
+            };
+            if H::hash_leaf(&leaf) == *leaf_node {
+                return RequestState::Completed(Some(leaf));
+            }
+        }
+        Self::begin_leaf_fetch::<H>(read_transaction, overlay, pos)
+    }
+
     fn begin_leaf_fetch<H: HashAlgorithm>(
         read_transaction: &BeatreeReadTx,
         overlay: &LiveOverlay,
@@ -634,12 +674,15 @@ impl<H: HashAlgorithm> Seeker<H> {
     }
 
     /// Push a request for key path.
-    pub fn push(&mut self, key: KeyPath) {
+    ///
+    /// `known_value_hash` is the hash of the value the caller read for `key`, if any.
+    pub fn push(&mut self, key: KeyPath, known_value_hash: Option<ValueHash>) {
         let request_index = self.processed + self.requests.len();
         self.requests.push_back(SeekRequest::new::<H>(
             &self.beatree_read_transaction,
             &self.overlay,
             key,
+            known_value_hash,
             self.root,
         ));
         self.idle_requests.push_back(request_index);
diff --git a/nomt/src/merkle/worker.rs b/nomt/src/merkle/worker.rs
index cb6f060983..9d19015332 100644
--- a/nomt/src/merkle/worker.rs
+++ b/nomt/src/merkle/worker.rs
@@ -171,7 +171,7 @@ fn warm_up_phase<H: HashAlgorithm>(
                     Err(_) => panic!("Warm-Up worker, unexpected failure of the warmup channel"),
                 };
 
-                seeker.push(warm_up_command.key_path);
+                seeker.push(warm_up_command.key_path, None);
             } else if index == page_idx {
                 seeker.try_recv_page(&mut page_set)?;
             } else {
@@ -510,7 +510,8 @@ impl<H: HashAlgorithm> RangeUpdater<H> {
                         break;
                     }
                 } else {
-                    seeker.push(self.shared.read_write[next_push].0);
+                    let (key, read_write) = &self.shared.read_write[next_push];
+                    seeker.push(*key, read_write.read_value_hash());
                     seeker.submit_all(page_set);
                 }
             }
@@ -538,10 +539,10 @@ fn subtrie_ops(read_write: &[(KeyPath, KeyReadWrite)]) -> Vec<(KeyPath, Option<V
     read_write
         .iter()
         .filter_map(|(key, read_write)| match read_write {
-            KeyReadWrite::Write(val) | KeyReadWrite::ReadThenWrite(val) => {
+            KeyReadWrite::Write(val) | KeyReadWrite::ReadThenWrite(_, val) => {
                 Some((key.clone(), val.clone()))
             }
-            KeyReadWrite::Read => None,
+            KeyReadWrite::Read(_) => None,
         })
         .collect::<Vec<_>>()
 }

From 5788a55a78c77a57cabb55c063517ea93a3fefdf Mon Sep 17 00:00:00 2001
From: Nikolai Golub <nikolay.v.golub@gmail.com>
Date: Wed, 21 Oct 2026 11:47:57 +0200
Subject: [PATCH 14/17] Fix formatting and tidy up leaf cache prepopulation

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 nomt/src/beatree/leaf_cache.rs |  6 +++++-
 nomt/src/beatree/mod.rs        | 11 +++++------
 nomt/src/options.rs            |  5 +++--
 3 files changed, 13 insertions(+), 9 deletions(-)

diff --git a/nomt/src/beatree/leaf_cache.rs b/nomt/src/beatree/leaf_cache.rs
index 5c83057611..f3ad2271f8 100644
--- a/nomt/src/beatree/leaf_cache.rs
+++ b/nomt/src/beatree/leaf_cache.rs
@@ -45,7 +45,11 @@ impl LeafCache {
 
     /// The maximum number of leaves the cache holds.
     pub fn capacity(&self) -> usize {
-        self.inner.shards.iter().map(|shard| shard.lock().max_items).sum()
+        self.inner
+            .shards
+            .iter()
+            .map(|shard| shard.lock().max_items)
+            .sum()
     }
 
     /// Insert a cache entry. This does not evict anything.
diff --git a/nomt/src/beatree/mod.rs b/nomt/src/beatree/mod.rs
index 3aa15f8d2d..0d32f20d3c 100644
--- a/nomt/src/beatree/mod.rs
+++ b/nomt/src/beatree/mod.rs
@@ -149,17 +149,16 @@ impl Tree {
         let leaves: Vec<PageNumber> = shared
             .bbn_index
             .branches()
-            .flat_map(|branch| {
-                (0..branch.n() as usize).map(|i| PageNumber(branch.node_pointer(i)))
-            })
+            .flat_map(|branch| (0..branch.n() as usize).map(|i| PageNumber(branch.node_pointer(i))))
             .take(leaf_cache.capacity())
             .collect();
         drop(shared);
 
-        let chunk = leaves.len().div_ceil(threads.max(1)).max(1);
+        // `chunks` panics on 0, so keep the size at least 1 even if there are no leaves.
+        let chunk_size = leaves.len().div_ceil(threads.max(1)).max(1);
+        let (leaf_cache, leaf_store) = (&leaf_cache, &leaf_store);
         std::thread::scope(|scope| {
-            for chunk in leaves.chunks(chunk) {
-                let (leaf_cache, leaf_store) = (&leaf_cache, &leaf_store);
+            for chunk in leaves.chunks(chunk_size) {
                 scope.spawn(move || {
                     for &pn in chunk {
                         let leaf = Arc::new(leaf::node::LeafNode {
diff --git a/nomt/src/options.rs b/nomt/src/options.rs
index bbbf0c2fa4..c736a00e83 100644
--- a/nomt/src/options.rs
+++ b/nomt/src/options.rs
@@ -29,6 +29,7 @@ pub struct Options {
     /// Whether to prepopulate the upper layers of the page cache on startup.
     /// This incurs some I/O on startup but leads to predictable worst-case performance.
     pub(crate) prepopulate_page_cache: bool,
+    /// Whether to fill the leaf cache with beatree leaves on startup.
     pub(crate) prepopulate_leaf_cache: bool,
     pub(crate) page_cache_upper_levels: usize,
 }
@@ -174,8 +175,8 @@ impl Options {
 
     /// Sets whether to fill the leaf cache with beatree leaves on startup, up to its capacity.
     ///
-    /// This reads up to [`Self::leaf_cache_size`] of leaves on startup, so that sessions after a
-    /// restart do not pay for cold leaf reads.
+    /// This reads up to [`Self::leaf_cache_size`] MiB of leaves on startup, so that sessions after
+    /// a restart do not pay for cold leaf reads.
     /// Default: false
     pub fn prepopulate_leaf_cache(&mut self, prepopulate: bool) {
         self.prepopulate_leaf_cache = prepopulate;

From 866b754f44eea20a34775baf5f50423b2a9ec9c1 Mon Sep 17 00:00:00 2001
From: Nikolai Golub <nikolay.v.golub@gmail.com>
Date: Wed, 21 Oct 2026 11:48:31 +0200
Subject: [PATCH 15/17] Parametrize concurrency

---
 nomt/src/lib.rs     | 14 ++++++++++----
 nomt/src/options.rs | 15 +++++++++++++++
 2 files changed, 25 insertions(+), 4 deletions(-)

diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs
index 1477b333e3..15eb4170c9 100644
--- a/nomt/src/lib.rs
+++ b/nomt/src/lib.rs
@@ -197,6 +197,7 @@ impl From<[u8; 32]> for Root {
 /// An instance of the Nearly-Optimal Merkle Trie Database.
 pub struct Nomt<T> {
     merkle_update_pool: UpdatePool,
+    actuals_concurrency: usize,
     /// The handle to the page cache.
     page_cache: PageCache,
     page_pool: PagePool,
@@ -237,6 +238,7 @@ impl<T: HashAlgorithm> Nomt<T> {
 
         Ok(Self {
             merkle_update_pool: UpdatePool::new(o.commit_concurrency, o.warm_up),
+            actuals_concurrency: o.actuals_concurrency,
             page_cache,
             page_pool,
             store,
@@ -328,6 +330,7 @@ impl<T: HashAlgorithm> Nomt<T> {
 
         Session {
             store,
+            actuals_concurrency: self.actuals_concurrency,
             merkle_updater: self.merkle_update_pool.begin::<T>(
                 self.page_cache.clone(),
                 self.page_pool.clone(),
@@ -477,6 +480,7 @@ impl SessionParams {
 /// correctness of replaying the same operations.
 pub struct Session<T> {
     store: Store,
+    actuals_concurrency: usize,
     merkle_updater: Updater,
     metrics: Metrics,
     rollback_delta: Option<rollback::ReverseDeltaBuilder>,
@@ -585,9 +589,10 @@ impl<T: HashAlgorithm> Session<T> {
             .take()
             .map(|delta_builder| delta_builder.finalize(&actuals));
 
-        let merkle_update_handle = self
-            .merkle_updater
-            .update_and_prove::<T>(compact_actuals::<T>(&actuals), self.witness_mode.0)?;
+        let merkle_update_handle = self.merkle_updater.update_and_prove::<T>(
+            compact_actuals::<T>(&actuals, self.actuals_concurrency),
+            self.witness_mode.0,
+        )?;
 
         let mut tx = self.store.new_value_tx();
         for (path, read_write) in actuals {
@@ -611,6 +616,7 @@ impl<T: HashAlgorithm> Session<T> {
 /// Hashes the values of `actuals` for the merkle update, splitting large inputs across scoped threads.
 fn compact_actuals<T: HashAlgorithm>(
     actuals: &[(KeyPath, KeyReadWrite)],
+    max_threads: usize,
 ) -> Vec<(KeyPath, crate::merkle::KeyReadWrite)> {
     const MIN_CHUNK: usize = 2048;
     let compact = |chunk: &[(KeyPath, KeyReadWrite)]| {
@@ -619,7 +625,7 @@ fn compact_actuals<T: HashAlgorithm>(
             .map(|(path, read_write)| (*path, read_write.to_compact::<T>()))
             .collect::<Vec<_>>()
     };
-    let threads = std::thread::available_parallelism().map_or(1, |n| n.get().min(16));
+    let threads = std::thread::available_parallelism().map_or(1, |n| n.get().min(max_threads));
     let chunk = actuals.len().div_ceil(threads).max(MIN_CHUNK);
     if actuals.len() <= chunk {
         return compact(actuals);
diff --git a/nomt/src/options.rs b/nomt/src/options.rs
index a44652b733..c0f9a1436e 100644
--- a/nomt/src/options.rs
+++ b/nomt/src/options.rs
@@ -7,6 +7,7 @@ pub struct Options {
     pub(crate) path: PathBuf,
     /// The number of commit workers. Values over 64 will be rounded down to 64.
     pub(crate) commit_concurrency: usize,
+    pub(crate) actuals_concurrency: usize,
     /// The number of io_uring instances, or I/O threads on non-Linux platforms.
     pub(crate) io_workers: usize,
     /// Enable or disable metrics collection.
@@ -40,6 +41,7 @@ impl Options {
         Self {
             path: PathBuf::from("nomt_db"),
             commit_concurrency: 1,
+            actuals_concurrency: 16,
             io_workers: 3,
             metrics: false,
             bitbox_num_pages: 64_000,
@@ -73,6 +75,19 @@ impl Options {
         self.commit_concurrency = commit_concurrency;
     }
 
+    /// Set the maximum number of threads hashing actuals per session.
+    ///
+    /// Limited by available CPU parallelism. Set to 1 to hash sequentially.
+    /// Independent of commit and I/O worker concurrency.
+    ///
+    /// Default: 16.
+    ///
+    /// Panics if `max_threads` is zero.
+    pub fn actuals_concurrency(&mut self, max_threads: usize) {
+        assert!(max_threads > 0);
+        self.actuals_concurrency = max_threads;
+    }
+
     /// Set metrics collection on or off.
     ///
     /// Default: off.

From fb5daaf77c6fdc3c61eddf57efe2cc1000fbf9d4 Mon Sep 17 00:00:00 2001
From: Nikolai Golub <nikolay.v.golub@gmail.com>
Date: Wed, 21 Oct 2026 11:57:58 +0200
Subject: [PATCH 16/17] smarter parallelism

---
 nomt/src/lib.rs | 75 +++++++++++++++++++++++++++++++++++++++++++------
 1 file changed, 67 insertions(+), 8 deletions(-)

diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs
index 15eb4170c9..e0b6a04852 100644
--- a/nomt/src/lib.rs
+++ b/nomt/src/lib.rs
@@ -618,27 +618,44 @@ fn compact_actuals<T: HashAlgorithm>(
     actuals: &[(KeyPath, KeyReadWrite)],
     max_threads: usize,
 ) -> Vec<(KeyPath, crate::merkle::KeyReadWrite)> {
-    const MIN_CHUNK: usize = 2048;
+    const MIN_HASHES_PER_THREAD: usize = 8192;
     let compact = |chunk: &[(KeyPath, KeyReadWrite)]| {
         chunk
             .iter()
             .map(|(path, read_write)| (*path, read_write.to_compact::<T>()))
             .collect::<Vec<_>>()
     };
-    let threads = std::thread::available_parallelism().map_or(1, |n| n.get().min(max_threads));
-    let chunk = actuals.len().div_ceil(threads).max(MIN_CHUNK);
-    if actuals.len() <= chunk {
+    if max_threads == 1 || actuals.len() < 2 * MIN_HASHES_PER_THREAD {
         return compact(actuals);
     }
+    let hashes = actuals
+        .iter()
+        .filter(|(_, read_write)| {
+            matches!(
+                read_write,
+                KeyReadWrite::Write(Some(_)) | KeyReadWrite::ReadThenWrite(_, Some(_))
+            )
+        })
+        .count();
+    let threads = max_threads.min(hashes / MIN_HASHES_PER_THREAD);
+    if threads < 2 {
+        return compact(actuals);
+    }
+    let threads = std::thread::available_parallelism().map_or(1, |n| n.get().min(threads));
+    if threads == 1 {
+        return compact(actuals);
+    }
+    let chunk = actuals.len().div_ceil(threads);
     std::thread::scope(|scope| {
         let handles: Vec<_> = actuals
             .chunks(chunk)
             .map(|chunk| scope.spawn(move || compact(chunk)))
             .collect();
-        handles
-            .into_iter()
-            .flat_map(|handle| handle.join().expect("compacting thread panicked"))
-            .collect()
+        let mut output = Vec::with_capacity(actuals.len());
+        for handle in handles {
+            output.extend(handle.join().expect("compacting thread panicked"));
+        }
+        output
     })
 }
 
@@ -1010,6 +1027,48 @@ pub fn validate_hashtable(options: &Options) -> anyhow::Result<HashTableUtilizat
 mod tests {
     use crate::hasher::Blake3Hasher;
 
+    #[test]
+    fn parallel_compaction_matches_sequential() {
+        use super::{compact_actuals, KeyReadWrite};
+        use crate::merkle::KeyReadWrite as Compact;
+
+        let actuals: Vec<_> = (0u64..131_073)
+            .map(|i| {
+                let mut path = [0; 32];
+                path[..8].copy_from_slice(&i.to_be_bytes());
+                let value = Some(i.to_le_bytes().to_vec());
+                let read_write = match i % 6 {
+                    0 => KeyReadWrite::Read(None),
+                    1 => KeyReadWrite::Read(value),
+                    2 => KeyReadWrite::Write(None),
+                    3 => KeyReadWrite::Write(value),
+                    4 => KeyReadWrite::ReadThenWrite(value, None),
+                    _ => KeyReadWrite::ReadThenWrite(None, value),
+                };
+                (path, read_write)
+            })
+            .collect();
+
+        for len in [0, 1, 16_383, 16_384, 49_151, 49_152, 49_153, 131_073] {
+            let sequential = compact_actuals::<Blake3Hasher>(&actuals[..len], 1);
+            for threads in [2, 16] {
+                let parallel = compact_actuals::<Blake3Hasher>(&actuals[..len], threads);
+                assert_eq!(parallel.len(), sequential.len());
+                for ((path, op), (expected_path, expected_op)) in parallel.iter().zip(&sequential) {
+                    assert_eq!(path, expected_path);
+                    match (op, expected_op) {
+                        (Compact::Read, Compact::Read) => {}
+                        (Compact::Write(value), Compact::Write(expected))
+                        | (Compact::ReadThenWrite(value), Compact::ReadThenWrite(expected)) => {
+                            assert_eq!(value, expected);
+                        }
+                        _ => panic!("operation changed: {op:?} != {expected_op:?}"),
+                    }
+                }
+            }
+        }
+    }
+
     #[test]
     fn session_is_sync() {
         fn is_sync<T: Sync>() {}

From 6e2eb4bb4edf7facdaa4f8f65edaa558c820d672 Mon Sep 17 00:00:00 2001
From: Nikolai Golub <nikolai@sovlabs.io>
Date: Wed, 21 Oct 2026 12:16:19 +0200
Subject: [PATCH 17/17] Update compact_actuals test for read value hashes in
 merkle KeyReadWrite

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 nomt/src/lib.rs | 12 +++++++++---
 1 file changed, 9 insertions(+), 3 deletions(-)

diff --git a/nomt/src/lib.rs b/nomt/src/lib.rs
index d3580b6887..6a81f379ff 100644
--- a/nomt/src/lib.rs
+++ b/nomt/src/lib.rs
@@ -1020,9 +1020,15 @@ mod tests {
                 for ((path, op), (expected_path, expected_op)) in parallel.iter().zip(&sequential) {
                     assert_eq!(path, expected_path);
                     match (op, expected_op) {
-                        (Compact::Read, Compact::Read) => {}
-                        (Compact::Write(value), Compact::Write(expected))
-                        | (Compact::ReadThenWrite(value), Compact::ReadThenWrite(expected)) => {
+                        (Compact::Read(value), Compact::Read(expected))
+                        | (Compact::Write(value), Compact::Write(expected)) => {
+                            assert_eq!(value, expected);
+                        }
+                        (
+                            Compact::ReadThenWrite(read, value),
+                            Compact::ReadThenWrite(expected_read, expected),
+                        ) => {
+                            assert_eq!(read, expected_read);
                             assert_eq!(value, expected);
                         }
                         _ => panic!("operation changed: {op:?} != {expected_op:?}"),