Merge remote-tracking branch 'origin/main'

# Conflicts:
#	Cargo.toml
#	src/txn/db/mod.rs

Author: farhan-syah

.github/workflows/test.yml

@@ -266,4 +266,4 @@ jobs:
         with:
           tool: cargo-deny
       - name: Run cargo-deny
-        run: cargo deny check --exclude-dev
+        run: cargo deny --exclude-dev check

Cargo.lock

@@ -23,10 +23,14 @@ pub struct BTree<V: Vfs> {
     pub(super) freed: Vec<u64>,
     pub(super) page_size: usize,
     /// Minimum `page_id` that may be recycled from `freed` within this session.
-    /// Pages below this threshold were live in prior snapshots and may still be
-    /// accessed by pinned readers; they must not be overwritten until the
-    /// deferred-free queue allows it. Set to `next_page_id` when any reader is
-    /// tracked; set to 0 when no readers are pinned (safe to recycle freely).
+    /// Pages below this threshold 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. They must not be overwritten
+    /// in-session; they re-enter circulation through the deferred-free queue
+    /// once the free-list naming them is durable. Callers set this to the
+    /// session-start `next_page_id`, so only pages bump-allocated within the
+    /// session are recyclable immediately.
     pub(super) reuse_threshold: u64,
     /// Leaves modified during this write session but not yet promoted via
     /// `CoW` to a fresh page. Keyed by the leaf's current `page_id` as referenced
@@ -108,8 +112,9 @@ impl<V: Vfs> BTree<V> {
 
     /// Set the reuse threshold. Any freed page with `page_id < threshold` will
     /// not be recycled within this session; it goes to `self.freed` for later
-    /// deferred-queue promotion. Call with `next_page_id` when tracked readers
-    /// are present; call with `0` when no readers are pinned.
+    /// deferred-queue promotion. Call with the session-start `next_page_id`:
+    /// pages below it are still referenced by the last durable header and must
+    /// keep their on-disk bytes until the header that frees them is durable.
     pub fn set_reuse_threshold(&mut self, threshold: u64) {
         self.reuse_threshold = threshold;
     }
@@ -153,16 +158,28 @@ impl<V: Vfs> BTree<V> {
         // the reuse threshold: a page below it may still be live in a pinned
         // reader's snapshot, so it can't be reused until the durable free-list
         // clears it (it leaves via `drain_freed` at commit instead).
+        //
+        // Exception: a below-threshold page originally drawn from the shared
+        // cache this session (recorded in `free_page_consumed`) is free per
+        // the last durable header regardless of what this session wrote to it
+        // since, so recycling it again in-session is crash-safe — a failed or
+        // torn commit leaves the durable header referencing none of its
+        // content. Without this, every cache-drawn page freed by a later
+        // in-session split is burned for the rest of the txn and allocation
+        // falls through to bump growth.
         if self.reuse_threshold == 0 {
             if let Some(id) = self.freed.pop() {
                 return id;
             }
-        } else if let Some(pos) = self
-            .freed
-            .iter()
-            .rposition(|&id| id >= self.reuse_threshold)
-        {
-            return self.freed.remove(pos);
+        } else {
+            let consumed = self.free_page_consumed.as_ref().map(|c| c.lock());
+            let pos = self.freed.iter().rposition(|&id| {
+                id >= self.reuse_threshold || consumed.as_ref().is_some_and(|c| c.contains(&id))
+            });
+            drop(consumed);
+            if let Some(pos) = pos {
+                return self.freed.remove(pos);
+            }
         }
         // Then draw from the shared cross-commit cache. It is loaded at txn
         // begin with *only* free-list pages below the reclamation floor — pages

src/btree/tree/core.rs

@@ -50,6 +50,15 @@ impl<V: Vfs> BTree<V> {
                 };
                 self.write_internal(new_root_page, &internal).await?;
                 self.root_page_id = new_root_page;
+                // The tree grew a level: every cached dirty-leaf parent path
+                // still starts at the old root (or its split halves). Prepend
+                // the new root so flush's spine walk reaches it — otherwise
+                // the new root's child pointers are never rewritten when those
+                // halves are CoW'd at flush, leaving the durable root pointing
+                // at freed pages.
+                for path in self.dirty_parent_paths.values_mut() {
+                    path.insert(0, new_root_page);
+                }
                 return Ok(());
             }
 
@@ -77,12 +86,18 @@ impl<V: Vfs> BTree<V> {
                 self.free_page(internal_page);
                 // An internal page that other dirty leaves' parent paths
                 // reference has been replaced and freed; remap them. The
-                // split case promotes the old internal: keys ≤ promoted live
-                // on `new_internal_page`, the rest on `right_internal_page`.
-                // For path-remapping (used only to find ancestors to CoW)
-                // either replacement is acceptable, so substitute with the
-                // left replacement.
-                self.remap_dirty_parent_paths(internal_page, new_internal_page);
+                // split distributes the old internal's children across two
+                // pages, so each path must be remapped to the half that
+                // actually holds its next hop — substituting the wrong half
+                // means flush's spine walk never visits the true parent, its
+                // stale child pointer survives, and the durable tree ends up
+                // referencing a freed leaf page.
+                self.remap_dirty_parent_paths_split(
+                    internal_page,
+                    new_internal_page,
+                    right_internal_page,
+                    &right,
+                );
                 child_old = internal_page;
                 left_replacement = new_internal_page;
                 right_to_insert = right_internal_page;
@@ -122,8 +137,44 @@ impl<V: Vfs> BTree<V> {
         }
     }
 
+    /// Substitute every occurrence of `old` in cached dirty-leaf parent paths
+    /// with whichever split half actually contains the path's next hop as a
+    /// child. Used after [`propagate_split_up`](Self::propagate_split_up)
+    /// splits an internal page that other dirty leaves' paths reference: the
+    /// old page's children are distributed across `left_new` and `right_new`,
+    /// and flush's spine walk only rewrites child pointers in internals the
+    /// paths name — so each path must name the half its leaf descends from.
+    /// The next hop is the path element after `old`, or the dirty leaf's own
+    /// `page_id` when `old` is the path's last element.
+    fn remap_dirty_parent_paths_split(
+        &mut self,
+        old: u64,
+        left_new: u64,
+        right_new: u64,
+        right: &Internal,
+    ) {
+        if self.dirty_parent_paths.is_empty() {
+            return;
+        }
+        let right_children: std::collections::HashSet<u64> = std::iter::once(right.leftmost_child)
+            .chain(right.entries.iter().map(|e| e.right_child))
+            .collect();
+        for (leaf_id, path) in &mut self.dirty_parent_paths {
+            for i in 0..path.len() {
+                if path[i] == old {
+                    let next_hop = path.get(i + 1).copied().unwrap_or(*leaf_id);
+                    path[i] = if right_children.contains(&next_hop) {
+                        right_new
+                    } else {
+                        left_new
+                    };
+                }
+            }
+        }
+    }
+
     /// Substitute every occurrence of `old` with `new` in cached dirty-leaf
-    /// parent paths. Used after a split or a spine `CoW` in
+    /// parent paths. Used after a spine `CoW` in
     /// [`propagate_split_up`](Self::propagate_split_up) replaces an internal
     /// page that other dirty leaves' paths reference.
     fn remap_dirty_parent_paths(&mut self, old: u64, new: u64) {

src/btree/tree/navigate.rs

@@ -604,6 +604,7 @@ impl<V: Vfs> Pager<V> {
     /// AAD is constructed from on-disk header bytes, not from `Pager.active_epoch`
     /// or the configured cipher. This makes mixed-epoch and mixed-cipher coexistence
     /// work correctly without global invariants.
+    #[allow(clippy::too_many_lines)]
     async fn read_page(
         &self,
         file: FileKey,
@@ -619,6 +620,14 @@ impl<V: Vfs> Pager<V> {
                 if page.realm_id_bytes != Some(realm_id.0) {
                     return Err(PagedbError::ChecksumFailure);
                 }
+                // Enforce the same kind binding a cold read's AAD enforces.
+                // Without this, a stale pointer reading a recycled page under
+                // the wrong kind succeeds while the page is warm and only
+                // starts failing after eviction — hiding structural damage
+                // until long after the write that caused it.
+                if page.kind_byte != 0 && page.kind_byte != expected_kind.as_byte() {
+                    return Err(PagedbError::ChecksumFailure);
+                }
                 self.inner.record_hit(file);
                 cache.pin((file, page_id));
                 return Ok(PageGuard {
@@ -720,9 +729,17 @@ impl<V: Vfs> Pager<V> {
                 Err(e) => return Err(e),
             }
         }
+        tracing::error!(
+            page_id,
+            ?file,
+            expected_kind = ?expected_kind,
+            realm = ?realm_id.0,
+            "page AEAD/MAC verification failed on read"
+        );
         Err(last_err.unwrap_or(PagedbError::ChecksumFailure))
     }
 
+    #[allow(clippy::too_many_lines)]
     async fn flush_file(
         &self,
         file: FileKey,
@@ -744,8 +761,11 @@ impl<V: Vfs> Pager<V> {
         let flush_epoch = self.active_epoch.load(AtomOrd::SeqCst);
 
         // Serial gather: snapshot each dirty page's plaintext + kind under the
-        // cache lock. Cheap memcpy; no AEAD work happens here.
+        // cache lock. Cheap memcpy; no AEAD work happens here. The gathered
+        // `Arc<Page>` is retained per pid so the dirty-clear below can detect
+        // pages replaced by a concurrent writer during the (slow) seal+write.
         let mut prepared: Vec<(u64, PageKind, Vec<u8>)> = Vec::with_capacity(dirty_ids.len());
+        let mut gathered: Vec<(u64, Arc<Page>)> = Vec::with_capacity(dirty_ids.len());
         for pid in &dirty_ids {
             let page = self
                 .inner
@@ -755,6 +775,7 @@ impl<V: Vfs> Pager<V> {
                 .ok_or_else(|| {
                     PagedbError::Io(std::io::Error::other("dirty page missing from cache"))
                 })?;
+            gathered.push((*pid, page.clone()));
             let kind = if page.kind_byte != 0 {
                 PageKind::from_byte(page.kind_byte)?
             } else {
@@ -765,6 +786,7 @@ impl<V: Vfs> Pager<V> {
             let mut wire = vec![0u8; page_size];
             let plain = body(&page.bytes);
             wire[HEADER_LEN..HEADER_LEN + plain.len()].copy_from_slice(plain);
+            tracing::trace!(page_id = *pid, ?kind, ?file, "flush: writing page");
             prepared.push((*pid, kind, wire));
         }
 
@@ -843,10 +865,25 @@ impl<V: Vfs> Pager<V> {
             f.sync().await?;
         }
 
-        // Clear dirty flags.
+        // Clear dirty flags — but ONLY for pages still holding the exact
+        // `Arc<Page>` we gathered. A concurrent writer replaces the Arc on
+        // every write; unconditionally clearing here would wipe its dirty
+        // flag while its content never reached disk (lost update → stale
+        // page on next cold read → AEAD/kind mismatch).
+        // A replaced page keeps its flag and flushes on the next cycle.
         let mut cache = self.inner.cache_for_key(file).lock();
-        for pid in dirty_ids {
-            cache.clear_dirty((file, pid));
+        for (pid, snapshot) in gathered {
+            match cache.get((file, pid)) {
+                Some(current) if Arc::ptr_eq(&current, &snapshot) => {
+                    cache.clear_dirty((file, pid));
+                }
+                _ => {
+                    tracing::debug!(
+                        page_id = pid,
+                        "page re-dirtied during flush; keeping dirty for next cycle"
+                    );
+                }
+            }
         }
         Ok(())
     }