fix(compaction): atomic scratch-file repack; honest single-shot compact_step

Author: farhan-syah

src/catalog/codec.rs

@@ -42,46 +42,13 @@ pub enum CatalogRowKind {
     /// catalog, reader pins are maintained in-memory only and the writer process
     /// must be trusted to honor the catalog pins.
     ReaderPin = 0x06,
-    /// Incremental compaction watermark. Singleton row; key is `[0x07]`.
-    /// Value: `target_root[8] || frontier_page_id[8] || started_at_commit_id[8] ||
-    ///          total_pages_estimate[8]` = 32 bytes.
-    ///
-    /// Present only while an incremental compaction is in progress. Cleared
-    /// atomically with the final compaction commit. On `Db::open`, a present
-    /// row means a prior compaction was interrupted; the embedder must call
-    /// `compact_step` again to resume — `Db::open` does NOT auto-resume.
+    /// 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
+    /// dropped during compaction.
     CompactionState = 0x07,
 }
 
-/// Incremental compaction watermark persisted in the catalog.
-///
-/// Present while a `compact_step` session is in progress. The embedder must
-/// call `compact_step` again to resume after a crash; `Db::open` does not
-/// auto-resume.
-///
-/// Encoding: `target_root[8] || frontier_page_id[8] || started_at_commit_id[8] ||
-///             total_pages_estimate[8] || frontier_key_len[4] || frontier_key[frontier_key_len]`
-/// (minimum 32 bytes; variable total).
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct CompactionStateRow {
-    /// Root page id of the main B+ tree at the time compaction started.
-    /// Used as the source snapshot to read from during each step.
-    pub target_root: u64,
-    /// The `next_page_id` of the partially-built compacted tree after the last
-    /// committed batch. Used to estimate progress and as the initial `next_page_id`
-    /// when no free-list pages are available.
-    pub frontier_page_id: u64,
-    /// The `commit_id` at which the compaction session began.
-    pub started_at_commit_id: u64,
-    /// Estimated total pages in the source tree (for progress reporting).
-    pub total_pages_estimate: u64,
-    /// The key (exclusive lower bound) from which the next step should begin
-    /// reading pairs. Empty means start from the beginning.
-    pub frontier_key: Vec<u8>,
-}
-
-pub const COMPACTION_STATE_FIXED_LEN: usize = 36; // 4 × u64 + 1 × u32 length prefix
-
 /// Rekey watermark persisted in the catalog during an online rekey operation.
 /// A present row means a rekey is in flight or was interrupted by a crash.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -246,61 +213,6 @@ impl Catalog {
         [CatalogRowKind::CompactionState as u8]
     }
 
-    /// Compaction-state row key: `[0x07]` (singleton).
-    #[must_use]
-    pub fn compaction_state_key() -> [u8; 1] {
-        [CatalogRowKind::CompactionState as u8]
-    }
-
-    /// Encode a `CompactionStateRow`. Returns a variable-length `Vec<u8>`.
-    pub fn encode_compaction_state(r: &CompactionStateRow) -> Result<Vec<u8>> {
-        let key_len = r.frontier_key.len();
-        let klen32 = u32::try_from(key_len).map_err(|_| PagedbError::ManifestTooLarge)?;
-        let mut o = Vec::with_capacity(COMPACTION_STATE_FIXED_LEN + key_len);
-        o.extend_from_slice(&r.target_root.to_le_bytes());
-        o.extend_from_slice(&r.frontier_page_id.to_le_bytes());
-        o.extend_from_slice(&r.started_at_commit_id.to_le_bytes());
-        o.extend_from_slice(&r.total_pages_estimate.to_le_bytes());
-        o.extend_from_slice(&klen32.to_le_bytes());
-        o.extend_from_slice(&r.frontier_key);
-        Ok(o)
-    }
-
-    /// Decode a `CompactionStateRow` from bytes.
-    pub fn decode_compaction_state(bytes: &[u8]) -> Result<CompactionStateRow> {
-        if bytes.len() < COMPACTION_STATE_FIXED_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)
-        };
-        let target_root = read_u64(0);
-        let frontier_page_id = read_u64(8);
-        let started_at_commit_id = read_u64(16);
-        let total_pages_estimate = read_u64(24);
-        let mut klen_buf = [0u8; 4];
-        klen_buf.copy_from_slice(&bytes[32..36]);
-        let key_len = u32::from_le_bytes(klen_buf) as usize;
-        if bytes.len() < COMPACTION_STATE_FIXED_LEN + key_len {
-            return Err(PagedbError::corruption(
-                crate::errors::CorruptionDetail::HeaderUnverifiable,
-            ));
-        }
-        let frontier_key =
-            bytes[COMPACTION_STATE_FIXED_LEN..COMPACTION_STATE_FIXED_LEN + key_len].to_vec();
-        Ok(CompactionStateRow {
-            target_root,
-            frontier_page_id,
-            started_at_commit_id,
-            total_pages_estimate,
-            frontier_key,
-        })
-    }
-
     /// Encode a `ReaderPinValue` as 41 bytes.
     #[must_use]
     pub fn encode_reader_pin(v: &ReaderPinValue) -> [u8; READER_PIN_VALUE_LEN] {

src/catalog/mod.rs

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

src/compaction/full.rs

@@ -1,28 +1,20 @@
 //! Full one-shot compaction (entry point: [`compact_now`]):
 //!
-//! 1. Collects all live data from main and catalog trees into memory.
-//! 2. Drains eligible deferred-free pages (now eligible since no pinned reader
-//!    can observe them).
-//! 3. Writes fresh compacted trees starting at page 4, producing a dense layout.
-//! 4. Commits a new header with the updated roots and reduced `next_page_id`.
-//! 5. Truncates main.db if no reader pins the old high-water-mark range.
-//! 6. Repacks segment files whose garbage ratio exceeds 5%.
+//! 1. Refuses while any reader pins the page range (relocation/truncation is
+//!    unsafe under a pinned reader).
+//! 2. Crash-atomically repacks the main + catalog trees into a dense low-address
+//!    layout and truncates the reclaimed tail (see [`super::repack`]).
+//! 3. Repacks segment files whose garbage ratio exceeds 5%.
 
-use crate::btree::BTree;
+use crate::Result;
 use crate::errors::PagedbError;
-use crate::pager::header::commit_header;
-use crate::pager::structural_header::MainDbHeaderFields;
 use crate::segment::reader::SegmentReader;
 use crate::segment::types::SegmentPageKind;
 use crate::segment::writer::SegmentWriter;
 use crate::txn::db::Db;
 use crate::vfs::{Vfs, VfsFile};
-use crate::{CommitId, Result};
 
-use super::helpers::{
-    collect_all_pairs, collect_catalog_split, find_segment_name_inner, list_all_segments_inner,
-    page_size_log2, replace_segment_compact,
-};
+use super::helpers::{find_segment_name_inner, list_all_segments_inner, replace_segment_compact};
 use super::types::CompactStats;
 
 /// Full online compaction. See module-level docs for the staged flow.
@@ -31,7 +23,7 @@ use super::types::CompactStats;
 /// `EnteredSpan` guard: an entered span guard is `!Send` and would be held
 /// across the many `.await` points below, making the returned future `!Send`
 /// and thus uncallable from `Send` async contexts (e.g. the nodedb-lite
-/// `#[async_trait]` StorageEngine impl, which requires `Send` futures).
+/// `#[async_trait]` `StorageEngine` impl, which requires `Send` futures).
 pub async fn compact_now<V: Vfs + Clone>(db: &Db<V>) -> Result<CompactStats> {
     use tracing::Instrument;
     compact_now_inner(db)
@@ -55,8 +47,6 @@ async fn compact_now_inner<V: Vfs + Clone>(db: &Db<V>) -> Result<CompactStats> {
     // commit can't recycle a page the repack now uses for live data.
     db.free_page_cache.lock().clear();
 
-    let old_next_page_id = state.next_page_id;
-
     // ── 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
@@ -81,124 +71,19 @@ async fn compact_now_inner<V: Vfs + Clone>(db: &Db<V>) -> Result<CompactStats> {
         return Ok(result);
     }
 
-    // ── 2. Full repack (no readers pinned) ───────────────────────────────────
-
-    // Collect all live data in memory BEFORE any writes.
-    let main_pairs = if state.root_page_id != 0 {
-        let old_tree = BTree::open(
-            db.pager.clone(),
-            db.realm_id,
-            state.root_page_id,
-            old_next_page_id,
-            db.page_size,
-        );
-        collect_all_pairs(&old_tree).await?
-    } else {
-        Vec::new()
-    };
-
-    // Collect catalog rows (housekeeping free-list rows dropped).
-    let cat_rows = collect_catalog_split(&db.pager, db.realm_id, &state).await?;
-
-    // ── 3. Write fresh compacted trees starting at page 4 ────────────────────
-    // Pages 0–3 are reserved (header slots A/B + two spares); never allocated.
-    let mut new_main = BTree::open(
-        db.pager.clone(),
-        db.realm_id,
-        0,
-        4, // first data page (pages 0-3 are reserved header slots)
-        db.page_size,
-    );
-    new_main.bulk_load(main_pairs).await?;
-    new_main.flush().await?;
-    let new_root = new_main.root_page_id();
-    let after_main = new_main.next_page_id();
-
-    let mut new_cat = BTree::open(db.pager.clone(), db.realm_id, 0, after_main, db.page_size);
-    new_cat.bulk_load(cat_rows).await?;
-    new_cat.flush().await?;
-    let new_cat_root = new_cat.root_page_id();
-    let new_next = new_cat.next_page_id();
-
-    // Pages reclaimed = reduction in next_page_id (the dense layout is contiguous,
-    // and the durable free-list is reset to empty below).
-    result.main_db_pages_reclaimed = old_next_page_id.saturating_sub(new_next);
-
-    // ── 4. Commit new header ─────────────────────────────────────────────────
-    let new_commit_id = state.latest_commit_id + 1;
-    let new_seq = state.seq + 1;
-    let counter_anchor = db.pager.pending_anchor();
-
-    let mut catalog_root_bytes = [0u8; 16];
-    catalog_root_bytes[..8].copy_from_slice(&new_cat_root.to_le_bytes());
-    catalog_root_bytes[8..].copy_from_slice(&new_commit_id.to_le_bytes());
-
-    let fields = MainDbHeaderFields {
-        format_version: 1,
-        cipher_id: db.cipher_id.as_byte(),
-        page_size_log2: page_size_log2(db.page_size)?,
-        flags: 0,
-        file_id: db.file_id,
-        kek_salt: db.kek_salt,
-        mk_epoch: db.mk_epoch.load(std::sync::atomic::Ordering::SeqCst),
-        seq: new_seq,
-        active_root_page_id: new_root,
-        active_root_txn_id: new_commit_id,
-        counter_anchor,
-        commit_id: CommitId(new_commit_id),
-        free_list_root: [0u8; 16],
-        catalog_root: catalog_root_bytes,
-        apply_journal_root_page_id: 0,
-        apply_journal_root_version: 0,
-        commit_history_root_page_id: 0,
-        commit_history_root_version: 0,
-        restore_mode: 0,
-        next_page_id: new_next,
-        commit_retain_policy_tag: 0,
-        commit_retain_policy_value: 0,
-    };
-
-    let hk_clone = { db.hk.read().clone() };
-    let new_slot = commit_header(
-        &*db.vfs,
-        &db.main_db_path,
-        &hk_clone,
-        &fields,
-        state.active_slot,
-        db.page_size,
-    )
-    .await?;
-    db.pager.commit_anchor(counter_anchor)?;
-
-    state.root_page_id = new_root;
-    state.catalog_root_page_id = new_cat_root;
-    state.next_page_id = new_next;
-    state.active_slot = new_slot;
-    state.seq = new_seq;
-    state.latest_commit_id = new_commit_id;
-    state.commit_history_root_page_id = 0;
-    state.commit_history_root_version = 0;
-    // The dense repack relocates/truncates every page, so the old free-list is
-    // gone; the new layout starts with an empty free-list.
-    state.free_list_root_page_id = 0;
-    db.latest_commit
-        .store(new_commit_id, std::sync::atomic::Ordering::SeqCst);
-
-    // ── 5. Truncate if no readers pin the old high-water range ───────────────
-    // (No readers are pinned at this point — checked above — so truncation is safe.)
-    if new_next < old_next_page_id {
-        let new_size = new_next.saturating_mul(db.page_size as u64);
-        let old_size = old_next_page_id.saturating_mul(db.page_size as u64);
-        let mut f = db
-            .vfs
-            .open(&db.main_db_path, crate::vfs::types::OpenMode::ReadWrite)
-            .await?;
-        f.set_len(new_size).await?;
-        f.sync().await?;
-        result.bytes_truncated = old_size.saturating_sub(new_size);
+    // ── 2. Crash-atomic dense repack of the main + catalog trees ─────────────
+    // An empty free-list means every page below the high-water mark is live —
+    // the store is already dense, so there is nothing to reclaim and we skip the
+    // repack entirely (no wasted rewrite). Otherwise repack via a scratch file +
+    // 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?;
+        result.main_db_pages_reclaimed = repack.pages_reclaimed;
+        result.bytes_truncated = repack.bytes_truncated;
     }
 
-    // ── 6. Repack segments ────────────────────────────────────────────────────
+    // ── 3. Repack segments ────────────────────────────────────────────────────
     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);

src/compaction/helpers.rs

@@ -4,7 +4,7 @@
 use std::sync::Arc;
 
 use crate::btree::BTree;
-use crate::catalog::codec::{Catalog, CatalogRowKind, CompactionStateRow, SegmentMeta};
+use crate::catalog::codec::{Catalog, CatalogRowKind, SegmentMeta};
 use crate::errors::PagedbError;
 use crate::pager::header::commit_header;
 use crate::pager::structural_header::MainDbHeaderFields;
@@ -195,72 +195,6 @@ pub(super) async fn replace_segment_compact<V: Vfs + Clone>(
     Ok(())
 }
 
-/// Return the "next key strictly greater than `key`" for range scanning.
-/// Appends a 0x00 byte, which gives the lexicographically smallest key
-/// greater than `key`. If `key` is empty, returns empty (scan from start).
-pub(super) fn next_key_after(key: &[u8]) -> Vec<u8> {
-    if key.is_empty() {
-        Vec::new()
-    } else {
-        let mut v = key.to_vec();
-        v.push(0x00);
-        v
-    }
-}
-
-/// Collect at most `limit` key-value pairs from `[start..end)` in `tree`.
-pub(super) async fn collect_range_limited<V: Vfs + Clone>(
-    tree: &BTree<V>,
-    start: &[u8],
-    end: &[u8],
-    limit: u64,
-) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
-    let mut pairs = tree.collect_range(start, end).await?;
-    #[allow(clippy::cast_possible_truncation)]
-    if pairs.len() as u64 > limit {
-        pairs.truncate(limit as usize);
-    }
-    Ok(pairs)
-}
-
-/// Load the compaction watermark from the catalog, if present.
-pub(super) async fn load_compaction_state<V: Vfs + Clone>(
-    pager: &Arc<crate::pager::Pager<V>>,
-    realm_id: crate::RealmId,
-    state: &WriterState,
-) -> Result<Option<CompactionStateRow>> {
-    if state.catalog_root_page_id == 0 {
-        return Ok(None);
-    }
-    let tree = BTree::open(
-        pager.clone(),
-        realm_id,
-        state.catalog_root_page_id,
-        state.next_page_id,
-        pager.page_size(),
-    );
-    let key = Catalog::compaction_state_key();
-    match tree.get(&key).await? {
-        Some(bytes) => {
-            let cs = Catalog::decode_compaction_state(&bytes)?;
-            Ok(Some(cs))
-        }
-        None => Ok(None),
-    }
-}
-
-/// Load just the `frontier_page_id` from the watermark (for progress reporting).
-pub(super) async fn load_frontier_page_id<V: Vfs + Clone>(
-    pager: &Arc<crate::pager::Pager<V>>,
-    realm_id: crate::RealmId,
-    state: &WriterState,
-) -> Option<u64> {
-    match load_compaction_state(pager, realm_id, state).await {
-        Ok(Some(cs)) => Some(cs.frontier_page_id),
-        _ => None,
-    }
-}
-
 /// Build [`MainDbHeaderFields`] for a compaction commit.
 ///
 /// Compaction relocates/truncates pages, so it discards the commit-history

src/compaction/mod.rs

@@ -9,6 +9,7 @@
 
 mod full;
 mod helpers;
+mod repack;
 mod step;
 mod types;