fix(recovery): harden replay and deep-walk reads

Author: presempathy-awb

src/recovery/deep_walk.rs

@@ -21,7 +21,7 @@ use crate::pager::{PageGuard, Pager};
 use crate::segment::authenticated_metadata::authenticate_segment_metadata;
 use crate::txn::db::Db;
 use crate::vfs::types::OpenMode;
-use crate::vfs::{Vfs, VfsFile};
+use crate::vfs::{Vfs, VfsFile, read_exact_at};
 
 /// A single page-level issue found during deep walk.
 #[derive(Debug, Clone)]
@@ -217,7 +217,7 @@ pub async fn run_deep_walk<V: Vfs + Clone>(db: &Db<V>) -> Result<DeepWalkReport>
 
     let vfs: &V = &db.vfs;
     let main_file_res = vfs.open(main_db_path, OpenMode::Read).await;
-    let main_file = match main_file_res {
+    let mut main_file = match main_file_res {
         Ok(f) => f,
         Err(e) => {
             report.page_issues.push(PageIssue {
@@ -232,24 +232,24 @@ pub async fn run_deep_walk<V: Vfs + Clone>(db: &Db<V>) -> Result<DeepWalkReport>
     // (HK-MAC, cleartext) and are already verified by `Db::open`. Skip them.
     // Page 2 and 3 are reserved (apply-journal). Walk from page 4.
     for page_id in 4..next_page_id {
-        let offset = page_id * page_size as u64;
-        let mut buf = vec![0u8; page_size];
-        match main_file.read_at(offset, &mut buf).await {
-            Ok(n) if n < page_size => {
-                // Short read at the tail — the file may be smaller than expected.
-                // Report but continue.
+        let offset = match page_offset(page_id, page_size) {
+            Ok(offset) => offset,
+            Err(error) => {
                 report.page_issues.push(PageIssue {
                     page_id,
-                    description: format!("short read: expected {page_size} bytes, got {n}"),
+                    description: format!("page offset overflow: {error}"),
                 });
                 report.pages_examined += 1;
                 continue;
             }
-            Ok(_) => {}
-            Err(e) => {
+        };
+        let mut buf = vec![0u8; page_size];
+        match read_exact_at(&mut main_file, offset, &mut buf).await {
+            Ok(()) => {}
+            Err(error) => {
                 report.page_issues.push(PageIssue {
                     page_id,
-                    description: format!("read error: {e}"),
+                    description: format!("read error: {error}"),
                 });
                 report.pages_examined += 1;
                 continue;
@@ -380,7 +380,7 @@ async fn check_segment<V: Vfs + Clone>(
     let page_size = pager.page_size();
 
     // Check file exists.
-    let Ok(file) = vfs.open(&live, OpenMode::Read).await else {
+    let Ok(mut file) = vfs.open(&live, OpenMode::Read).await else {
         report.drift_issues.push(DriftIssue {
             segment_id: meta.segment_id,
             description: "segment file missing from seg/".to_string(),
@@ -409,7 +409,16 @@ async fn check_segment<V: Vfs + Clone>(
     // We don't have a metadata API, but we can check via read: try reading one
     // byte past the expected end. If it succeeds (on some VFS) we skip the
     // check; if we read exactly `page_count * page_size` bytes we're consistent.
-    let expected_size = meta.page_count * page_size as u64;
+    let expected_size = match page_offset(meta.page_count, page_size) {
+        Ok(offset) => offset,
+        Err(error) => {
+            report.segment_issues.push(SegmentIssue {
+                segment_id: meta.segment_id,
+                description: format!("catalog page_count offset overflow: {error}"),
+            });
+            return;
+        }
+    };
     let mut probe = vec![0u8; 1];
     let over_read = file.read_at(expected_size, &mut probe).await;
     match over_read {
@@ -428,24 +437,23 @@ async fn check_segment<V: Vfs + Clone>(
     // Walk data pages (1 .. page_count - 1, skipping header=0 and footer=last).
     let last_data = footer_page_id;
     for page_id in 1..last_data {
-        let offset = page_id * page_size as u64;
-        let mut buf = vec![0u8; page_size];
-        let read_res = file.read_at(offset, &mut buf).await;
-        match read_res {
-            Ok(n) if n < page_size => {
+        let offset = match page_offset(page_id, page_size) {
+            Ok(offset) => offset,
+            Err(error) => {
                 report.segment_issues.push(SegmentIssue {
                     segment_id: meta.segment_id,
-                    description: format!(
-                        "short read at page {page_id}: expected {page_size} bytes, got {n}"
-                    ),
+                    description: format!("data page {page_id} offset overflow: {error}"),
                 });
                 continue;
             }
-            Ok(_) => {}
-            Err(e) => {
+        };
+        let mut buf = vec![0u8; page_size];
+        match read_exact_at(&mut file, offset, &mut buf).await {
+            Ok(()) => {}
+            Err(error) => {
                 report.segment_issues.push(SegmentIssue {
                     segment_id: meta.segment_id,
-                    description: format!("read error at page {page_id}: {e}"),
+                    description: format!("read error at page {page_id}: {error}"),
                 });
                 continue;
             }
@@ -498,6 +506,14 @@ async fn check_segment<V: Vfs + Clone>(
     }
 }
 
+fn page_offset(page_id: u64, page_size: usize) -> Result<u64> {
+    let page_size = u64::try_from(page_size)
+        .map_err(|_| crate::PagedbError::arithmetic_overflow("deep-walk page size"))?;
+    page_id
+        .checked_mul(page_size)
+        .ok_or_else(|| crate::PagedbError::arithmetic_overflow("deep-walk page offset"))
+}
+
 /// Collect the set of all page IDs reachable from the main B+ tree root,
 /// the catalog root, the commit-history root, and the free-list root.
 /// Pages 0..=3 (reserved) are always considered reachable.
@@ -822,10 +838,10 @@ async fn diagnose_overflow_chain<V: Vfs + Clone>(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::OpenOptions;
     use crate::btree::node::body_capacity;
     use crate::pager::format::data_page::ENVELOPE_OVERHEAD;
     use crate::vfs::memory::MemVfs;
+    use crate::{OpenOptions, SegmentKind, SegmentPageKind};
 
     const PAGE: usize = 4096;
     const REALM: crate::RealmId = crate::RealmId::new([0xD3; 16]);
@@ -1019,6 +1035,59 @@ mod tests {
         );
     }
 
+    #[tokio::test(flavor = "current_thread")]
+    async fn deep_walk_reports_segment_page_count_offset_overflow() {
+        let db = open_db().await;
+        let mut segment = db
+            .create_segment(REALM, SegmentKind::Unspecified)
+            .await
+            .unwrap();
+        segment
+            .append_page(SegmentPageKind::Data, b"deep-walk")
+            .await
+            .unwrap();
+        let mut meta = segment.seal().await.unwrap();
+        {
+            let mut txn = db.begin_write().await.unwrap();
+            txn.link_segment("overflow", &meta).await.unwrap();
+            txn.commit().await.unwrap();
+        }
+
+        meta.page_count = (u64::MAX / PAGE as u64) + 2;
+        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,
+        );
+        let key = Catalog::segment_key(REALM, b"overflow").unwrap();
+        tree.put(&key, &Catalog::encode_segment_meta(&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());
+        }
+
+        let report = run_deep_walk(&db).await.unwrap();
+        assert!(
+            report.segment_issues.iter().any(|issue| {
+                issue.segment_id == meta.segment_id
+                    && issue
+                        .description
+                        .contains("authenticated segment metadata invalid")
+            }),
+            "overflowing segment geometry must become a structured issue, got {report:?}"
+        );
+    }
+
     async fn sole_overflow_root(db: &Db<MemVfs>, leaf_page_id: u64) -> u64 {
         let (guard, _) = db.pager.read_main_node(leaf_page_id, REALM).await.unwrap();
         let leaf = Leaf::decode(guard.body_ref()).unwrap();

src/recovery/journal.rs

@@ -311,7 +311,7 @@ pub async fn execute_journal_actions<V: Vfs>(vfs: &V, actions: &[JournalAction])
                     crate::hex::to_hex_lower(segment_id),
                     tombstone_commit_id,
                 );
-                if !path_exists(vfs, &dst).await? && path_exists(vfs, &src).await? {
+                if !path_exists(vfs, &dst).await? {
                     vfs.rename(&src, &dst).await?;
                 }
             }

src/recovery/reconcile.rs

@@ -226,7 +226,13 @@ mod tests {
     use crate::vfs::{Vfs, VfsFile};
     use crate::{RealmId, btree::BTree};
 
-    use super::repair_catalog;
+    use super::{repair_catalog, sweep_orphans};
+
+    fn segment_id(value: u64) -> [u8; 16] {
+        let mut id = [0; 16];
+        id[..8].copy_from_slice(&value.to_le_bytes());
+        id
+    }
 
     #[tokio::test(flavor = "current_thread")]
     async fn malformed_catalog_key_prevents_reconciliation_mutation() {
@@ -296,4 +302,45 @@ mod tests {
         ));
         assert!(vfs.open(marker, OpenMode::Read).await.is_ok());
     }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn sweep_orphans_handles_many_expected_segments() {
+        let vfs = MemVfs::new();
+        vfs.mkdir_all("seg/.staging").await.unwrap();
+        let expected: Vec<[u8; 16]> = (0..1024).map(segment_id).collect();
+
+        for id in expected.iter().take(8) {
+            let path = crate::segment::writer::live_path(id);
+            let mut file = vfs.open(&path, OpenMode::CreateOrOpen).await.unwrap();
+            file.write_at(0, b"live").await.unwrap();
+        }
+
+        let live_orphan = segment_id(10_000);
+        let live_orphan_path = crate::segment::writer::live_path(&live_orphan);
+        let mut live_file = vfs
+            .open(&live_orphan_path, OpenMode::CreateOrOpen)
+            .await
+            .unwrap();
+        live_file.write_at(0, b"orphan").await.unwrap();
+
+        let staged_orphan = segment_id(10_001);
+        let staged_orphan_path = crate::segment::writer::staging_path(&staged_orphan);
+        let mut staged_file = vfs
+            .open(&staged_orphan_path, OpenMode::CreateOrOpen)
+            .await
+            .unwrap();
+        staged_file.write_at(0, b"orphan").await.unwrap();
+
+        sweep_orphans(&vfs, &expected, 77).await.unwrap();
+
+        let expected_live = crate::segment::writer::live_path(&expected[0]);
+        assert!(vfs.open(&expected_live, OpenMode::Read).await.is_ok());
+        assert!(vfs.open(&live_orphan_path, OpenMode::Read).await.is_err());
+        let tombstone = format!(
+            "seg/.tombstone/{}.77",
+            crate::hex::to_hex_lower(&live_orphan)
+        );
+        assert!(vfs.open(&tombstone, OpenMode::Read).await.is_ok());
+        assert!(vfs.open(&staged_orphan_path, OpenMode::Read).await.is_err());
+    }
 }

tests/durability/apply_journal_crash.rs

@@ -1,6 +1,7 @@
-/// Verify apply-journal encode/decode round-trips and idempotent action replay.
-/// These tests exercise the journal machinery at the level of the public
-/// encode/decode API and the VFS rename layer without requiring a full crash.
+//! Verify apply-journal encoding and idempotent action replay at the VFS
+//! boundary without requiring a full process crash.
+
+use pagedb::PagedbError;
 use pagedb::recovery::journal::{ApplyJournalRecord, JournalAction, decode_record, encode_record};
 use pagedb::vfs::Vfs;
 use pagedb::vfs::memory::MemVfs;
@@ -134,13 +135,54 @@ async fn promote_action_renames_staging_to_live() {
 }
 
 #[tokio::test(flavor = "current_thread")]
-async fn tombstone_action_is_idempotent_when_live_absent() {
-    // If the live file is absent, the tombstone rename is a no-op.
+async fn tombstone_action_errors_when_live_and_tombstone_are_absent() {
     use pagedb::recovery::journal::execute_journal_actions;
+
     let vfs = MemVfs::new();
     let actions = vec![JournalAction::Tombstone {
         segment_id: [0xEE; 16],
         tombstone_commit_id: 5,
     }];
+    let err = execute_journal_actions(&vfs, &actions)
+        .await
+        .expect_err("missing tombstone source is only idempotent when the destination exists");
+    assert!(
+        matches!(err, PagedbError::Io(ref io) if io.kind() == std::io::ErrorKind::NotFound),
+        "expected missing tombstone source and destination to surface NotFound, got {err:?}"
+    );
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn tombstone_action_is_idempotent_when_tombstone_exists() {
+    use pagedb::recovery::journal::execute_journal_actions;
+    use pagedb::vfs::VfsFile;
+    use pagedb::vfs::types::OpenMode;
+
+    let vfs = MemVfs::new();
+    let segment_id = [0xEF; 16];
+    let tombstone = format!(
+        "seg/.tombstone/{}.5",
+        segment_id
+            .iter()
+            .map(|byte| format!("{byte:02x}"))
+            .collect::<String>()
+    );
+    vfs.mkdir_all("seg/.tombstone").await.unwrap();
+    {
+        let mut file = vfs.open(&tombstone, OpenMode::CreateNew).await.unwrap();
+        file.write_at(0, b"already-tombstoned").await.unwrap();
+        file.sync().await.unwrap();
+    }
+
+    let actions = vec![JournalAction::Tombstone {
+        segment_id,
+        tombstone_commit_id: 5,
+    }];
     execute_journal_actions(&vfs, &actions).await.unwrap();
+
+    let file = vfs.open(&tombstone, OpenMode::Read).await.unwrap();
+    let mut bytes = vec![0; 18];
+    let read = file.read_at(0, &mut bytes).await.unwrap();
+    assert_eq!(read, bytes.len());
+    assert_eq!(&bytes, b"already-tombstoned");
 }