fix(btree): report precise chain-cycle and sibling-mismatch corruption

Replace the generic HeaderUnverifiable/IO stand-ins used for btree
descent cycles, leaf sibling-link mismatches, and non-monotonic
bulk_load input with dedicated PagedbError variants that name the
walk (btree_descent, leaf_siblings, free_list) and the pages
involved, so fsck and callers can act on what actually went wrong.

Also stop discarding put_append's cached rightmost path before its
monotonic check has passed, so a rejected append leaves the session
exactly as it found it, and log (rather than silently drop) an
overflow chain that can't be reclaimed on a failed put.

Author: farhan-syah

src/btree/tree/bulk.rs

@@ -37,10 +37,7 @@ impl<V: Vfs> BTree<V> {
 
         for pair in pairs.windows(2) {
             if pair[0].0 >= pair[1].0 {
-                return Err(PagedbError::Io(std::io::Error::new(
-                    std::io::ErrorKind::InvalidInput,
-                    "bulk_load input keys must be strictly increasing",
-                )));
+                return Err(PagedbError::BulkLoadNotMonotonic);
             }
         }
         for (key, value) in &pairs {

src/btree/tree/core.rs

@@ -3,7 +3,7 @@
 use std::collections::{BTreeSet, HashMap};
 use std::sync::Arc;
 
-use crate::errors::{CorruptionDetail, PagedbError};
+use crate::errors::PagedbError;
 use crate::pager::format::page_kind::PageKind;
 use crate::pager::{PageGuard, Pager};
 use crate::vfs::Vfs;
@@ -14,18 +14,29 @@ use crate::btree::leaf::Leaf;
 use crate::btree::node::{NodeKind, body_capacity, read_header};
 use crate::btree::overflow;
 
+/// Encoded size a value of `value_len` bytes will occupy in a leaf record,
+/// computed before the value is built.
+///
+/// The preflight counterpart to [`LeafValue::encoded_size`], and it must agree
+/// with it — both sides read the same constant rather than restating the
+/// layout.
 fn stored_leaf_value_encoded_size(value_len: usize, page_size: usize) -> Result<usize> {
     if value_len > overflow::inline_value_threshold(page_size) {
-        Ok(2 + 8 + 8)
+        Ok(crate::btree::leaf::OVERFLOW_REF_ENCODED_SIZE)
     } else {
         2usize
             .checked_add(value_len)
             .ok_or(PagedbError::PayloadTooLarge)
     }
 }
 
-pub(super) fn malformed_btree_topology() -> PagedbError {
-    PagedbError::corruption(CorruptionDetail::HeaderUnverifiable)
+/// A descent reached a page twice, or followed a zero child pointer.
+///
+/// Both mean the pointer graph has no terminator, which no honest writer can
+/// produce. Named for the walk rather than reported as a generic header
+/// failure, so `fsck` can say which page repeated and in which structure.
+pub(super) fn malformed_btree_topology(structure: &'static str, page_id: u64) -> PagedbError {
+    PagedbError::page_chain_cycle(structure, page_id)
 }
 
 // Keep common shallow descents stack-only and small; deeper trees spill into
@@ -35,22 +46,25 @@ const INLINE_SEEN_PAGE_IDS: usize = 4;
 /// Duplicate-page detector for B-tree descents. Shallow trees stay on the
 /// inline array; deeper trees preserve exact detection through the overflow set.
 pub(super) struct SeenPageIds {
+    /// Which walk this is, so a rejection can name the structure it was in.
+    structure: &'static str,
     inline: [u64; INLINE_SEEN_PAGE_IDS],
     len: usize,
     overflow: Option<BTreeSet<u64>>,
 }
 
 impl SeenPageIds {
-    pub(super) fn new() -> Self {
+    pub(super) fn new(structure: &'static str) -> Self {
         Self {
+            structure,
             inline: [0; INLINE_SEEN_PAGE_IDS],
             len: 0,
             overflow: None,
         }
     }
 
-    pub(super) fn from_existing(path: &[u64]) -> Result<Self> {
-        let mut seen = Self::new();
+    pub(super) fn from_existing(structure: &'static str, path: &[u64]) -> Result<Self> {
+        let mut seen = Self::new(structure);
         for &page_id in path {
             seen.insert(page_id)?;
         }
@@ -59,7 +73,7 @@ impl SeenPageIds {
 
     pub(super) fn insert(&mut self, page_id: u64) -> Result<()> {
         if page_id == 0 || !self.insert_inner(page_id) {
-            return Err(malformed_btree_topology());
+            return Err(malformed_btree_topology(self.structure, page_id));
         }
         Ok(())
     }

src/btree/tree/navigate.rs

@@ -195,7 +195,7 @@ impl<V: Vfs> BTree<V> {
     pub(super) async fn path_to_leaf_for_key(&self, key: &[u8]) -> Result<Vec<u64>> {
         let mut path = Vec::new();
         let mut page_id = self.root_page_id;
-        let mut seen = SeenPageIds::new();
+        let mut seen = SeenPageIds::new("btree_descent");
         loop {
             seen.insert(page_id)?;
             path.push(page_id);
@@ -226,7 +226,7 @@ impl<V: Vfs> BTree<V> {
             // Root is a leaf; no next leaf.
             return Ok(None);
         }
-        let mut seen = SeenPageIds::from_existing(path)?;
+        let mut seen = SeenPageIds::from_existing("btree_descent", path)?;
         let mut child = path[path.len() - 1];
         for i in (0..path.len() - 1).rev() {
             let (guard, _kind) = self.read_node_guard(path[i]).await?;
@@ -264,7 +264,7 @@ impl<V: Vfs> BTree<V> {
     pub(super) async fn path_to_rightmost_leaf(&self) -> Result<Vec<u64>> {
         let mut path = Vec::new();
         let mut page_id = self.root_page_id;
-        let mut seen = SeenPageIds::new();
+        let mut seen = SeenPageIds::new("btree_descent");
         loop {
             seen.insert(page_id)?;
             path.push(page_id);

src/btree/tree/read.rs

@@ -93,7 +93,7 @@ impl<V: Vfs> BTree<V> {
             NodeKind::Internal => InternalAccessor::new(start_guard.body_ref())?.child_for(key),
         };
         // Descend from the first child onward. Subsequent guards are owned.
-        let mut seen = SeenPageIds::new();
+        let mut seen = SeenPageIds::new("btree_descent");
         seen.insert(start_page_id)?;
         let mut page_id = next_page_id;
         loop {

src/btree/tree/write.rs

@@ -3,7 +3,7 @@
 use std::sync::Arc;
 
 use crate::Result;
-use crate::errors::{CorruptionDetail, PagedbError};
+use crate::errors::PagedbError;
 use crate::vfs::Vfs;
 
 use crate::btree::leaf::{Leaf, LeafValue};
@@ -49,15 +49,28 @@ impl<V: Vfs> BTree<V> {
         }
     }
 
+    /// Return the pages of a chain that was written for a value the caller
+    /// never got to commit.
+    ///
+    /// Best-effort by design: this runs on an error path, and a failure here
+    /// must not replace the error the caller actually needs to see. It is
+    /// logged rather than swallowed, because the cost of giving up is a page
+    /// leak that nothing else will notice.
     async fn discard_uncommitted_value(&mut self, value: Option<LeafValue>) {
-        if let Some(LeafValue::Overflow { root_page_id, .. }) = value {
-            if let Ok(page_ids) =
-                overflow::collect_chain(&self.pager, self.realm_id, root_page_id).await
-            {
+        let Some(LeafValue::Overflow { root_page_id, .. }) = value else {
+            return;
+        };
+        match overflow::collect_chain(&self.pager, self.realm_id, root_page_id).await {
+            Ok(page_ids) => {
                 for page_id in page_ids {
                     self.free_page(page_id);
                 }
             }
+            Err(error) => tracing::warn!(
+                root_page_id,
+                error = %error,
+                "could not reclaim the overflow chain of an uncommitted value; its pages are leaked"
+            ),
         }
     }
 
@@ -236,7 +249,11 @@ impl<V: Vfs> BTree<V> {
     /// # Errors
     ///
     /// Returns [`PagedbError::AppendNotMonotonic`] if `key` is not strictly
-    /// greater than the previously-appended key in this `BTree` session.
+    /// greater than the previously-appended key in this `BTree` session, or —
+    /// on the first append of a session, when there is no such key yet — not
+    /// strictly greater than the largest key already in the tree. Both are the
+    /// same invariant: the cached rightmost path is only meaningful while every
+    /// append lands to the right of everything before it.
     /// Intended for op-logs, time-series indexes, FTS posting-list builds —
     /// any workload where the embedder can guarantee monotonic-key insert.
     pub async fn put_append(&mut self, key: &[u8], value: &[u8]) -> Result<()> {
@@ -262,11 +279,12 @@ impl<V: Vfs> BTree<V> {
         }
 
         // Fast path: cached rightmost path is valid → skip descent.
-        let path = if let Some(cached) = self.append_cached_path.take() {
-            cached
-        } else {
-            self.path_to_rightmost_leaf().await?
+        let path = match &self.append_cached_path {
+            Some(cached) => cached.clone(),
+            None => self.path_to_rightmost_leaf().await?,
         };
+        // Checked before the cached path is consumed: a rejected append must
+        // leave the session exactly as it found it, cache included.
         if self.append_last_key.is_none()
             && self
                 .max_key_at_path(&path)
@@ -275,6 +293,7 @@ impl<V: Vfs> BTree<V> {
         {
             return Err(PagedbError::AppendNotMonotonic);
         }
+        self.append_cached_path = None;
         let leaf_value = self.leaf_value_for(value).await?;
         let path_for_retry = path.clone();
         let split = self.put_at_path(path, key, leaf_value).await?;
@@ -394,7 +413,7 @@ impl<V: Vfs> BTree<V> {
         // their sibling links remain zero until flush.
         let mut path = self.path_to_leaf_for_key(start).await?;
         let mut keys_to_delete: Vec<Vec<u8>> = Vec::new();
-        let mut seen_leaves = SeenPageIds::new();
+        let mut seen_leaves = SeenPageIds::new("leaf_siblings");
         'outer: loop {
             let leaf_page_id = *path.last().expect("non-empty path");
             seen_leaves.insert(leaf_page_id)?;
@@ -416,9 +435,11 @@ impl<V: Vfs> BTree<V> {
                 {
                     path = parent_next;
                 }
-                _ => {
-                    return Err(PagedbError::corruption(
-                        CorruptionDetail::HeaderUnverifiable,
+                (right_sibling, parent_next) => {
+                    return Err(PagedbError::leaf_sibling_mismatch(
+                        leaf_page_id,
+                        right_sibling,
+                        parent_next.and_then(|path| path.last().copied()),
                     ));
                 }
             }

src/errors.rs

@@ -127,6 +127,16 @@ pub enum PagedbError {
     #[error("put_append called with non-monotonic key")]
     AppendNotMonotonic,
 
+    /// `BTree::bulk_load` was handed input whose keys are not strictly
+    /// increasing — descending, or repeating the same key.
+    ///
+    /// The sibling of [`Self::AppendNotMonotonic`]: same invariant, different
+    /// entry point. Bulk load builds the tree bottom-up from the input order
+    /// itself, so unsorted input does not merely misplace a record — it would
+    /// produce a tree whose separators do not describe its leaves.
+    #[error("bulk_load keys must be strictly increasing")]
+    BulkLoadNotMonotonic,
+
     /// The deferred-free backlog exceeds the configured threshold and
     /// active reader pins prevent draining it.
     #[non_exhaustive]
@@ -306,6 +316,33 @@ pub enum CorruptionDetail {
     /// has no terminator. Distinct from a truncated chain: the links
     /// authenticate, they just form a loop.
     OverflowChainCycle { root_page_id: u64, page_id: u64 },
+    /// A linked page structure revisited a page it had already walked, so the
+    /// walk has no terminator.
+    ///
+    /// The overflow-chain form is [`Self::OverflowChainCycle`], which can also
+    /// name the chain's root. This is the general case — a B+ tree root-to-leaf
+    /// descent, a leaf sibling walk, or the durable free-list chain — where
+    /// `structure` says which walk found the loop and `page_id` is the page it
+    /// reached twice. Every link authenticates; they simply form a loop, which
+    /// no honest writer can produce.
+    PageChainCycle {
+        structure: &'static str,
+        page_id: u64,
+    },
+    /// A leaf's persisted right-sibling link disagrees with the leaf its parent
+    /// path says comes next.
+    ///
+    /// Both encode "the next leaf", so both must name it. A disagreement means
+    /// one of them outlived the page it points at — a sibling link left behind
+    /// by a page that was freed and handed to another node, or a parent whose
+    /// child pointer was rewritten without its leaves. `parent_next` is `None`
+    /// when the parent path has no successor at all yet the leaf still claims
+    /// one.
+    LeafSiblingMismatch {
+        leaf_page_id: u64,
+        right_sibling: u64,
+        parent_next: Option<u64>,
+    },
     /// One physical page was reached twice in a single traversal under two
     /// incompatible page kinds.
     ///
@@ -426,6 +463,29 @@ impl PagedbError {
         })
     }
 
+    /// Canonical constructor for a cyclic linked page structure. `structure`
+    /// names the walk that found the loop — `"btree_descent"`,
+    /// `"leaf_siblings"`, `"free_list"`.
+    #[must_use]
+    pub const fn page_chain_cycle(structure: &'static str, page_id: u64) -> Self {
+        Self::Corruption(CorruptionDetail::PageChainCycle { structure, page_id })
+    }
+
+    /// Canonical constructor for a leaf whose sibling link and parent path
+    /// disagree about which leaf comes next.
+    #[must_use]
+    pub const fn leaf_sibling_mismatch(
+        leaf_page_id: u64,
+        right_sibling: u64,
+        parent_next: Option<u64>,
+    ) -> Self {
+        Self::Corruption(CorruptionDetail::LeafSiblingMismatch {
+            leaf_page_id,
+            right_sibling,
+            parent_next,
+        })
+    }
+
     /// Canonical constructor for a cyclic overflow chain.
     #[must_use]
     pub const fn overflow_chain_cycle(root_page_id: u64, page_id: u64) -> Self {

src/pager/freelist.rs

@@ -50,11 +50,7 @@ pub async fn read_chain<V: Vfs + Clone>(
     let mut page = head;
     while page != 0 {
         if !seen.insert(page) {
-            return Err(PagedbError::corruption(
-                crate::errors::CorruptionDetail::CatalogRowInvalid {
-                    field: "freelist chain contains a cycle",
-                },
-            ));
+            return Err(PagedbError::page_chain_cycle("free_list", page));
         }
         let guard = pager.read_main_page(page, realm_id, PageKind::Free).await?;
         let body = guard.body_ref();
@@ -238,6 +234,15 @@ mod tests {
             .await
             .expect("cycle detection should return before timeout")
             .expect_err("free-list cycles must be corruption");
-        assert!(matches!(error, PagedbError::Corruption(_)));
+        assert!(
+            matches!(
+                error,
+                PagedbError::Corruption(crate::errors::CorruptionDetail::PageChainCycle {
+                    structure: "free_list",
+                    page_id: 10,
+                })
+            ),
+            "expected a free-list PageChainCycle naming page 10, got {error:?}"
+        );
     }
 }