fix(db): publish durable commits atomically

Author: farhan-syah

.gitignore

@@ -96,3 +96,4 @@ dmypy.json
 
 .cargo/
 .pi
+.agents

src/catalog/codec.rs

@@ -26,22 +26,7 @@ pub enum CatalogRowKind {
     // 0x04 and 0x05 are reserved: they were the in-catalog free-list and
     // deferred-free queue, superseded by the durable free-list chain rooted in
     // the A/B header (see `crate::pager::freelist`). Do not reuse these bytes.
-    /// Durable reader pin. One row per active cross-process read transaction.
-    /// Key: `[0x06] || pid_u32_be[4] || lease_id_u64_be[8]` (13 bytes).
-    /// Value: `commit_id[8] || root_page_id[8] || catalog_root_page_id[8] ||
-    ///          free_list_root_page_id[8] || expires_unix_seconds[8] || flags[1]`
-    /// (41 bytes).
-    ///
-    /// On `begin_read` the writer inserts a row; on `ReadTxn::drop` the row is
-    /// deleted. If a reader crashes, its row is cleaned up at the next
-    /// `Db::open` of a writer handle. A row whose `expires_unix_seconds` is
-    /// older than the current wall-clock time is treated as expired by GC and
-    /// does not block page reclamation.
-    ///
-    /// On a `ReadOnly` or `Follower` handle that cannot write to its own
-    /// catalog, reader pins are maintained in-memory only and the writer process
-    /// must be trusted to honor the catalog pins.
-    ReaderPin = 0x06,
+    // 0x06 is reserved, deliberately uninterpreted, and must never be reused.
     /// Reserved (`0x07`). Older builds wrote an incremental-compaction watermark
     /// here; compaction is now a single atomic operation and never writes it.
     /// Retained as a row-kind boundary and so any legacy row is recognised and
@@ -112,20 +97,6 @@ pub struct RealmQuotas {
     pub max_segment_bytes: Option<u64>,
 }
 
-/// Value for a durable reader-pin row.
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct ReaderPinValue {
-    pub commit_id: u64,
-    pub root_page_id: u64,
-    pub catalog_root_page_id: u64,
-    pub free_list_root_page_id: u64,
-    pub expires_unix_seconds: u64,
-    pub flags: u8,
-}
-
-pub const READER_PIN_VALUE_LEN: usize = 41;
-pub const READER_PIN_KEY_LEN: usize = 13;
-
 pub const SEGMENT_META_LEN: usize = 94;
 pub const REALM_QUOTAS_LEN: usize = 33;
 
@@ -191,63 +162,6 @@ impl Catalog {
         })
     }
 
-    /// Reader-pin row key: `[0x06] || pid_u32_be[4] || lease_id_u64_be[8]`.
-    #[must_use]
-    pub fn reader_pin_key(pid: u32, lease_id: u64) -> [u8; READER_PIN_KEY_LEN] {
-        let mut k = [0u8; READER_PIN_KEY_LEN];
-        k[0] = CatalogRowKind::ReaderPin as u8;
-        k[1..5].copy_from_slice(&pid.to_be_bytes());
-        k[5..13].copy_from_slice(&lease_id.to_be_bytes());
-        k
-    }
-
-    /// Range start key for all reader-pin rows: `[0x06]`.
-    #[must_use]
-    pub fn reader_pin_range_start() -> [u8; 1] {
-        [CatalogRowKind::ReaderPin as u8]
-    }
-
-    /// Range end key (exclusive) for all reader-pin rows: `[0x07]`.
-    #[must_use]
-    pub fn reader_pin_range_end() -> [u8; 1] {
-        [CatalogRowKind::CompactionState as u8]
-    }
-
-    /// Encode a `ReaderPinValue` as 41 bytes.
-    #[must_use]
-    pub fn encode_reader_pin(v: &ReaderPinValue) -> [u8; READER_PIN_VALUE_LEN] {
-        let mut o = [0u8; READER_PIN_VALUE_LEN];
-        o[0..8].copy_from_slice(&v.commit_id.to_le_bytes());
-        o[8..16].copy_from_slice(&v.root_page_id.to_le_bytes());
-        o[16..24].copy_from_slice(&v.catalog_root_page_id.to_le_bytes());
-        o[24..32].copy_from_slice(&v.free_list_root_page_id.to_le_bytes());
-        o[32..40].copy_from_slice(&v.expires_unix_seconds.to_le_bytes());
-        o[40] = v.flags;
-        o
-    }
-
-    /// Decode a `ReaderPinValue` from a 41-byte slice.
-    pub fn decode_reader_pin(bytes: &[u8]) -> Result<ReaderPinValue> {
-        if bytes.len() != READER_PIN_VALUE_LEN {
-            return Err(PagedbError::corruption(
-                crate::errors::CorruptionDetail::HeaderUnverifiable,
-            ));
-        }
-        let read_u64 = |off: usize| {
-            let mut b = [0u8; 8];
-            b.copy_from_slice(&bytes[off..off + 8]);
-            u64::from_le_bytes(b)
-        };
-        Ok(ReaderPinValue {
-            commit_id: read_u64(0),
-            root_page_id: read_u64(8),
-            catalog_root_page_id: read_u64(16),
-            free_list_root_page_id: read_u64(24),
-            expires_unix_seconds: read_u64(32),
-            flags: bytes[40],
-        })
-    }
-
     /// Counter row key: `[0x02] || name_bytes`. Rejects names longer than
     /// `MAX_SEGMENT_NAME_LEN`. Counter rows are per-`Db`, not per-realm.
     pub fn counter_key(name: &[u8]) -> Result<Vec<u8>> {

src/catalog/mod.rs

@@ -4,4 +4,4 @@
 
 pub mod codec;
 
-pub use codec::{Catalog, CatalogRowKind, ReaderPinValue, RekeyStateRow};
+pub use codec::{Catalog, CatalogRowKind, RekeyStateRow};

src/compaction/full.rs

@@ -33,6 +33,7 @@ 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);
     }
@@ -41,6 +42,12 @@ async fn compact_now_inner<V: Vfs + Clone>(db: &Db<V>) -> Result<CompactStats> {
 
     // Acquire exclusive writer lock for the entire compact operation.
     let mut state = db.writer.lock().await;
+    db.ensure_usable()?;
+    #[cfg(test)]
+    db.notify_writer_waiting();
+    // Keep reader admission closed from the pin scan through the main-file
+    // swap, directory sync, and published replacement snapshot.
+    let visibility_guard = db.visibility_gate.write().await;
 
     // Compaction relocates and/or truncates pages, invalidating every page id
     // cached for runtime reuse. Drop those reuse hints so a post-compaction
@@ -49,23 +56,13 @@ async fn compact_now_inner<V: Vfs + Clone>(db: &Db<V>) -> Result<CompactStats> {
 
     // ── 1. Refuse while readers are pinned ───────────────────────────────────
     // A dense repack relocates the current tree and truncates the file; pinned
-    // readers (in-process or cross-process durable) still reference the old
-    // pages, so neither is safe under them. Runtime free-page reuse already
+    // in-process readers still reference the old pages, so neither is safe
+    // under them. Runtime free-page reuse already
     // reclaims space on ordinary commits, so there is nothing for compaction to
     // do here until the readers drop.
     let has_readers = {
-        let in_mem_min = {
-            let readers = db.tracked_readers.lock();
-            readers
-                .iter()
-                .map(|r| r.commit_id.0)
-                .min()
-                .unwrap_or(u64::MAX)
-        };
-        let durable_min = db
-            .min_durable_reader_commit(state.catalog_root_page_id, state.next_page_id)
-            .await;
-        in_mem_min.min(durable_min) < u64::MAX
+        let readers = db.tracked_readers.lock();
+        !readers.is_empty()
     };
     if has_readers {
         return Ok(result);
@@ -78,7 +75,7 @@ async fn compact_now_inner<V: Vfs + Clone>(db: &Db<V>) -> Result<CompactStats> {
     // atomic rename; main.db is never modified until the rename (see
     // `super::repack`).
     if state.free_list_root_page_id != 0 {
-        let repack = super::repack::atomic_dense_repack(db, &mut state).await?;
+        let repack = super::repack::atomic_dense_repack(db, &mut state, &visibility_guard).await?;
         result.main_db_pages_reclaimed = repack.pages_reclaimed;
         result.bytes_truncated = repack.bytes_truncated;
     }
@@ -131,7 +128,15 @@ async fn compact_now_inner<V: Vfs + Clone>(db: &Db<V>) -> Result<CompactStats> {
         let new_meta = writer.seal().await?;
         let seg_name =
             find_segment_name_inner(&db.pager, db.realm_id, &state, &meta.segment_id).await?;
-        replace_segment_compact(db, &mut state, &seg_name, &meta.segment_id, &new_meta).await?;
+        replace_segment_compact(
+            db,
+            &mut state,
+            &visibility_guard,
+            &seg_name,
+            &meta.segment_id,
+            &new_meta,
+        )
+        .await?;
         result.segments_repacked += 1;
     }
 

src/compaction/helpers.rs

@@ -9,6 +9,7 @@ use crate::errors::PagedbError;
 use crate::pager::header::commit_header;
 use crate::pager::structural_header::MainDbHeaderFields;
 use crate::txn::db::{Db, WriterState};
+use crate::txn::write::SegmentSideEffect;
 use crate::vfs::Vfs;
 use crate::{CommitId, Result};
 
@@ -102,6 +103,7 @@ pub(super) async fn find_segment_name_inner<V: Vfs + Clone>(
 pub(super) async fn replace_segment_compact<V: Vfs + Clone>(
     db: &Db<V>,
     state: &mut WriterState,
+    visibility: &tokio::sync::RwLockWriteGuard<'_, ()>,
     name: &str,
     old_segment_id: &[u8; 16],
     new_meta: &SegmentMeta,
@@ -164,34 +166,33 @@ pub(super) async fn replace_segment_compact<V: Vfs + Clone>(
         db.page_size,
     )
     .await?;
-    db.pager.commit_anchor(counter_anchor)?;
-
-    // Promote staging file to live.
-    db.vfs.mkdir_all("seg").await?;
-    let staging = crate::segment::writer::staging_path(&new_meta.segment_id);
-    let live = crate::segment::writer::live_path(&new_meta.segment_id);
-    db.vfs.rename(&staging, &live).await?;
-    db.vfs.sync_dir("seg").await.ok();
-
-    // Tombstone old segment.
-    let old_live = crate::segment::writer::live_path(old_segment_id);
-    let tomb = format!(
-        "seg/.tombstone/{}.{}",
-        crate::hex::to_hex_lower(old_segment_id),
-        new_commit_id,
-    );
-    db.vfs.mkdir_all("seg/.tombstone").await?;
-    db.vfs.rename(&old_live, &tomb).await.ok();
-    db.vfs.sync_dir("seg/.tombstone").await.ok();
-
+    // The catalog is durable at this point. Preserve its state internally but
+    // retain the prior reader snapshot until segment replacement is reconciled.
     state.catalog_root_page_id = new_cat_root;
+    state.catalog_root_txn_id = new_commit_id;
     state.next_page_id = new_next;
     state.active_slot = new_slot;
     state.seq = new_seq;
     state.latest_commit_id = new_commit_id;
-    db.latest_commit
-        .store(new_commit_id, std::sync::atomic::Ordering::SeqCst);
 
+    let effects = [
+        SegmentSideEffect::Tombstone {
+            segment_id: *old_segment_id,
+            tombstone_commit_id: None,
+        },
+        SegmentSideEffect::Promote {
+            segment_id: new_meta.segment_id,
+        },
+    ];
+    let _ = db
+        .finish_durable_commit_visible(
+            visibility,
+            state,
+            CommitId(new_commit_id),
+            counter_anchor,
+            &effects,
+        )
+        .await?;
     Ok(())
 }
 

src/compaction/repack.rs

@@ -46,13 +46,14 @@ fn scratch_path(db: &Db<impl Vfs + Clone>) -> String {
 pub(super) async fn atomic_dense_repack<V: Vfs + Clone>(
     db: &Db<V>,
     state: &mut WriterState,
+    visibility: &tokio::sync::RwLockWriteGuard<'_, ()>,
 ) -> Result<RepackOutcome> {
     let scratch = scratch_path(db);
     // Remove any leftover scratch from an earlier interrupted compaction.
     db.vfs.remove(&scratch).await.ok();
 
     match build_scratch(db, state, &scratch).await {
-        Ok(pending) => commit_swap(db, state, &scratch, pending).await,
+        Ok(pending) => commit_swap(db, state, visibility, &scratch, pending).await,
         Err(e) => {
             // Nothing was written to main.db; drop the never-persisted compacted
             // pages from the cache so the old tree is read back from disk, and
@@ -166,32 +167,52 @@ pub(super) async fn build_scratch<V: Vfs + Clone>(
 async fn commit_swap<V: Vfs + Clone>(
     db: &Db<V>,
     state: &mut WriterState,
+    visibility: &tokio::sync::RwLockWriteGuard<'_, ()>,
     scratch: &str,
     pending: PendingSwap,
 ) -> Result<RepackOutcome> {
     // Close the cached main.db handle so the rename can replace the file
     // (Windows) and the next access reopens the new inode (Unix).
     db.pager.close_main_handle().await;
-    db.vfs.rename(scratch, &db.main_db_path).await?;
-    db.vfs.sync_dir("/").await.ok();
-
-    db.pager.commit_anchor(pending.counter_anchor)?;
-    // The cached pages were sealed for the scratch file (now main.db); drop them
-    // so reads re-fetch from the renamed file rather than serving stale entries.
-    db.pager.reset_main_pages();
-
     let new_commit_id = state.latest_commit_id + 1;
+    if db.vfs.rename(scratch, &db.main_db_path).await.is_err() {
+        // After closing the old handle, a backend may have performed an
+        // ambiguous replacement even when it reports an error. Reopen is the
+        // only safe way to establish the durable image.
+        let commit = crate::CommitId(new_commit_id);
+        db.poison(commit);
+        return Err(crate::errors::PagedbError::durably_committed_but_unpublished(commit));
+    }
+
+    // Rename is the durable-image boundary. Advance internal state and drop
+    // cache pages immediately, but do not publish until the parent directory
+    // sync and nonce-anchor commit both succeed.
     state.root_page_id = pending.new_root;
     state.catalog_root_page_id = pending.new_cat_root;
+    state.catalog_root_txn_id = new_commit_id;
     state.next_page_id = pending.new_next;
     state.active_slot = ActiveSlot::A;
     state.seq += 1;
     state.latest_commit_id = new_commit_id;
     state.commit_history_root_page_id = 0;
     state.commit_history_root_version = 0;
     state.free_list_root_page_id = 0;
-    db.latest_commit
-        .store(new_commit_id, std::sync::atomic::Ordering::SeqCst);
+    db.pager.reset_main_pages();
+
+    if db.vfs.sync_dir(db.main_db_parent_dir()).await.is_err() {
+        let commit = crate::CommitId(new_commit_id);
+        db.poison(commit);
+        return Err(crate::errors::PagedbError::durably_committed_but_unpublished(commit));
+    }
+    let _ = db
+        .finish_durable_commit_visible(
+            visibility,
+            state,
+            crate::CommitId(new_commit_id),
+            pending.counter_anchor,
+            &[],
+        )
+        .await?;
 
     let pages_reclaimed = pending.old_next.saturating_sub(pending.new_next);
     Ok(RepackOutcome {

src/errors.rs

@@ -50,6 +50,12 @@ pub enum PagedbError {
     #[error("identity forked; apply_incremental refused")]
     IdentityForked,
 
+    #[error("incremental snapshot is incompatible: {field}")]
+    SnapshotIncompatible { field: &'static str },
+
+    #[error("commit {commit:?} is durable but unpublished; reopen required")]
+    DurablyCommittedButUnpublished { commit: CommitId },
+
     #[error("commit {commit:?} gone; oldest_available={oldest_available:?}")]
     CommitGone {
         commit: CommitId,
@@ -202,6 +208,20 @@ impl PagedbError {
         Self::Corruption(detail)
     }
 
+    /// Canonical constructor for an incremental snapshot that cannot be
+    /// applied to this handle's current identity or reader-visible state.
+    #[must_use]
+    pub const fn snapshot_incompatible(field: &'static str) -> Self {
+        Self::SnapshotIncompatible { field }
+    }
+
+    /// Canonical constructor for a handle whose newest durable commit could
+    /// not be reconciled into its reader-visible state.
+    #[must_use]
+    pub const fn durably_committed_but_unpublished(commit: CommitId) -> Self {
+        Self::DurablyCommittedButUnpublished { commit }
+    }
+
     /// Canonical constructor for arithmetic-overflow errors.
     #[must_use]
     pub const fn arithmetic_overflow(operation: &'static str) -> Self {

src/pager/core.rs

@@ -432,8 +432,8 @@ impl<V: Vfs> Pager<V> {
         self.inner.cache_for_key(key).lock().clear_file(key);
         self.journal_nonces.lock().remove(&journal_id);
         let path = format!("applyjournal/{}", crate::hex::to_hex_lower(&journal_id));
-        self.vfs.remove(&path).await.ok();
-        Ok(())
+        self.vfs.remove(&path).await?;
+        self.vfs.sync_dir("applyjournal").await
     }
 
     /// Drop an apply-journal sidecar's cached pages without removing the file,

src/recovery/deep_walk.rs

@@ -144,6 +144,7 @@ impl DeepWalkReport {
 /// page including free, spill, and unreferenced pages.
 #[allow(clippy::too_many_lines)]
 pub async fn run_deep_walk<V: Vfs + Clone>(db: &Db<V>) -> Result<DeepWalkReport> {
+    db.ensure_usable()?;
     let mut report = DeepWalkReport::default();
 
     let (next_page_id, catalog_root, catalog_next, free_list_root) = {