perf(btree): decide page-reuse eligibility at free time, not per allocation

src/btree/mod.rs

@@ -12,4 +12,4 @@ pub(crate) mod tree;
 
 // `pub` here is crate-scoped in effect: the `btree` module itself is
 // `pub(crate)`, so this does not escape the crate.
-pub use tree::BTree;
+pub use tree::{BTree, ConsumedPages, PageSource};

src/btree/tree/flush.rs

@@ -100,10 +100,13 @@ impl<V: Vfs> BTree<V> {
 
     /// Drain and return all `page_ids` that were freed during this tree's
     /// mutation session, plus any leaf pages scheduled for freeing at the
-    /// upcoming `flush`. After this call, both `self.freed` and
-    /// `self.scheduled_frees` are empty.
+    /// upcoming `flush`. After this call, `self.freed`, `self.freed_reusable`
+    /// and `self.scheduled_frees` are all empty — a page eligible for
+    /// in-session reuse that no allocation claimed is still a page this session
+    /// freed, and the deferred-free queue must hear about it.
     pub fn drain_freed(&mut self) -> Vec<u64> {
         let mut out = std::mem::take(&mut self.freed);
+        out.append(&mut self.freed_reusable);
         out.append(&mut self.scheduled_frees);
         out
     }

src/btree/tree/mod.rs

@@ -5,8 +5,10 @@ pub(crate) mod core;
 pub(crate) mod flush;
 pub(crate) mod maintenance;
 pub(crate) mod navigate;
+pub(crate) mod page_source;
 pub(crate) mod read;
 pub(crate) mod scan;
 pub(crate) mod write;
 
 pub use core::BTree;
+pub use page_source::{ConsumedPages, PageSource};

src/btree/tree/page_source.rs

@@ -0,0 +1,133 @@
+//! Where a write session's pages come from, supplied to a tree as one value.
+//!
+//! Reuse eligibility is decided when a page is freed, not when one is
+//! allocated (see [`BTree::free_page`](super::core::BTree::free_page)). That is
+//! only sound while every input to the decision is already in place at the
+//! first free and cannot move afterwards, so the inputs travel together in
+//! [`PageSource`] and are handed over once, at
+//! [`BTree::open_session`](super::core::BTree::open_session). There is no
+//! setter for any of them: a tree either has a session's page source for its
+//! whole life or has none and bump-allocates.
+//!
+//! [`ConsumedPages`] carries the other half of that guarantee. A page recorded
+//! there is eligible for in-session reuse, and pages already banked on the
+//! strength of that record cannot be un-banked — so the sink must not shrink
+//! while a session is running. Trees therefore hold a [`ConsumedHandle`], which
+//! can record and test but cannot empty; emptying is [`ConsumedPages::take`],
+//! reachable only from whoever owns the sink, at a transaction boundary.
+
+use std::sync::Arc;
+
+use rustc_hash::FxHashSet;
+
+/// Page ids the allocator drew from the shared free-page cache and reused this
+/// session. The commit path removes them from the durable free-list — they now
+/// hold live committed data — and [`BTree::free_page`] consults them: a page
+/// drawn from the cache is free as of the last durable header regardless of
+/// what this session wrote to it, so it may be recycled again in-session even
+/// from below the reuse threshold.
+///
+/// A set, not a list: membership is tested once per freed page, and the size is
+/// bounded by the free-list window (`WINDOW_PAGES × chain_capacity(page_size)`)
+/// rather than by anything small — a linear test would keep the per-free cost
+/// proportional to that window. Nothing reads insertion order or needs
+/// duplicates.
+///
+/// [`BTree::free_page`]: super::core::BTree::free_page
+#[derive(Debug, Default)]
+pub struct ConsumedPages(parking_lot::Mutex<FxHashSet<u64>>);
+
+impl ConsumedPages {
+    /// An empty sink, ready to be shared with a session's trees via
+    /// [`Self::handle`].
+    #[must_use]
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self::default())
+    }
+
+    /// A tree's view of this sink: record and test, never empty. Handing trees
+    /// this instead of the `Arc` is what makes the sink's monotonicity within a
+    /// session a property of the types rather than of caller discipline.
+    #[must_use]
+    pub fn handle(self: &Arc<Self>) -> ConsumedHandle {
+        ConsumedHandle(Arc::clone(self))
+    }
+
+    /// Empty the sink and return what it held.
+    ///
+    /// A transaction-boundary operation, and the only way to empty it. Calling
+    /// this while a session's trees are still allocating would strip the record
+    /// that made already-banked pages eligible, leaving them banked on a reason
+    /// that no longer exists — which is why no [`ConsumedHandle`] can reach it.
+    pub fn take(&self) -> FxHashSet<u64> {
+        std::mem::take(&mut *self.0.lock())
+    }
+}
+
+/// A tree's handle on a session's [`ConsumedPages`]. Clone-shared across the
+/// trees of one transaction.
+#[derive(Clone, Debug)]
+pub struct ConsumedHandle(Arc<ConsumedPages>);
+
+impl ConsumedHandle {
+    pub(crate) fn record(&self, page_id: u64) {
+        self.0.0.lock().insert(page_id);
+    }
+
+    pub(crate) fn contains(&self, page_id: u64) -> bool {
+        self.0.0.lock().contains(&page_id)
+    }
+}
+
+/// The pages a write session may allocate from, beyond bumping its own cursor.
+///
+/// Constructed once per session and cloned to each of its trees, so a tree can
+/// never observe a partly-wired source — a cache without the sink that records
+/// draws from it, or a threshold arriving after pages have already been
+/// classified against the default.
+#[derive(Clone, Debug)]
+pub struct PageSource {
+    /// Minimum `page_id` that may be recycled from within this session. Pages
+    /// below it existed before the session began: they are still referenced by
+    /// the last durable header (a copy-on-write free does not unreference them
+    /// on disk until the *next* header lands) and may also be pinned by reader
+    /// snapshots, so their bytes must survive until the header that frees them
+    /// is durable. They re-enter circulation through the deferred-free queue.
+    ///
+    /// Callers pass the session-start `next_page_id`, so only pages
+    /// bump-allocated within the session are recyclable immediately. Zero means
+    /// every freed page is (the commit-history tree, whose pages no reader
+    /// snapshot names).
+    pub(crate) reuse_threshold: u64,
+    /// Cross-commit pool of reusable page ids, shared across the main, catalog
+    /// and history trees of one `Db`. Loaded at `begin_write` with *only*
+    /// free-list pages below the reclamation floor — pages no live reader and
+    /// no retained-history root can observe — so drawing from it is safe
+    /// regardless of `reuse_threshold`.
+    ///
+    /// Draw-only: nothing may push into it. The commit deletes a recycled id
+    /// from the chain by finding it in the scanned window, so an id pushed in
+    /// from elsewhere would be handed out while the unscanned tail still named
+    /// it — the same page id given to two live structures.
+    pub(crate) cache: Arc<parking_lot::Mutex<Vec<u64>>>,
+    /// Where draws from `cache` are recorded.
+    pub(crate) consumed: ConsumedHandle,
+}
+
+impl PageSource {
+    /// Bind a session's allocation state to one value. `reuse_threshold` is the
+    /// session-start `next_page_id` for trees a reader may hold a snapshot of,
+    /// and zero for those none can.
+    #[must_use]
+    pub fn new(
+        reuse_threshold: u64,
+        cache: Arc<parking_lot::Mutex<Vec<u64>>>,
+        consumed: &Arc<ConsumedPages>,
+    ) -> Self {
+        Self {
+            reuse_threshold,
+            cache,
+            consumed: consumed.handle(),
+        }
+    }
+}

src/txn/db/catalog.rs

@@ -232,13 +232,6 @@ impl<V: Vfs + Clone> Db<V> {
             readers.iter().map(|r| r.commit_id.0).min()
         };
 
-        let mut hist_tree = BTree::open(
-            self.pager.clone(),
-            self.realm_id,
-            state.commit_history_root_page_id,
-            state.next_page_id,
-            self.page_size,
-        );
         // The commit-history tree is not part of any reader's pinned snapshot
         // (readers track the data and catalog roots, never the history root), so
         // every page its copy-on-write/prune frees is immediately reusable
@@ -250,9 +243,18 @@ impl<V: Vfs + Clone> Db<V> {
         // pushes into it. Its own frees leave through `drain_freed` into the
         // commit's entry set instead, so no page reaches an allocator without
         // the window entry that names it having been located first.
-        hist_tree.set_reuse_threshold(0);
-        hist_tree.set_free_page_cache(self.free_page_cache.clone());
-        hist_tree.set_free_page_consumed(self.free_page_consumed.clone());
+        let mut hist_tree = BTree::open_session(
+            self.pager.clone(),
+            self.realm_id,
+            state.commit_history_root_page_id,
+            state.next_page_id,
+            self.page_size,
+            crate::btree::PageSource::new(
+                0,
+                self.free_page_cache.clone(),
+                &self.free_page_consumed,
+            ),
+        );
 
         // Insert the new entry.
         let key = new_commit_id.to_be_bytes().to_vec();

src/txn/db/core.rs

@@ -231,10 +231,15 @@ pub struct Db<V: Vfs + Clone> {
     /// history) draw from the same pool. Cleared by `compact_now`'s full repack,
     /// which relocates pages and invalidates every cached id.
     pub(crate) free_page_cache: Arc<parking_lot::Mutex<Vec<u64>>>,
-    /// Per-txn sink (cleared at `begin_write`) recording page ids the allocator
+    /// Per-txn sink (emptied at `begin_write`) recording page ids the allocator
     /// drew from `free_page_cache`. The commit path removes them from the
     /// durable free-list — they now hold live committed data.
-    pub(crate) free_page_consumed: Arc<parking_lot::Mutex<Vec<u64>>>,
+    ///
+    /// The trees see it through a handle that cannot empty it: a page banked
+    /// for in-session reuse on the strength of a record here must not have that
+    /// record withdrawn under it, so emptying belongs to the transaction
+    /// boundaries alone.
+    pub(crate) free_page_consumed: Arc<crate::btree::ConsumedPages>,
     #[cfg(test)]
     pub(crate) visibility_test_hook: parking_lot::Mutex<Option<Arc<VisibilityTestHook>>>,
     #[cfg(test)]

src/txn/db/open/create.rs

@@ -130,7 +130,7 @@ impl<V: Vfs + Clone> Db<V> {
             })),
             poisoned_commit: parking_lot::Mutex::new(None),
             free_page_cache: Arc::new(parking_lot::Mutex::new(Vec::new())),
-            free_page_consumed: Arc::new(parking_lot::Mutex::new(Vec::new())),
+            free_page_consumed: crate::btree::ConsumedPages::new(),
             #[cfg(test)]
             visibility_test_hook: parking_lot::Mutex::new(None),
             #[cfg(test)]

src/txn/db/open/existing.rs

@@ -321,7 +321,7 @@ impl<V: Vfs + Clone> Db<V> {
             })),
             poisoned_commit: parking_lot::Mutex::new(None),
             free_page_cache: Arc::new(parking_lot::Mutex::new(Vec::new())),
-            free_page_consumed: Arc::new(parking_lot::Mutex::new(Vec::new())),
+            free_page_consumed: crate::btree::ConsumedPages::new(),
             #[cfg(test)]
             visibility_test_hook: parking_lot::Mutex::new(None),
             #[cfg(test)]

src/txn/write/commit.rs

@@ -144,9 +144,9 @@ impl<V: Vfs + Clone> WriteTxn<'_, V> {
             .collect();
         // Pages the allocator reused from the cache this txn — they were free
         // before, now hold live committed data, so they leave the free-list.
-        let consumed: HashSet<u64> = std::mem::take(&mut *self.db.free_page_consumed.lock())
-            .into_iter()
-            .collect();
+        // Emptied here rather than mid-session: the trees have all drained, so
+        // nothing can still be banking pages against it.
+        let consumed = self.db.free_page_consumed.take();
 
         // The scanned window after this commit: its entries minus the ones
         // reused, kept with their original freeing-commit tag. Every id in

src/txn/write/txn.rs

@@ -159,28 +159,34 @@ impl<'db, V: Vfs + Clone> WriteTxn<'db, V> {
                 }
             }
         }
-        db.free_page_consumed.lock().clear();
+        // Emptied here, at the boundary, and not again until commit: the trees
+        // decide reuse eligibility against this sink as they free, so it must
+        // only grow for as long as the session runs.
+        db.free_page_consumed.take();
 
-        let mut btree = BTree::open(
+        // One source for every tree in the session, so all three draw from the
+        // same cache and record into the same sink.
+        let source = crate::btree::PageSource::new(
+            reuse_threshold,
+            db.free_page_cache.clone(),
+            &db.free_page_consumed,
+        );
+        let btree = BTree::open_session(
             db.pager.clone(),
             db.realm_id,
             guard.root_page_id,
             guard.next_page_id,
             db.page_size,
+            source.clone(),
         );
-        btree.set_reuse_threshold(reuse_threshold);
-        btree.set_free_page_cache(db.free_page_cache.clone());
-        btree.set_free_page_consumed(db.free_page_consumed.clone());
-        let mut catalog_tree = BTree::open(
+        let catalog_tree = BTree::open_session(
             db.pager.clone(),
             db.realm_id,
             guard.catalog_root_page_id,
             guard.next_page_id,
             db.page_size,
+            source,
         );
-        catalog_tree.set_reuse_threshold(reuse_threshold);
-        catalog_tree.set_free_page_cache(db.free_page_cache.clone());
-        catalog_tree.set_free_page_consumed(db.free_page_consumed.clone());
         // Assign a txn_seq starting from 1: fetch_add returns the old value (0
         // for the first call), so we add 1 to produce 1-based ids.
         let txn_seq = db.txn_seq.fetch_add(1, Ordering::Relaxed) + 1;