refactor(errors): unify handle-mode gating and corruption reporting

Replace the scattered ad hoc mode checks (ReadOnly/Unsupported/IdentityForked
returned inconsistently from begin_write, create_segment, rekey_db,
compaction, and apply_incremental) with a single PagedbError::WrongMode
variant and a DbModeCapabilities matrix that every gate resolves against
through Db::require_mode, so an embedder gets one variant naming both the
attempted operation and the mode that would have worked.

Fold FooterUnverifiable and StagingMissing down to the identity they can
actually authenticate (drop the unreliable embedder-visible name), add
PageUnverifiable so a per-page AEAD failure names the failing page instead of
collapsing into a bare ChecksumFailure, and drop the never-raised
FreeListExhausted/SegmentTombstoneStalled variants. Classify std::io::Error
at the one boundary every VFS call crosses so device exhaustion surfaces as
NoSpace instead of hiding inside a generic Io error. Add a RichlyReported
diagnostics guard so a site-specific corruption capture is not duplicated by
the generic constructor-level one.

Db::open (bootstrap-or-reopen decided from what is actually on disk) becomes
the only production entry point; the open_internal*/open_existing* family
that raced or skipped that decision moves behind #[cfg(test)]. OpenOptions
gains a cipher selector for stores being created. Tighten apply_incremental's
visibility window so reader admission stays closed continuously from the
tombstone pin scan through snapshot publication, and add one-shot fault
injection for the nonce-anchor commit/refresh path and two more rekey resume
points, with tests exercising each interruption.

Update every bench, benchmark, and integration test to the new API and error
shapes.

Author: farhan-syah

benches/authenticated_node_read.rs

@@ -3,9 +3,11 @@
 //!
 //! Every timed lookup descends a multi-level tree with an empty main-page
 //! cache, so each internal and leaf node costs a real decrypt plus the
-//! envelope/body kind agreement check in `BTree::read_node_guard`. The cache
-//! eviction that creates those conditions is deliberately *outside* the timed
-//! region — measuring it would charge cache bookkeeping to the read path.
+//! envelope/body kind agreement check in `BTree::read_node_guard`. The store is
+//! built once; each iteration opens its own handle over those same bytes, and a
+//! fresh handle starts with an empty buffer pool. Opening and retiring handles
+//! is deliberately *outside* the timed region — measuring it would charge
+//! handle setup to the read path.
 //!
 //! Runs on `MemVfs`: the figure is CPU + AEAD for an authenticated cold
 //! descent, not the cost of reaching real storage.
@@ -28,60 +30,85 @@ use pagedb::{Db, RealmId};
 const PAGE: usize = 4096;
 const KEY_COUNT: usize = 1_000;
 const REALM: RealmId = RealmId::new([0x51; 16]);
+const KEK: [u8; 32] = [0xA5; 32];
 const VALUE: &[u8] = b"authenticated-node-read";
 
 thread_local! {
-    /// Built once and reused: this workload only reads, so rebuilding the tree
-    /// per iteration would add setup noise without changing what is measured.
-    static DB: RefCell<Option<Rc<Db<MemVfs>>>> = const { RefCell::new(None) };
+    /// The populated store's bytes, built once: this workload only reads, so
+    /// rebuilding the tree per iteration would add setup noise without changing
+    /// what is measured.
+    static STORE: RefCell<Option<MemVfs>> = const { RefCell::new(None) };
 }
 
 fn key(index: usize) -> Vec<u8> {
     format!("node:{index:08}").into_bytes()
 }
 
-fn shared_db() -> Rc<Db<MemVfs>> {
-    DB.with(|cell| {
+fn populated_store() -> MemVfs {
+    STORE.with(|cell| {
         if cell.borrow().is_none() {
-            let db = block_on(async {
+            let vfs = MemVfs::new();
+            block_on(async {
                 // Commit history is irrelevant here and would only add
                 // unrelated catalog writes to the setup.
                 let options =
                     OpenOptions::default().with_commit_history_retain(RetainPolicy::Disabled);
-                let db =
-                    Db::open_internal_with_options(MemVfs::new(), [0xA5; 32], PAGE, REALM, options)
-                        .await
-                        .expect("open bench store");
+                let db = Db::open(vfs.clone(), KEK, PAGE, REALM, options)
+                    .await
+                    .expect("open bench store");
                 let mut write = db.begin_write().await.expect("begin txn");
                 for index in 0..KEY_COUNT {
                     write.put(&key(index), VALUE).await.expect("insert");
                 }
                 write.commit().await.expect("commit");
-                db
+                // The writer handle drops with this block, releasing its
+                // sentinel so the read-only handles below can attach.
             });
-            *cell.borrow_mut() = Some(Rc::new(db));
+            *cell.borrow_mut() = Some(vfs);
         }
-        cell.borrow().as_ref().expect("db initialised").clone()
+        cell.borrow().as_ref().expect("store initialised").clone()
+    })
+}
+
+/// A handle whose buffer pool is empty, over an already-populated store.
+fn cold_handle(store: &MemVfs) -> Db<MemVfs> {
+    block_on(async {
+        Db::open_read_only(store.clone(), KEK, PAGE, REALM, OpenOptions::default())
+            .await
+            .expect("open cold read-only handle")
     })
 }
 
 #[bench(group = "pager/authenticated-node-read")]
 fn cold_tree_get(b: &mut Bencher) {
-    let db = shared_db();
+    let store = populated_store();
+    // `Rc` so the timed region can lift the handle out of the cell *before* it
+    // awaits: a `RefCell` borrow held across an await point is what
+    // `clippy::await_holding_refcell_ref` rejects, and cloning the `Rc` costs a
+    // refcount bump rather than moving the handle's teardown into the
+    // measurement.
+    let handle: RefCell<Option<Rc<Db<MemVfs>>>> = RefCell::new(None);
     let mut index = 0usize;
 
     b.iter_with_setup(
         || {
-            // Untimed: drop the warm pages and pre-build the lookup key, so the
-            // timed region is descent plus authentication only.
+            // Untimed: retire the previous iteration's warm handle, open a cold
+            // one, and pre-build the lookup key, so the timed region is descent
+            // plus authentication only.
             let lookup_key = key(index % KEY_COUNT);
             index = index.wrapping_add(1);
-            db.evict_main_pages(REALM);
+            handle.borrow_mut().take();
+            *handle.borrow_mut() = Some(Rc::new(cold_handle(&store)));
             lookup_key
         },
         |lookup_key| {
             with_rt(|rt| {
                 rt.block_on(async {
+                    let db = handle
+                        .borrow()
+                        .as_ref()
+                        .map(Rc::clone)
+                        .expect("handle opened in setup");
                     let read = db.begin_read().await.expect("begin read");
                     read.get(&lookup_key).await.expect("get")
                 })

benches/compaction.rs

@@ -14,7 +14,7 @@ use fluxbench::bench;
 use fluxbench::prelude::*;
 
 use pagedb::vfs::memory::MemVfs;
-use pagedb::{Db, RealmId};
+use pagedb::{Db, OpenOptions, RealmId};
 
 const PAGE: usize = 4096;
 const KEK: [u8; 32] = [7; 32];
@@ -33,7 +33,7 @@ fn prepared_dense_repack() -> Db<MemVfs> {
         // Retire the previous iteration's store here, where it is not measured.
         drain_parked();
 
-        let db = Db::open_internal(MemVfs::new(), KEK, PAGE, REALM)
+        let db = Db::open(MemVfs::new(), KEK, PAGE, REALM, OpenOptions::default())
             .await
             .expect("open bench store");
         let value = [0x2A; VALUE_LEN];

benches/segment.rs

@@ -18,7 +18,7 @@ use fluxbench::{bench, compare, synthetic};
 use tokio::sync::Mutex as AsyncMutex;
 
 use pagedb::vfs::memory::MemVfs;
-use pagedb::{Db, RealmId, SegmentKind, SegmentPageKind};
+use pagedb::{Db, OpenOptions, RealmId, SegmentKind, SegmentPageKind};
 
 const PAGE: usize = 4096;
 /// Payload bytes per page (must fit inside `page_size - envelope_overhead`).
@@ -35,9 +35,15 @@ fn shared_db() -> Arc<AsyncMutex<Db<MemVfs>>> {
         if cell.borrow().is_none() {
             let db = with_rt(|rt| {
                 rt.block_on(async {
-                    Db::open_internal(MemVfs::new(), [9u8; 32], PAGE, RealmId::new([1; 16]))
-                        .await
-                        .unwrap()
+                    Db::open(
+                        MemVfs::new(),
+                        [9u8; 32],
+                        PAGE,
+                        RealmId::new([1; 16]),
+                        OpenOptions::default(),
+                    )
+                    .await
+                    .unwrap()
                 })
             });
             *cell.borrow_mut() = Some(Arc::new(AsyncMutex::new(db)));

benchmarks/engine-comparison/btree.rs

@@ -57,13 +57,12 @@ thread_local! {
 }
 
 async fn open_mem(cipher: CipherId, seed: u8) -> SharedDb<MemVfs> {
-    let db = Db::open_internal_with_options_and_cipher(
+    let db = Db::open(
         MemVfs::new(),
         [seed; 32],
         PAGE,
         RealmId::new([seed; 16]),
-        bench_opts(),
-        cipher,
+        bench_opts().with_cipher(cipher),
     )
     .await
     .unwrap();
@@ -73,13 +72,12 @@ async fn open_mem(cipher: CipherId, seed: u8) -> SharedDb<MemVfs> {
 async fn open_file(cipher: CipherId, seed: u8) -> (SharedDb<TokioVfs>, tempfile::TempDir) {
     let dir = tempfile::TempDir::new().unwrap();
     let vfs = TokioVfs::new(dir.path());
-    let db = Db::open_internal_with_options_and_cipher(
+    let db = Db::open(
         vfs,
         [seed; 32],
         PAGE,
         RealmId::new([seed; 16]),
-        bench_opts(),
-        cipher,
+        bench_opts().with_cipher(cipher),
     )
     .await
     .unwrap();

benchmarks/engine-comparison/comparison.rs

@@ -94,7 +94,7 @@ fn open_pagedb_fresh() -> PagedbBench {
         .with_anchor_budget(10_000_000);
     let db = with_rt(|rt| {
         rt.block_on(async {
-            Db::open_internal_with_options(vfs, [0xAB; 32], 4096, RealmId::new([1; 16]), opts)
+            Db::open(vfs, [0xAB; 32], 4096, RealmId::new([1; 16]), opts)
                 .await
                 .unwrap()
         })

src/compaction/full.rs

@@ -35,9 +35,11 @@ pub async fn compact_now<V: Vfs + Clone>(db: &Db<V>) -> Result<CompactStats> {
 #[allow(clippy::too_many_lines)]
 async fn compact_now_inner<V: Vfs + Clone>(db: &Db<V>) -> Result<CompactStats> {
     db.ensure_usable()?;
-    if !matches!(db.mode, crate::txn::mode::DbMode::Standalone) {
-        return Err(PagedbError::Unsupported);
-    }
+    db.require_mode(
+        "compaction",
+        crate::txn::mode::DbMode::Standalone,
+        crate::txn::db::DbModeCapabilities::allows_store_maintenance,
+    )?;
 
     let mut result = CompactStats::default();
 

src/compaction/step.rs

@@ -24,7 +24,7 @@ use super::types::{CompactBudget, CompactProgress};
 /// be safely chunked). Compaction runs atomically to completion and always
 /// returns `more_work = false`; to reclaim periodically, call this again later.
 ///
-/// Returns `PagedbError::Unsupported` if the handle is not in `Standalone` mode.
+/// Returns `PagedbError::WrongMode` if the handle is not in `Standalone` mode.
 pub async fn compact_step<V: Vfs + Clone>(
     db: &Db<V>,
     budget: CompactBudget,

src/diag.rs

@@ -15,9 +15,57 @@
 
 #[cfg(all(feature = "diagnostics", not(target_arch = "wasm32")))]
 mod imp {
+    use std::cell::Cell;
+    use std::marker::PhantomData;
+
     use crate::RealmId;
     use crate::errors::CorruptionDetail;
 
+    thread_local! {
+        /// Live [`RichlyReported`] guards on this thread. A depth rather than a
+        /// flag: nesting must not leave the suppression stuck on when the inner
+        /// guard drops.
+        static RICH_REPORTS_IN_FLIGHT: Cell<u32> = const { Cell::new(0) };
+    }
+
+    /// Evidence that a site-specific report already covers the corruption about
+    /// to be constructed.
+    ///
+    /// Every rich capture in this module returns one. Bind it across the
+    /// `PagedbError::corruption(...)` call that reports the same failure and
+    /// [`corruption_captured`] stands down for that construction, so one
+    /// failure files one report — the site's, which carries forensics (the
+    /// failing page, the fsck hint, the store path) that the constructor, which
+    /// sees only a `CorruptionDetail`, cannot know.
+    ///
+    /// This is the mechanism, not a special case: a second site that grows its
+    /// own `DomainContext` gets the same suppression by returning this type
+    /// from its capture function, with nothing to rediscover.
+    ///
+    /// Deliberately not `Send`. The guard covers the construction immediately
+    /// following it and nothing else; holding it across an `.await` would
+    /// silence unrelated corruption raised by whatever ran in between, and the
+    /// compiler rejects that wherever the surrounding future must be `Send`.
+    #[must_use = "bind the guard so it outlives the corruption() call it covers"]
+    pub struct RichlyReported {
+        _not_send: PhantomData<*const ()>,
+    }
+
+    impl RichlyReported {
+        fn begin() -> Self {
+            RICH_REPORTS_IN_FLIGHT.with(|depth| depth.set(depth.get().saturating_add(1)));
+            Self {
+                _not_send: PhantomData,
+            }
+        }
+    }
+
+    impl Drop for RichlyReported {
+        fn drop(&mut self) {
+            RICH_REPORTS_IN_FLIGHT.with(|depth| depth.set(depth.get().saturating_sub(1)));
+        }
+    }
+
     /// Breadcrumb: a store was (re)opened — marks epoch boundaries in the trail,
     /// the dimension along which freed-page use-after-free surfaces.
     pub fn reopened(latest_commit: u64) {
@@ -98,13 +146,18 @@ mod imp {
     /// authenticate. Records the failing page, expected binding, and realm so
     /// the failure is diagnosable from the report alone; the host application
     /// preserves the store bytes (it owns the real path behind the VFS).
+    ///
+    /// Returns a [`RichlyReported`] guard: bind it across the
+    /// `PagedbError::corruption(...)` that turns this same failure into an
+    /// error, so the caller gets the precise variant while this report — not
+    /// the constructor's generic one — is what gets filed.
     pub fn page_read_verify_failed(
         main_db_path: &str,
         page_id: u64,
         file: &str,
         binding: &str,
         realm: &RealmId,
-    ) {
+    ) -> RichlyReported {
         let ctx = PageReadVerifyFailure {
             page_id,
             file: file.to_owned(),
@@ -119,6 +172,7 @@ mod imp {
         .domain(&ctx)
         .with_backtrace()
         .emit();
+        RichlyReported::begin()
     }
 
     /// Forensic context for a [`CorruptionDetail`] captured at the moment
@@ -268,6 +322,12 @@ mod imp {
     /// reporting layer instead of only the one AEAD failure site this module
     /// originally covered.
     pub fn corruption_captured(detail: &CorruptionDetail) {
+        // A site-specific capture is already in flight for this exact failure
+        // and carries strictly more than this one could. Filing both would mean
+        // two reports for one event, the second of them less useful.
+        if RICH_REPORTS_IN_FLIGHT.with(Cell::get) > 0 {
+            return;
+        }
         let (kind, grouping_key) = classify(detail);
         let ctx = CorruptionConstructed {
             kind,
@@ -288,6 +348,10 @@ mod imp {
 mod imp {
     use crate::RealmId;
 
+    /// No-op counterpart of the recording build's suppression guard, so call
+    /// sites bind the same value under every cfg.
+    pub struct RichlyReported;
+
     pub fn reopened(_latest_commit: u64) {}
     pub fn committed(_commit_id: u64, _freed_pages: usize) {}
     pub fn lock_acquired(_mode: &str, _path: &str) {}
@@ -299,12 +363,16 @@ mod imp {
         _file: &str,
         _binding: &str,
         _realm: &RealmId,
-    ) {
+    ) -> RichlyReported {
+        RichlyReported
     }
 
     pub fn corruption_captured(_detail: &crate::errors::CorruptionDetail) {}
 }
 
+// `RichlyReported` is deliberately not re-exported: a call site binds it as the
+// return value of a rich capture (`let _report = diag::page_read_verify_failed(…)`)
+// and never names the type, so exporting it would only be an unused path.
 pub use imp::{
     committed, corruption_captured, flushed, lock_acquired, lock_rejected, page_read_verify_failed,
     reopened,