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

CHANGELOG.md

@@ -29,7 +29,9 @@ version, since none exists yet.
 - **Cross-platform VFS** — Linux (`io_uring`), Windows (IOCP), macOS/iOS
   (Grand Central Dispatch), Android, WASM/OPFS, and WASI backends, plus a
   tokio thread-pool fallback and an in-memory backend, with format-bit identity
-  across targets.
+  across targets. Backends may complete a positional read or write in several
+  calls; PageDB finishes the caller's whole buffer before treating authenticated
+  metadata as present or durable.
 - **Snapshots** — `snapshot_to`, `restore_from`, and incremental apply, each
   authenticated against the exact state its manifest describes. Destinations
   must be empty; malformed or incomplete artifacts fail closed.
@@ -38,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/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/errors.rs

@@ -169,6 +169,19 @@ pub enum PagedbError {
     #[error("recorded rekey replacement segment {replacement_segment_id:?} is missing or invalid")]
     RekeyReplacementMissing { replacement_segment_id: [u8; 16] },
 
+    /// A VFS backend reported positional-I/O progress that cannot be true —
+    /// more bytes transferred than the caller's remaining buffer.
+    ///
+    /// Not corruption: nothing on disk is known to be wrong. The backend broke
+    /// the [`VfsFile`](crate::vfs::VfsFile) contract, so the reported count
+    /// cannot be reasoned about at all and the transfer stops instead of
+    /// advancing by a length it cannot trust.
+    #[error("vfs backend violated the {operation} contract: {detail}")]
+    VfsContractViolated {
+        operation: &'static str,
+        detail: &'static str,
+    },
+
     #[error("unsupported by backend")]
     Unsupported,
 
@@ -473,6 +486,13 @@ impl PagedbError {
         Self::ArithmeticOverflow { operation }
     }
 
+    /// Canonical constructor for a VFS backend that broke the positional-I/O
+    /// contract.
+    #[must_use]
+    pub const fn vfs_contract_violated(operation: &'static str, detail: &'static str) -> Self {
+        Self::VfsContractViolated { operation, detail }
+    }
+
     /// Canonical constructor for a rekey admission that needs both KEKs.
     #[must_use]
     pub const fn rekey_resume_key_required(source_epoch: u64, target_epoch: u64) -> Self {

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,14 @@ 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;
-                    }
-                }
+                // Returned rather than retried, unlike the AEAD failures below.
+                // The transient this loop absorbs is a *torn* read — a
+                // full-length buffer of mixed old and new bytes — which is why
+                // retrying it can succeed. A transfer that ends short or errors
+                // is not that condition, and retrying it would only mask a
+                // one-shot backend fault.
+                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.
@@ -82,6 +82,28 @@ pub async fn bootstrap_header<V: Vfs>(
 /// Open an existing main.db, return the active header fields and which slot
 /// they came from.
 ///
+/// Read one header slot into `buf`, treating an incomplete slot as an
+/// unverifiable one rather than as a fatal error.
+///
+/// The A/B protocol survives one unusable slot by design, and a file truncated
+/// mid-slot is exactly that case: the surviving copy must still open the
+/// database. `buf` keeps whatever prefix was transferred, so the caller's
+/// ordinary decode rejects it on the magic or HK-MAC check like any other
+/// damaged slot. Every other backend error propagates unchanged — only
+/// end-of-file is absence.
+pub(crate) async fn read_header_slot<F: VfsFile + ?Sized>(
+    file: &mut F,
+    offset: u64,
+    buf: &mut [u8],
+) -> Result<()> {
+    match read_exact_at(file, offset, buf).await {
+        Err(PagedbError::Io(ref error)) if error.kind() == std::io::ErrorKind::UnexpectedEof => {
+            Ok(())
+        }
+        other => other,
+    }
+}
+
 /// Reads both slots; verifies each via HK-MAC; picks the one with the
 /// greater `seq`. If only one verifies, it wins. If neither verifies, returns
 /// `Corruption(HeaderUnverifiable)` — unrecoverable from inside the header
@@ -92,13 +114,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_header_slot(&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_header_slot(&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 +161,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, read_exact_at_borrowed};
 
 use super::types::{EXTENT_INDEX_ENTRY_LEN, ExtentIndexEntry};
 use super::writer::{live_path, staging_path};
@@ -85,18 +84,12 @@ 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"));
-    }
+    read_exact_at_borrowed!(file, 0, &mut header_bytes[..])?;
     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"));
-    }
+    read_exact_at_borrowed!(file, footer_offset, &mut footer_bytes[..])?;
     let (footer, manifest) = {
         let mut lru = pager.dek_lru().lock();
         let cipher = lru.get_or_derive(meta.realm_id, meta.mk_epoch, cipher_id, &master_key)?;
@@ -332,9 +325,7 @@ 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"));
-    }
+    read_exact_at_borrowed!(context.file, offset, &mut page[..])?;
     let (cipher_id, mk_epoch) = extract_page_header_ids(&page)?;
     if cipher_id != context.cipher_id {
         return Err(PagedbError::segment_metadata_mismatch(

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, read_exact_at_borrowed};
 
 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,
@@ -225,10 +228,7 @@ 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);
-        }
+        read_exact_at_borrowed!(self.file, offset, &mut buf[..])?;
 
         // Try each segment page kind; AAD binding rejects wrong ones.
         let try_kinds = [

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?;