fix(btree): detect leaf-revisit cycles during forward scans

The per-descent guard inside next_leaf_after is rebuilt on every call and
only proves a single step doesn't revisit a page. An internal node whose
leftmost_child and a later entry both point at the same child answers
"after A comes B" and "after B comes A", so a scan built from individually
acyclic steps can alternate between two leaves forever. Track visited
leaves for the lifetime of the whole scan instead of per descent.

Author: farhan-syah

src/btree/tree/scan.rs

@@ -3,7 +3,27 @@
 use crate::Result;
 use crate::vfs::Vfs;
 
-use super::core::BTree;
+use super::core::{BTree, SeenPageIds};
+
+/// Duplicate-leaf detector spanning one whole forward scan.
+///
+/// Every scan below is a loop over `next_leaf_after`, and that function builds
+/// a fresh descent guard per call, seeded only from the path it was handed. So
+/// it can prove one *step* does not revisit a page, and cannot see the scan as
+/// a whole revisiting a leaf it already yielded.
+///
+/// That gap is reachable from a single authenticated internal node, because
+/// the leaf successor is parent-mediated and resolves a child by its first
+/// occurrence: a node whose `leftmost_child` is `A` and whose entries are
+/// `[k1 → B, k2 → A]` answers "after A comes B" and "after B comes A". Each
+/// step is individually acyclic and each page authenticates; the scan simply
+/// alternates forever. Only a guard that lives as long as the scan ends it.
+///
+/// A healthy tree visits each leaf exactly once in key order, so this never
+/// fires on a well-formed scan.
+fn scan_guard() -> SeenPageIds {
+    SeenPageIds::new("btree_scan")
+}
 
 impl<V: Vfs> BTree<V> {
     /// Forward range scan: `start` inclusive, `end` exclusive.
@@ -18,9 +38,11 @@ impl<V: Vfs> BTree<V> {
             return Ok(Vec::new());
         }
         let mut path = self.path_to_leaf_for_key(start).await?;
+        let mut seen_leaves = scan_guard();
         let mut out: Vec<(Vec<u8>, Vec<u8>)> = Vec::new();
         loop {
             let leaf_id = *path.last().expect("non-empty path");
+            seen_leaves.insert(leaf_id)?;
             let leaf = self.read_leaf(leaf_id).await?;
             for (k, v) in &leaf.records {
                 if k.as_slice() >= end {
@@ -53,9 +75,11 @@ impl<V: Vfs> BTree<V> {
         // The empty key sorts below every stored key, so the descent lands on
         // the leftmost leaf.
         let mut path = self.path_to_leaf_for_key(&[]).await?;
+        let mut seen_leaves = scan_guard();
         let mut out: Vec<(Vec<u8>, Vec<u8>)> = Vec::new();
         loop {
             let leaf_id = *path.last().expect("non-empty path");
+            seen_leaves.insert(leaf_id)?;
             let leaf = self.read_leaf(leaf_id).await?;
             for (k, v) in &leaf.records {
                 let val = self.resolve_leaf_value(v).await?;
@@ -90,9 +114,11 @@ impl<V: Vfs> BTree<V> {
             return Ok(Vec::new());
         }
         let mut path = self.path_to_leaf_for_key(start).await?;
+        let mut seen_leaves = scan_guard();
         let mut out: Vec<(Vec<u8>, Vec<u8>)> = Vec::with_capacity(limit);
         loop {
             let leaf_id = *path.last().expect("non-empty path");
+            seen_leaves.insert(leaf_id)?;
             let leaf = self.read_leaf(leaf_id).await?;
             for (k, v) in &leaf.records {
                 if k.as_slice() < start {
@@ -166,9 +192,11 @@ impl<V: Vfs> BTree<V> {
             return Ok(Vec::new());
         }
         let mut path = self.path_to_leaf_for_key(prefix).await?;
+        let mut seen_leaves = scan_guard();
         let mut out: Vec<(Vec<u8>, Vec<u8>)> = Vec::new();
         loop {
             let leaf_id = *path.last().expect("non-empty path");
+            seen_leaves.insert(leaf_id)?;
             let leaf = self.read_leaf(leaf_id).await?;
             let mut past_prefix = false;
             for (k, v) in &leaf.records {

tests/btree_basic.rs

@@ -3,7 +3,7 @@ use std::sync::mpsc;
 use std::time::Duration;
 
 use pagedb::btree::BTree;
-use pagedb::btree::internal::Internal;
+use pagedb::btree::internal::{Internal, InternalEntry};
 use pagedb::btree::leaf::{Leaf, LeafValue};
 use pagedb::btree::node::body_capacity;
 use pagedb::btree::overflow::encode_overflow;
@@ -585,6 +585,81 @@ fn put_rejects_internal_child_cycle_without_hanging() {
     );
 }
 
+/// A forward scan's leaf successor is parent-mediated, and a child is resolved
+/// by its *first* occurrence in the parent. An internal node that lists one
+/// child twice therefore answers "after A comes B" and "after B comes A", and
+/// every individual step is acyclic — the cycle only exists across steps. The
+/// per-descent guards inside `next_leaf_after` are rebuilt on each call and
+/// cannot see it, so the scan needs a guard of its own.
+#[test]
+fn collect_all_rejects_alternating_leaf_successors_without_hanging() {
+    let (tx, rx) = mpsc::channel();
+
+    std::thread::spawn(move || {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        let result = runtime.block_on(async {
+            let pager = fresh_pager().await;
+            let realm = RealmId::new([1; 16]);
+            let (root_page_id, left_leaf, right_leaf) = (71u64, 72u64, 73u64);
+
+            for (page_id, key) in [(left_leaf, b"a"), (right_leaf, b"b")] {
+                let mut leaf = Leaf::new();
+                leaf.upsert(key, LeafValue::Inline(b"v".to_vec()));
+                let mut body = vec![0u8; body_capacity(PAGE)];
+                leaf.encode(&mut body).unwrap();
+                pager
+                    .write_main_page(page_id, realm, PageKind::BTreeLeaf, &body)
+                    .await
+                    .unwrap();
+            }
+
+            // `left_leaf` appears twice: as the leftmost child and again as the
+            // child to the right of `right_leaf`.
+            let internal = Internal {
+                leftmost_child: left_leaf,
+                entries: vec![
+                    InternalEntry {
+                        key: b"b".to_vec(),
+                        right_child: right_leaf,
+                    },
+                    InternalEntry {
+                        key: b"c".to_vec(),
+                        right_child: left_leaf,
+                    },
+                ],
+            };
+            let mut body = vec![0u8; body_capacity(PAGE)];
+            internal.encode(&mut body).unwrap();
+            pager
+                .write_main_page(root_page_id, realm, PageKind::BTreeInternal, &body)
+                .await
+                .unwrap();
+
+            let tree = BTree::open(pager, realm, root_page_id, 74, PAGE);
+            tree.collect_all().await.map(|_| ())
+        });
+        let _ = tx.send(result);
+    });
+
+    let result = rx
+        .recv_timeout(Duration::from_secs(5))
+        .expect("scan leaf-revisit detection should return before the timeout");
+    let error = result.expect_err("a scan that revisits a leaf must not be accepted");
+    assert!(
+        matches!(
+            error,
+            PagedbError::Corruption(CorruptionDetail::PageChainCycle {
+                structure: "btree_scan",
+                ..
+            })
+        ),
+        "expected a scan PageChainCycle, got {error:?}"
+    );
+}
+
 #[test]
 fn bulk_load_rejects_separator_that_cannot_fit_without_hanging() {
     let (tx, rx) = mpsc::channel();