feat(db): enforce mode-aware opens and durable identities

Author: farhan-syah

.gitignore

@@ -95,4 +95,4 @@ dmypy.json
 .gradle/
 
 .cargo/
-
+.pi

Cargo.toml

@@ -31,6 +31,7 @@ bytes = "1"
 tracing = "0.1"
 parking_lot = "0.12"
 rayon = "1"
+getrandom = { version = "0.2", features = ["js"] }
 
 # tokio's `rt-multi-thread` and `fs` features are not supported on
 # `wasm32-unknown-unknown`. Split per target so OPFS builds compile.
@@ -67,7 +68,6 @@ lz4_flex = { version = "0.13", optional = true, default-features = false, featur
 ] }
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
-getrandom = { version = "0.2", features = ["js"] }
 wasm-bindgen = { version = "0.2", optional = true }
 wasm-bindgen-futures = { version = "0.4", optional = true }
 js-sys = { version = "0.3", optional = true }

src/compaction/full.rs

@@ -106,7 +106,7 @@ async fn compact_now_inner<V: Vfs + Clone>(db: &Db<V>) -> Result<CompactStats> {
         )
         .await?;
         db.vfs.mkdir_all("seg/.staging").await?;
-        let new_segment_id = db.next_segment_id();
+        let new_segment_id = crate::crypto::random::segment_id()?;
         let mut writer = SegmentWriter::create_internal(
             db.pager.clone(),
             meta.realm_id,

src/crypto/mod.rs

@@ -6,6 +6,7 @@ pub mod kdf;
 pub mod key_manager;
 pub mod keys;
 pub mod nonce;
+pub(crate) mod random;
 
 pub use aad::{Aad, AadFields};
 pub use cipher::{Cipher, CipherId};

src/crypto/random/identity.rs

@@ -0,0 +1,42 @@
+//! Purpose-specific CSPRNG helpers for nonce-space partitioning identities.
+
+use crate::Result;
+
+fn random_bytes() -> Result<[u8; 16]> {
+    let mut bytes = [0u8; 16];
+    getrandom::getrandom(&mut bytes)?;
+    Ok(bytes)
+}
+
+fn random_nonzero_identity() -> Result<[u8; 16]> {
+    loop {
+        let bytes = random_bytes()?;
+        if bytes != [0u8; 16] {
+            return Ok(bytes);
+        }
+    }
+}
+
+pub(crate) fn database_identity() -> Result<([u8; 16], [u8; 16])> {
+    Ok((random_nonzero_identity()?, random_bytes()?))
+}
+
+pub(crate) fn segment_id() -> Result<[u8; 16]> {
+    random_nonzero_identity()
+}
+
+pub(crate) fn journal_id() -> Result<[u8; 16]> {
+    random_nonzero_identity()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn purpose_specific_identities_reserve_zero() {
+        assert_ne!(database_identity().unwrap().0, [0u8; 16]);
+        assert_ne!(segment_id().unwrap(), [0u8; 16]);
+        assert_ne!(journal_id().unwrap(), [0u8; 16]);
+    }
+}

src/crypto/random/mod.rs

@@ -0,0 +1,5 @@
+//! Cryptographically secure generation of persistent identities and salts.
+
+mod identity;
+
+pub(crate) use identity::{database_identity, journal_id, segment_id};

src/errors.rs

@@ -26,6 +26,9 @@ pub enum PagedbError {
     #[error("nonce counter exhausted (per-file 2^48 limit reached); rekey required")]
     NonceCounterExhausted,
 
+    #[error("arithmetic overflow while computing {operation}")]
+    ArithmeticOverflow { operation: &'static str },
+
     #[error("read-only handle")]
     ReadOnly,
 
@@ -117,6 +120,9 @@ pub enum PagedbError {
     #[error("unsupported by backend")]
     Unsupported,
 
+    #[error("cryptographically secure randomness unavailable: {0}")]
+    Randomness(#[from] getrandom::Error),
+
     #[error("io: {0}")]
     Io(#[from] std::io::Error),
 }
@@ -196,6 +202,12 @@ impl PagedbError {
         Self::Corruption(detail)
     }
 
+    /// Canonical constructor for arithmetic-overflow errors.
+    #[must_use]
+    pub const fn arithmetic_overflow(operation: &'static str) -> Self {
+        Self::ArithmeticOverflow { operation }
+    }
+
     /// Canonical constructor for deferred-free backlog errors.
     #[must_use]
     pub fn deferred_free_backlog(pages_pending: u64, oldest_pinning_commit: u64) -> Self {

src/pager/core.rs

@@ -4,11 +4,10 @@
 
 use std::collections::BTreeMap;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, Ordering as AtomOrd};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering as AtomOrd};
 
 use bytes::Bytes;
 use tokio::sync::Mutex as AsyncMutex;
-use tracing;
 
 use crate::crypto::aad::{AadFields, MAIN_DB_SEGMENT_ID};
 use crate::crypto::key_manager::DekLru;
@@ -41,7 +40,7 @@ pub struct PagerConfig {
     pub anchor_budget: u64,
     pub dek_lru_capacity: usize,
     /// Number of AEAD-verification retries on a cache miss before surfacing
-    /// a `ChecksumFailure`. Set to > 0 only in Observer mode to absorb torn
+    /// a `ChecksumFailure`. Set to > 0 only in `Observer` mode to absorb torn
     /// reads; all other modes keep this at 0 so that AEAD failures remain
     /// hard corruption signals.
     pub observer_retry_count: u32,
@@ -162,6 +161,7 @@ pub struct Pager<V: Vfs> {
     /// in the cache. Reads use per-page epoch routing; writes always use this
     /// atomic to pick the DEK.
     active_epoch: AtomicU64,
+    read_only: AtomicBool,
     files: AsyncMutex<BTreeMap<FileKey, Arc<AsyncMutex<V::File>>>>,
     dek_lru: parking_lot::Mutex<DekLru>,
     main_nonce: parking_lot::Mutex<MainDbNonceGen>,
@@ -172,7 +172,7 @@ pub struct Pager<V: Vfs> {
     journal_nonces: parking_lot::Mutex<BTreeMap<[u8; 16], SegmentNonceGen>>,
     pub(crate) inner: Arc<PagerInner>,
     /// Retries on AEAD failure before surfacing `ChecksumFailure`. Non-zero
-    /// only in Observer mode to absorb torn reads from a concurrent writer.
+    /// only in `Observer` mode to absorb torn reads from a concurrent writer.
     observer_retry_count: u32,
 }
 
@@ -236,13 +236,28 @@ impl<V: Vfs> Pager<V> {
             files: AsyncMutex::new(BTreeMap::new()),
             inner,
             active_epoch: AtomicU64::new(initial_epoch),
+            read_only: AtomicBool::new(false),
             mk: parking_lot::RwLock::new(mk),
             observer_retry_count,
             vfs,
             cfg,
         })
     }
 
+    /// Restrict all lazily-opened persistent files to read access and reject
+    /// pager flushes. This only changes in-memory state.
+    pub(crate) fn set_read_only(&self) {
+        self.read_only.store(true, AtomOrd::SeqCst);
+    }
+
+    /// Enable persistent writes after a frozen read-only handle transitions
+    /// to Follower mode. Cached read-only file handles are discarded so later
+    /// pager operations reopen them with read/write access.
+    pub(crate) async fn enable_write_access(&self) {
+        self.read_only.store(false, AtomOrd::SeqCst);
+        self.files.lock().await.clear();
+    }
+
     /// Atomically advance the active epoch used for flush (write) operations.
     /// Also installs a new master key. Called by `Db::rekey_db` immediately
     /// before the final page flush so all dirty pages are re-sealed under the
@@ -602,14 +617,22 @@ impl<V: Vfs> Pager<V> {
         // mk_epoch before constructing AAD and selecting the DEK.
         self.inner.record_miss(file);
         let page_size = self.cfg.page_size;
+        let page_size_u64 =
+            u64::try_from(page_size).map_err(|_| PagedbError::arithmetic_overflow("page size"))?;
+        let page_offset = page_id
+            .checked_mul(page_size_u64)
+            .ok_or_else(|| PagedbError::arithmetic_overflow("page read offset"))?;
         let file_handle = self.open_file_handle(file).await?;
 
         // Observer-mode retry loop: on AEAD failure retry up to
         // `observer_retry_count` times (10 ms backoff) to absorb torn reads
         // from a concurrent writer. In non-observer mode (retry_count == 0)
         // the loop body executes exactly once and any AEAD failure is a hard
         // corruption signal.
-        let max_attempts = self.observer_retry_count + 1;
+        let max_attempts = self
+            .observer_retry_count
+            .checked_add(1)
+            .ok_or_else(|| PagedbError::arithmetic_overflow("observer retry attempts"))?;
         let mut last_err: Option<PagedbError> = None;
         for attempt in 0..max_attempts {
             if attempt > 0 {
@@ -618,7 +641,7 @@ impl<V: Vfs> Pager<V> {
             let mut buf = vec![0u8; page_size];
             {
                 let f = file_handle.lock().await;
-                let n = f.read_at(page_id * page_size as u64, &mut buf).await?;
+                let n = f.read_at(page_offset, &mut buf).await?;
                 if n < page_size {
                     for b in &mut buf[n..] {
                         *b = 0;
@@ -690,12 +713,17 @@ impl<V: Vfs> Pager<V> {
         segment_id: [u8; 16],
         dest_path: Option<&str>,
     ) -> Result<()> {
+        if self.read_only.load(AtomOrd::SeqCst) {
+            return Err(PagedbError::ReadOnly);
+        }
         let dirty_ids = self.inner.cache_for_key(file).lock().dirty_for_file(file);
         if dirty_ids.is_empty() {
             return Ok(());
         }
 
         let page_size = self.cfg.page_size;
+        let page_size_u64 =
+            u64::try_from(page_size).map_err(|_| PagedbError::arithmetic_overflow("page size"))?;
         let flush_epoch = self.active_epoch.load(AtomOrd::SeqCst);
 
         // Serial gather: snapshot each dirty page's plaintext + kind under the
@@ -723,6 +751,15 @@ impl<V: Vfs> Pager<V> {
             prepared.push((*pid, kind, wire));
         }
 
+        // Validate every physical offset before consuming any nonce.
+        let offsets: Vec<u64> = prepared
+            .iter()
+            .map(|(pid, _, _)| {
+                pid.checked_mul(page_size_u64)
+                    .ok_or_else(|| PagedbError::arithmetic_overflow("page write offset"))
+            })
+            .collect::<Result<_>>()?;
+
         // Pre-allocate a nonce per page (counter increments — single-threaded
         // by design; cheap).
         let mut nonces: Vec<Nonce> = Vec::with_capacity(prepared.len());
@@ -773,11 +810,8 @@ impl<V: Vfs> Pager<V> {
 
         // Issue physical-id-order vectored writes.
         let mut reqs: Vec<WriteReq<'_>> = Vec::with_capacity(prepared.len());
-        for (pid, _kind, wire) in &prepared {
-            reqs.push(WriteReq {
-                offset: *pid * page_size as u64,
-                buf: wire,
-            });
+        for ((_, _kind, wire), offset) in prepared.iter().zip(offsets) {
+            reqs.push(WriteReq { offset, buf: wire });
         }
         if let Some(path) = dest_path {
             // Alternate destination (compaction's compacted copy): open it
@@ -801,21 +835,33 @@ impl<V: Vfs> Pager<V> {
     }
 
     async fn open_file_handle(&self, file: FileKey) -> Result<Arc<AsyncMutex<V::File>>> {
-        let mut files = self.files.lock().await;
-        if let Some(h) = files.get(&file) {
-            return Ok(h.clone());
+        let cached = {
+            let files = self.files.lock().await;
+            files.get(&file).cloned()
+        };
+        if let Some(handle) = cached {
+            return Ok(handle);
         }
+
         let path = match file {
             FileKey::Main => self.cfg.main_db_path.clone(),
             FileKey::Segment(id) => format!("seg/{}", crate::hex::to_hex_lower(&id)),
             FileKey::ApplyJournal(id) => {
                 format!("applyjournal/{}", crate::hex::to_hex_lower(&id))
             }
         };
-        let f = self.vfs.open(&path, OpenMode::CreateOrOpen).await?;
-        let arc = Arc::new(AsyncMutex::new(f));
-        files.insert(file, arc.clone());
-        Ok(arc)
+        let mode = if self.read_only.load(AtomOrd::SeqCst) {
+            OpenMode::Read
+        } else {
+            OpenMode::CreateOrOpen
+        };
+        let opened = Arc::new(AsyncMutex::new(self.vfs.open(&path, mode).await?));
+        let mut files = self.files.lock().await;
+        if let Some(handle) = files.get(&file) {
+            return Ok(handle.clone());
+        }
+        files.insert(file, opened.clone());
+        Ok(opened)
     }
 
     fn next_nonce_for_flush(&self, file: FileKey) -> Result<Nonce> {

src/recovery/mod.rs

@@ -10,4 +10,4 @@ pub use deep_walk::{DeepWalkReport, run_deep_walk};
 pub use journal::{
     ApplyJournalRecord, JournalAction, execute_journal_actions, replay_apply_journal,
 };
-pub use reconcile::reconcile_catalog;
+pub use reconcile::{repair_catalog, verify_catalog};