fix(catalog): validate counter and segment rows in bounded batches

Reading an entire prefix scan into memory to validate it at open, or
to sum segment bytes for stats(), sizes an allocation by however many
counters or segments the embedder happens to have. Add
collect_prefix_batch_from to the B+ tree for bounded, resumable
prefix scans, and stream both catalog walks through it. Also replace
a saturating segment-bytes sum with a checked one that reports
overflow instead of silently wrapping.

Author: farhan-syah

CHANGELOG.md

@@ -40,6 +40,12 @@ version, since none exists yet.
 - **Online rekey** — rekey the database under a new key with mixed-cipher and
   mixed-epoch page coexistence (no full-file migration).
 - **Handle modes** — `Standalone`, `Follower`, `ReadOnly`, and `Observer`.
+- **Failures report themselves** — an unreadable free-list chain, main file, or
+  segment catalog fails `stats()` instead of reporting zero; compaction never
+  skips a catalog entry whose file it cannot open; segment open distinguishes a
+  missing file from a permission or backend error; and only genuine contention
+  is reported as contention. Persisted named-counter rows are validated at open,
+  and commit-history keys are rejected unless exactly eight bytes.
 
 ### Known limitations
 

src/btree/tree/scan.rs

@@ -114,6 +114,29 @@ impl<V: Vfs> BTree<V> {
         }
     }
 
+    /// Collect at most `limit` records at or after `start` whose keys still
+    /// carry `prefix`, in ascending key order.
+    ///
+    /// The bounded counterpart to [`Self::scan_prefix`]. Rows sharing a prefix
+    /// sort contiguously, so the first key outside it ends the range — the scan
+    /// stops there rather than reading the rest of the tree.
+    ///
+    /// Resume as with [`Self::collect_batch_from`]: pass the last returned key
+    /// with a `0x00` byte appended. A batch shorter than `limit` means the
+    /// prefix range ended.
+    pub async fn collect_prefix_batch_from(
+        &self,
+        prefix: &[u8],
+        start: &[u8],
+        limit: usize,
+    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+        let mut batch = self.collect_batch_from(start, limit).await?;
+        if let Some(end) = batch.iter().position(|(key, _)| !key.starts_with(prefix)) {
+            batch.truncate(end);
+        }
+        Ok(batch)
+    }
+
     /// Return the smallest key in the tree, or `None` if the tree is empty.
     /// Descends the leftmost spine only — O(tree height), not O(tree size).
     pub async fn first_key(&self) -> Result<Option<Vec<u8>>> {

src/txn/db/catalog.rs

@@ -15,6 +15,11 @@ use super::core::{
     encode_root_ref,
 };
 
+/// Named-counter rows read per batch while validating them at open. Rows are a
+/// fixed-width authenticated value, so this is a few KiB resident regardless of
+/// how many counters the embedder has named.
+const COUNTER_ROW_BATCH: usize = 512;
+
 impl<V: Vfs + Clone> Db<V> {
     /// The oldest commit id still retained in the commit-history index, or
     /// `None` when history is disabled or the index is empty. Pages reachable
@@ -56,6 +61,9 @@ impl<V: Vfs + Clone> Db<V> {
     ///
     /// Named counters are already atomic with catalog-root publication, so
     /// recovery validates their encoding but never rewrites their values.
+    ///
+    /// Rows are streamed in bounded batches: how many counters an embedder has
+    /// named is its business, and open must not size an allocation by it.
     pub(super) async fn validate_counter_rows(
         &self,
         catalog_root_page_id: u64,
@@ -73,10 +81,28 @@ impl<V: Vfs + Clone> Db<V> {
             next_page_id,
             self.page_size,
         );
-        for (_key, value) in tree.scan_prefix(&prefix).await? {
-            Catalog::decode_counter(&value)?;
+        let mut cursor: Vec<u8> = prefix.to_vec();
+        loop {
+            let batch = tree
+                .collect_prefix_batch_from(&prefix, &cursor, COUNTER_ROW_BATCH)
+                .await?;
+            let Some((last_key, _)) = batch.last() else {
+                return Ok(());
+            };
+            cursor.clear();
+            cursor.extend_from_slice(last_key);
+            // The exact successor of `last_key` in the key ordering: resume
+            // strictly past the row just validated without re-reading it.
+            cursor.push(0);
+            let exhausted = batch.len() < COUNTER_ROW_BATCH;
+
+            for (_key, value) in &batch {
+                Catalog::decode_counter(value)?;
+            }
+            if exhausted {
+                return Ok(());
+            }
         }
-        Ok(())
     }
 
     /// Write per-realm quota caps into the catalog B+ tree and persist the

src/txn/db/misc.rs

@@ -5,6 +5,7 @@ use crate::Result;
 use crate::btree::BTree;
 use crate::catalog::codec::{Catalog, CatalogRowKind};
 use crate::crypto::SecretKey;
+use crate::errors::PagedbError;
 use crate::observability::DbStats;
 use crate::vfs::types::OpenMode;
 use crate::vfs::{Vfs, VfsFile};
@@ -13,6 +14,11 @@ use std::sync::atomic::Ordering as AtOrd;
 use super::super::mode::DbMode;
 use super::core::Db;
 
+/// Segment catalog rows read per batch while aggregating `stats()`. Rows are a
+/// fixed-width authenticated value, so this is a few KiB resident regardless of
+/// how many segments a realm has linked.
+const SEGMENT_ROW_BATCH: usize = 512;
+
 impl<V: Vfs + Clone> Db<V> {
     /// Return the mode this handle was opened with.
     pub fn mode(&self) -> DbMode {
@@ -157,13 +163,36 @@ impl<V: Vfs + Clone> Db<V> {
                 self.page_size,
             );
 
-            let seg_start = vec![CatalogRowKind::Segment as u8];
-            let seg_rows = tree.scan_prefix(&seg_start).await?;
-            let seg_count = u32::try_from(seg_rows.len()).unwrap_or(u32::MAX);
+            // Streamed in bounded batches: how many segments a realm has linked
+            // is the embedder's business, and a metrics call must not size an
+            // allocation by it.
+            let seg_prefix = [CatalogRowKind::Segment as u8];
+            let mut cursor: Vec<u8> = seg_prefix.to_vec();
+            let mut seg_count = 0u32;
             let mut seg_bytes = 0u64;
-            for (_key, value) in &seg_rows {
-                seg_bytes =
-                    seg_bytes.saturating_add(Catalog::decode_segment_meta(value)?.total_bytes);
+            loop {
+                let batch = tree
+                    .collect_prefix_batch_from(&seg_prefix, &cursor, SEGMENT_ROW_BATCH)
+                    .await?;
+                let Some((last_key, _)) = batch.last() else {
+                    break;
+                };
+                cursor.clear();
+                cursor.extend_from_slice(last_key);
+                cursor.push(0);
+                let exhausted = batch.len() < SEGMENT_ROW_BATCH;
+
+                for (_key, value) in &batch {
+                    seg_count = seg_count.saturating_add(1);
+                    seg_bytes = seg_bytes
+                        .checked_add(Catalog::decode_segment_meta(value)?.total_bytes)
+                        .ok_or_else(|| {
+                            PagedbError::arithmetic_overflow("stats.segments_total_bytes")
+                        })?;
+                }
+                if exhausted {
+                    break;
+                }
             }
 
             (seg_count, seg_bytes)