fix(storage): complete metadata I/O and surface backend errors

Author: presempathy-awb

src/compaction/full.rs

@@ -84,10 +84,12 @@ async fn compact_now_inner<V: Vfs + Clone>(db: &Db<V>) -> Result<CompactStats> {
     let all_segments = list_all_segments_inner(&db.pager, db.realm_id, &state).await?;
     for meta in all_segments {
         let live = crate::segment::writer::live_path(&meta.segment_id);
-        let file_size = match db.vfs.open(&live, crate::vfs::types::OpenMode::Read).await {
-            Ok(f) => f.len().await.unwrap_or(meta.total_bytes),
-            Err(_) => continue,
-        };
+        let file_size = db
+            .vfs
+            .open(&live, crate::vfs::types::OpenMode::Read)
+            .await?
+            .len()
+            .await?;
         // Skip segments with < 5% garbage.
         let threshold = meta.total_bytes.saturating_add(meta.total_bytes / 20);
         if file_size <= threshold {

src/pager/core.rs

@@ -23,7 +23,7 @@ use crate::pager::format::data_page::{
 use crate::pager::format::page_kind::PageKind;
 use crate::txn::db::rekey::EpochKeyring;
 use crate::vfs::types::{OpenMode, WriteReq};
-use crate::vfs::{Vfs, VfsFile};
+use crate::vfs::{Vfs, VfsFile, read_exact_at};
 use crate::{RealmId, Result};
 use rayon::prelude::*;
 
@@ -741,13 +741,8 @@ impl<V: Vfs> Pager<V> {
             }
             let mut buf = vec![0u8; page_size];
             {
-                let f = file_handle.lock().await;
-                let n = f.read_at(page_offset, &mut buf).await?;
-                if n < page_size {
-                    for b in &mut buf[n..] {
-                        *b = 0;
-                    }
-                }
+                let mut f = file_handle.lock().await;
+                read_exact_at(&mut *f, page_offset, &mut buf).await?;
             }
 
             // Extract the cipher_id and mk_epoch recorded in this specific page's

src/pager/header.rs

@@ -11,7 +11,7 @@ use crate::pager::format::structural_header::{
     MainDbHeaderFields, decode_main_db_header, encode_main_db_header,
 };
 use crate::vfs::types::OpenMode;
-use crate::vfs::{Vfs, VfsFile};
+use crate::vfs::{Vfs, VfsFile, read_exact_at, write_all_at};
 
 /// Which header slot is the authoritative current header.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -57,15 +57,15 @@ pub async fn bootstrap_header<V: Vfs>(
     let bytes = encode_main_db_header(initial, hk, page_size)?;
     let mut f = vfs.open(path, OpenMode::CreateNew).await?;
     // Slot A at offset 0.
-    f.write_at(0, &bytes).await?;
+    write_all_at(&mut f, 0, &bytes).await?;
     // Slot B at offset page_size — write a zero-filled page so the slot is
     // materialised on disk. Decode of a zero buffer fails the magic check
     // and returns `Corruption(HeaderUnverifiable)`, which `open_header`
     // treats as "this slot is unverifiable; skip it."
     let zero = vec![0u8; page_size];
     let page_size_u64 = u64::try_from(page_size)
         .map_err(|_| PagedbError::Io(std::io::Error::other("page_size > u64")))?;
-    f.write_at(page_size_u64, &zero).await?;
+    write_all_at(&mut f, page_size_u64, &zero).await?;
     f.sync().await?;
     // Make the directory entry for the newly created file durable so a
     // power loss immediately after creation does not lose the file.
@@ -92,13 +92,13 @@ pub async fn open_header<V: Vfs>(
     hk: &DerivedKey,
     page_size: usize,
 ) -> Result<(MainDbHeaderFields, ActiveSlot)> {
-    let f = vfs.open(path, OpenMode::ReadWrite).await?;
+    let mut f = vfs.open(path, OpenMode::ReadWrite).await?;
     let mut buf_a = vec![0u8; page_size];
     let mut buf_b = vec![0u8; page_size];
-    let _ = f.read_at(0, &mut buf_a).await?;
+    read_exact_at(&mut f, 0, &mut buf_a).await?;
     let page_size_u64 = u64::try_from(page_size)
         .map_err(|_| PagedbError::Io(std::io::Error::other("page_size > u64")))?;
-    let _ = f.read_at(page_size_u64, &mut buf_b).await?;
+    read_exact_at(&mut f, page_size_u64, &mut buf_b).await?;
     let a = decode_main_db_header(&buf_a, hk, page_size).ok();
     let b = decode_main_db_header(&buf_b, hk, page_size).ok();
     match (a, b) {
@@ -139,7 +139,7 @@ pub async fn commit_header<V: Vfs>(
         .ok()
         .map(|s| next.page_id().saturating_mul(s))
         .ok_or_else(|| PagedbError::Io(std::io::Error::other("offset arithmetic overflow")))?;
-    f.write_at(offset, &bytes).await?;
+    write_all_at(&mut f, offset, &bytes).await?;
     f.sync().await?;
     // No `sync_dir` here: a header rewrite is a data write to an existing,
     // already-durable inode (main.db). Architecture §883 requires `sync_dir`

src/segment/authenticated_metadata.rs

@@ -11,8 +11,7 @@ use crate::pager::format::data_page::{body, extract_page_header_ids, open_data_p
 use crate::pager::format::page_kind::PageKind;
 use crate::pager::format::segment_footer::{SegmentFooterFields, decode_segment_footer};
 use crate::pager::format::structural_header::{SegmentHeaderFields, decode_segment_header};
-use crate::vfs::Vfs;
-use crate::vfs::VfsFile;
+use crate::vfs::{Vfs, VfsFile, checked_read_progress};
 
 use super::types::{EXTENT_INDEX_ENTRY_LEN, ExtentIndexEntry};
 use super::writer::{live_path, staging_path};
@@ -85,17 +84,23 @@ pub(crate) async fn authenticate_segment_metadata<V: Vfs + Clone>(
     let master_key = pager.mk_for(meta.mk_epoch, cipher_id)?;
     let hk = derive_hk(&master_key)?;
     let mut header_bytes = vec![0u8; page_size];
-    let header_read = file.read_at(0, &mut header_bytes).await?;
-    if header_read != page_size {
-        return Err(PagedbError::segment_geometry_invalid("header_read"));
+    let mut header_offset = 0;
+    let mut header_remaining = &mut header_bytes[..];
+    while !header_remaining.is_empty() {
+        let read = file.read_at(header_offset, header_remaining).await?;
+        checked_read_progress(&mut header_offset, read, header_remaining.len())?;
+        header_remaining = header_remaining.split_at_mut(read).1;
     }
     let header = decode_segment_header(&header_bytes, &hk, page_size)?;
     validate_header(&header, meta, parent_file_id, page_size)?;
 
     let mut footer_bytes = vec![0u8; page_size];
-    let footer_read = file.read_at(footer_offset, &mut footer_bytes).await?;
-    if footer_read != page_size {
-        return Err(PagedbError::segment_geometry_invalid("footer_read"));
+    let mut next_footer_offset = footer_offset;
+    let mut footer_remaining = &mut footer_bytes[..];
+    while !footer_remaining.is_empty() {
+        let read = file.read_at(next_footer_offset, footer_remaining).await?;
+        checked_read_progress(&mut next_footer_offset, read, footer_remaining.len())?;
+        footer_remaining = footer_remaining.split_at_mut(read).1;
     }
     let (footer, manifest) = {
         let mut lru = pager.dek_lru().lock();
@@ -332,8 +337,12 @@ async fn collect_and_decode_index_page<V: Vfs + Clone>(
         .checked_mul(page_size_u64)
         .ok_or_else(|| PagedbError::segment_geometry_invalid("index.offset"))?;
     let mut page = vec![0u8; context.page_size];
-    if context.file.read_at(offset, &mut page).await? != context.page_size {
-        return Err(PagedbError::segment_geometry_invalid("index.read"));
+    let mut read_offset = offset;
+    let mut remaining = &mut page[..];
+    while !remaining.is_empty() {
+        let read = context.file.read_at(read_offset, remaining).await?;
+        checked_read_progress(&mut read_offset, read, remaining.len())?;
+        remaining = remaining.split_at_mut(read).1;
     }
     let (cipher_id, mk_epoch) = extract_page_header_ids(&page)?;
     if cipher_id != context.cipher_id {

src/segment/reader.rs

@@ -14,7 +14,7 @@ use crate::pager::Pager;
 use crate::pager::format::data_page::{body, extract_page_header_ids, open_data_page};
 use crate::pager::format::page_kind::PageKind;
 use crate::vfs::types::OpenMode;
-use crate::vfs::{Vfs, VfsFile};
+use crate::vfs::{Vfs, VfsFile, checked_read_progress};
 
 use super::authenticated_metadata::{
     ExtentIndexDecodeContext, authenticate_segment_metadata, decode_extent_index,
@@ -67,11 +67,14 @@ impl<V: Vfs + Clone> SegmentReader<V> {
     ) -> Result<Self> {
         let page_size = pager.page_size();
         let live = live_path(&catalog_meta.segment_id);
-        let file = pager
-            .vfs()
-            .open(&live, OpenMode::Read)
-            .await
-            .map_err(|_| PagedbError::NotFound)?;
+        let file = match pager.vfs().open(&live, OpenMode::Read).await {
+            Ok(file) => file,
+            Err(PagedbError::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => {
+                return Err(PagedbError::NotFound);
+            }
+            Err(PagedbError::NotFound) => return Err(PagedbError::NotFound),
+            Err(error) => return Err(error),
+        };
         Self::finish_open(
             pager,
             catalog_meta,
@@ -202,9 +205,12 @@ impl<V: Vfs + Clone> SegmentReader<V> {
             .checked_mul(page_size)
             .ok_or_else(|| PagedbError::arithmetic_overflow("segment page offset"))?;
         let mut buf = vec![0u8; self.page_size];
-        let n = self.file.read_at(offset, &mut buf).await?;
-        if n < self.page_size {
-            return Err(PagedbError::NotFound);
+        let mut read_offset = offset;
+        let mut remaining = &mut buf[..];
+        while !remaining.is_empty() {
+            let read = self.file.read_at(read_offset, remaining).await?;
+            checked_read_progress(&mut read_offset, read, remaining.len())?;
+            remaining = remaining.split_at_mut(read).1;
         }
 
         // Try each segment page kind; AAD binding rejects wrong ones.

src/segment/writer.rs

@@ -16,7 +16,7 @@ use crate::pager::format::segment_footer::{
 };
 use crate::pager::format::structural_header::{SegmentHeaderFields, encode_segment_header};
 use crate::vfs::types::OpenMode;
-use crate::vfs::{Vfs, VfsFile};
+use crate::vfs::{Vfs, VfsFile, write_all_at};
 use crate::{RealmId, Result};
 use tracing;
 
@@ -89,7 +89,7 @@ impl<V: Vfs + Clone> SegmentWriter<V> {
             flags: 0,
         };
         let header_bytes = encode_segment_header(&header_fields, &hk, page_size)?;
-        file.write_at(0, &header_bytes).await?;
+        write_all_at(&mut file, 0, &header_bytes).await?;
 
         let total_bytes = u64::try_from(page_size)
             .map_err(|_| PagedbError::Io(std::io::Error::other("page_size > u64")))?;
@@ -185,7 +185,7 @@ impl<V: Vfs + Clone> SegmentWriter<V> {
         let offset = page_id
             .checked_mul(self.page_size as u64)
             .ok_or_else(|| PagedbError::Io(std::io::Error::other("offset overflow")))?;
-        self.file.write_at(offset, &buf).await?;
+        write_all_at(&mut self.file, offset, &buf).await?;
         self.next_page_id += 1;
         self.total_bytes = self.total_bytes.saturating_add(self.page_size as u64);
         Ok(page_id)
@@ -327,7 +327,7 @@ impl<V: Vfs + Clone> SegmentWriter<V> {
                 let offset = page_id
                     .checked_mul(self.page_size as u64)
                     .ok_or_else(|| PagedbError::Io(std::io::Error::other("offset overflow")))?;
-                self.file.write_at(offset, &buf).await?;
+                write_all_at(&mut self.file, offset, &buf).await?;
                 self.next_page_id += 1;
                 self.total_bytes = self.total_bytes.saturating_add(self.page_size as u64);
             }
@@ -362,7 +362,7 @@ impl<V: Vfs + Clone> SegmentWriter<V> {
         let offset = footer_page_id
             .checked_mul(self.page_size as u64)
             .ok_or_else(|| PagedbError::Io(std::io::Error::other("offset overflow")))?;
-        self.file.write_at(offset, &footer_bytes).await?;
+        write_all_at(&mut self.file, offset, &footer_bytes).await?;
         self.file.sync().await?;
         self.pager.vfs().sync_dir("seg/.staging").await?;
 

src/txn/db/misc.rs

@@ -112,20 +112,17 @@ impl<V: Vfs + Clone> Db<V> {
         let latest_commit_id = snapshot.commit_id;
 
         // Durable free-list depth (chain rooted at the header's free_list_root).
-        let free_list_pending_entries =
-            crate::pager::freelist::read_chain(&self.pager, self.realm_id, free_list_root)
-                .await
-                .map_or(0, |(entries, _)| entries.len() as u64);
+        let (free_list_entries, _) =
+            crate::pager::freelist::read_chain(&self.pager, self.realm_id, free_list_root).await?;
+        let free_list_pending_entries = free_list_entries.len() as u64;
 
         // Main database file size.
-        let main_db_bytes = match self
+        let main_db_bytes = self
             .vfs
             .open(&self.main_db_path, crate::vfs::types::OpenMode::Read)
-            .await
-        {
-            Ok(f) => f.len().await.unwrap_or(0),
-            Err(_) => 0,
-        };
+            .await?
+            .len()
+            .await?;
 
         // Buffer pool stats from cache.
         let buffer_pool_pages = { self.pager.inner.buffer_pool.lock().len() as u64 };
@@ -161,7 +158,7 @@ impl<V: Vfs + Clone> Db<V> {
             );
 
             let seg_start = vec![CatalogRowKind::Segment as u8];
-            let seg_rows = tree.scan_prefix(&seg_start).await.unwrap_or_default();
+            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()

src/txn/db/open/existing.rs

@@ -12,7 +12,7 @@ use crate::options::OpenOptions;
 use crate::pager::header::ActiveSlot;
 use crate::pager::structural_header::MainDbHeaderFields;
 use crate::pager::{Pager, PagerConfig};
-use crate::vfs::{Vfs, VfsFile};
+use crate::vfs::{Vfs, read_exact_at};
 use crate::{RealmId, Result};
 
 use super::super::super::mode::DbMode;
@@ -104,13 +104,13 @@ impl<V: Vfs + Clone> Db<V> {
         let capabilities = mode.open_capabilities();
         let file_mode = capabilities.main_db_open_mode();
         let read_only = capabilities.read_only_file_access();
-        let f = vfs.open(&main_db_path, file_mode).await?;
+        let mut f = vfs.open(&main_db_path, file_mode).await?;
         let mut buf_a = vec![0u8; page_size];
         let mut buf_b = vec![0u8; page_size];
-        let _ = f.read_at(0, &mut buf_a).await?;
+        read_exact_at(&mut f, 0, &mut buf_a).await?;
         let page_size_u64 = u64::try_from(page_size)
             .map_err(|_| PagedbError::Io(std::io::Error::other("page_size > u64")))?;
-        let _ = f.read_at(page_size_u64, &mut buf_b).await?;
+        read_exact_at(&mut f, page_size_u64, &mut buf_b).await?;
         drop(f);
 
         let try_decode = |buf: &[u8]| -> Option<(MainDbHeaderFields, bool)> {