fix(catalog): reject malformed persisted metadata

Author: presempathy-awb

src/txn/db/catalog.rs

@@ -44,14 +44,41 @@ impl<V: Vfs + Clone> Db<V> {
         let Some(key) = hist.first_key().await? else {
             return Ok(None);
         };
-        if key.len() < 8 {
-            return Ok(None);
+        if key.len() != 8 {
+            return Err(PagedbError::catalog_row_invalid("commit_history.key"));
         }
         let mut b = [0u8; 8];
         b.copy_from_slice(&key[..8]);
         Ok(Some(u64::from_be_bytes(b)))
     }
 
+    /// Authenticate and decode every persisted named-counter row during open.
+    ///
+    /// Named counters are already atomic with catalog-root publication, so
+    /// recovery validates their encoding but never rewrites their values.
+    pub(super) async fn validate_counter_rows(
+        &self,
+        catalog_root_page_id: u64,
+        next_page_id: u64,
+    ) -> Result<()> {
+        if catalog_root_page_id == 0 {
+            return Ok(());
+        }
+
+        let prefix = [crate::catalog::codec::CatalogRowKind::Counter as u8];
+        let tree = BTree::open(
+            self.pager.clone(),
+            self.realm_id,
+            catalog_root_page_id,
+            next_page_id,
+            self.page_size,
+        );
+        for (_key, value) in tree.scan_prefix(&prefix).await? {
+            Catalog::decode_counter(&value)?;
+        }
+        Ok(())
+    }
+
     /// Write per-realm quota caps into the catalog B+ tree and persist the
     /// updated catalog root to the A/B header.
     pub async fn set_realm_quotas(&self, realm: RealmId, quotas: RealmQuotas) -> Result<()> {
@@ -307,3 +334,73 @@ impl<V: Vfs + Clone> Db<V> {
         Ok(freed)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::vfs::memory::MemVfs;
+    use crate::{Db, PagedbError, RealmId};
+
+    use super::*;
+
+    const PAGE: usize = 4096;
+    const REALM: RealmId = RealmId::new([0xA7; 16]);
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn counter_recovery_surfaces_malformed_counter_row() {
+        let db = Db::open_internal(MemVfs::new(), [9u8; 32], PAGE, REALM)
+            .await
+            .unwrap();
+        {
+            let mut txn = db.begin_write().await.unwrap();
+            let mut counter = txn.counter("bad-counter").unwrap();
+            counter.set(5).await.unwrap();
+            drop(counter);
+            txn.commit().await.unwrap();
+        }
+
+        let (catalog_root, next_page_id) = {
+            let state = db.writer.lock().await;
+            (state.catalog_root_page_id, state.next_page_id)
+        };
+        let mut tree = BTree::open(
+            db.pager.clone(),
+            db.realm_id,
+            catalog_root,
+            next_page_id,
+            db.page_size,
+        );
+        tree.put(&Catalog::counter_key(&[0xFF]).unwrap(), b"bad")
+            .await
+            .unwrap();
+        tree.flush().await.unwrap();
+
+        let err = db
+            .validate_counter_rows(tree.root_page_id(), tree.next_page_id())
+            .await
+            .expect_err("malformed counter row must surface during recovery validation");
+        assert!(matches!(err, PagedbError::Corruption(_)));
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn oldest_retained_history_commit_surfaces_malformed_history_key() {
+        for malformed_key in [b"x".as_slice(), b"123456789".as_slice()] {
+            let db = Db::open_internal(MemVfs::new(), [9u8; 32], PAGE, REALM)
+                .await
+                .unwrap();
+            let next_page_id = db.writer.lock().await.next_page_id;
+            let mut history =
+                BTree::open(db.pager.clone(), db.realm_id, 0, next_page_id, db.page_size);
+            history
+                .put(malformed_key, b"malformed history")
+                .await
+                .unwrap();
+            history.flush().await.unwrap();
+
+            let err = db
+                .oldest_retained_history_commit(history.root_page_id(), history.next_page_id())
+                .await
+                .expect_err("malformed history key must surface");
+            assert!(matches!(err, PagedbError::Corruption(_)));
+        }
+    }
+}

src/txn/db/misc.rs

@@ -160,11 +160,11 @@ impl<V: Vfs + Clone> Db<V> {
             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);
-            let seg_bytes: u64 = seg_rows
-                .iter()
-                .filter_map(|(_k, v)| Catalog::decode_segment_meta(v).ok())
-                .map(|m| m.total_bytes)
-                .sum();
+            let mut seg_bytes = 0u64;
+            for (_key, value) in &seg_rows {
+                seg_bytes =
+                    seg_bytes.saturating_add(Catalog::decode_segment_meta(value)?.total_bytes);
+            }
 
             (seg_count, seg_bytes)
         };
@@ -189,3 +189,66 @@ impl<V: Vfs + Clone> Db<V> {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::btree::BTree;
+    use crate::catalog::codec::Catalog;
+    use crate::vfs::memory::MemVfs;
+    use crate::{Db, PagedbError, RealmId, SegmentKind, SegmentPageKind};
+
+    const PAGE: usize = 4096;
+    const REALM: RealmId = RealmId::new([0xA5; 16]);
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn stats_surfaces_malformed_segment_catalog_row() {
+        let db = Db::open_internal(MemVfs::new(), [9u8; 32], PAGE, REALM)
+            .await
+            .unwrap();
+        let mut segment = db
+            .create_segment(REALM, SegmentKind::Unspecified)
+            .await
+            .unwrap();
+        segment
+            .append_page(SegmentPageKind::Data, b"stats")
+            .await
+            .unwrap();
+        let meta = segment.seal().await.unwrap();
+        {
+            let mut txn = db.begin_write().await.unwrap();
+            txn.link_segment("good", &meta).await.unwrap();
+            txn.commit().await.unwrap();
+        }
+
+        let (catalog_root, next_page_id) = {
+            let state = db.writer.lock().await;
+            (state.catalog_root_page_id, state.next_page_id)
+        };
+        let mut tree = BTree::open(
+            db.pager.clone(),
+            db.realm_id,
+            catalog_root,
+            next_page_id,
+            db.page_size,
+        );
+        tree.put(
+            &Catalog::segment_key(REALM, b"bad").unwrap(),
+            b"not a segment meta",
+        )
+        .await
+        .unwrap();
+        tree.flush().await.unwrap();
+        {
+            let mut state = db.writer.lock().await;
+            state.catalog_root_page_id = tree.root_page_id();
+            state.next_page_id = state.next_page_id.max(tree.next_page_id());
+            db.publish_snapshot(&state);
+        }
+
+        let err = db
+            .stats()
+            .await
+            .expect_err("malformed segment metadata must surface");
+        assert!(matches!(err, PagedbError::Corruption(_)));
+    }
+}

src/txn/db/open/recovery.rs

@@ -72,5 +72,8 @@ pub(super) async fn recover_open_state<V: Vfs + Clone>(
         .await?;
     }
 
+    db.validate_counter_rows(catalog_root_page_id, next_page_id)
+        .await?;
+
     Ok(())
 }

src/txn/db/reader.rs

@@ -262,17 +262,18 @@ impl<V: Vfs + Clone> Db<V> {
                 false,
             ))
         } else {
-            // Find the oldest available: scan the whole history tree. An
-            // exclusive `u64::MAX` upper bound would hide the newest commit
-            // once ids reach the top of the range.
-            let oldest = tree.collect_all().await?.into_iter().next().map_or(
-                CommitId(latest_commit_id),
-                |(k, _)| {
-                    let mut b = [0u8; 8];
-                    b.copy_from_slice(&k[..8]);
-                    CommitId(u64::from_be_bytes(b))
-                },
-            );
+            // Only the leftmost key is needed. This avoids scanning retained
+            // history while still validating its fixed-width key encoding.
+            let oldest = if let Some(key) = tree.first_key().await? {
+                if key.len() != 8 {
+                    return Err(PagedbError::catalog_row_invalid("commit_history.key"));
+                }
+                let mut bytes = [0u8; 8];
+                bytes.copy_from_slice(&key[..8]);
+                CommitId(u64::from_be_bytes(bytes))
+            } else {
+                CommitId(latest_commit_id)
+            };
             Err(PagedbError::CommitGone {
                 commit,
                 oldest_available: oldest,
@@ -285,17 +286,51 @@ impl<V: Vfs + Clone> Db<V> {
 mod tests {
     use std::sync::Arc;
 
+    use crate::btree::BTree;
     use crate::catalog::codec::SegmentKind;
+    use crate::errors::PagedbError;
     use crate::segment::types::SegmentPageKind;
     use crate::vfs::memory::MemVfs;
-    use crate::{Db, RealmId};
+    use crate::{CommitId, Db, RealmId};
 
     use super::super::core::VisibilityTestHook;
 
     const PAGE: usize = 4096;
     const KEK: [u8; 32] = [0xD1; 32];
     const REALM: RealmId = RealmId::new([0xD2; 16]);
 
+    #[tokio::test(flavor = "current_thread")]
+    async fn begin_read_at_surfaces_malformed_oldest_history_key() {
+        for malformed_key in [b"x".as_slice(), b"123456789".as_slice()] {
+            let db = Db::open_internal(MemVfs::new(), [9u8; 32], PAGE, REALM)
+                .await
+                .unwrap();
+            let next_page_id = db.writer.lock().await.next_page_id;
+            let mut history =
+                BTree::open(db.pager.clone(), db.realm_id, 0, next_page_id, db.page_size);
+            history
+                .put(malformed_key, b"malformed history")
+                .await
+                .unwrap();
+            history.flush().await.unwrap();
+            {
+                let mut state = db.writer.lock().await;
+                state.commit_history_root_page_id = history.root_page_id();
+                state.next_page_id = state.next_page_id.max(history.next_page_id());
+                db.publish_snapshot(&state);
+            }
+
+            let err = match db.begin_read_at(CommitId::new(42)).await {
+                Ok(read) => {
+                    drop(read);
+                    panic!("malformed oldest history key must not produce a reader");
+                }
+                Err(err) => err,
+            };
+            assert!(matches!(err, PagedbError::Corruption(_)));
+        }
+    }
+
     async fn linked_segment(db: &Db<MemVfs>, name: &str, bytes: &[u8]) {
         let mut writer = db
             .create_segment(REALM, SegmentKind::Unspecified)