From 2b7f4fd6fdadecb6463a27e509a7e6c175e631c1 Mon Sep 17 00:00:00 2001 From: Andrew Briscoe Date: Sun, 26 Jul 2026 18:33:38 -0600 Subject: [PATCH 1/3] fix(txn): preserve cache and transaction lifecycle state --- src/pager/cache.rs | 71 +++++++++++++++++++++++++++++++++++++++-- src/pager/core.rs | 9 ++++++ src/txn/db/catalog.rs | 4 +-- src/txn/db/misc.rs | 5 +-- src/txn/write/txn.rs | 11 +++++++ tests/commit_history.rs | 34 +++++++++++++++++++- tests/spill_basic.rs | 32 +++++++++++++++++++ tests/stats_basic.rs | 32 +++++++++++++++++++ 8 files changed, 190 insertions(+), 8 deletions(-) diff --git a/src/pager/cache.rs b/src/pager/cache.rs index d2a6830..d030f2d 100644 --- a/src/pager/cache.rs +++ b/src/pager/cache.rs @@ -153,9 +153,10 @@ impl PageCache { } fn evict_one(&mut self) -> Option<(FileKey, u64)> { - // Walk the hand at most `capacity * 2` steps to bound worst case - // (every entry either visited or unevictable triggers a wrap). - let max_steps = self.capacity.saturating_mul(2).max(1); + // Two passes are sufficient to clear visited bits and then evict. + // Use the live list length rather than the configured capacity because + // pinned or dirty pages can temporarily grow the cache past capacity. + let max_steps = self.map.len().saturating_mul(2).max(1); let mut cur = self.hand.or(self.tail); for _ in 0..max_steps { let Some(idx) = cur else { @@ -293,6 +294,33 @@ impl PageCache { self.dirty.remove(&key); } } + + /// Drop every unpinned clean entry for `file`, leaving dirty entries + /// available for a later flush and pinned entries available to readers. + pub fn clear_clean_file(&mut self, file: FileKey) { + let keys: Vec<(FileKey, u64)> = self + .map + .keys() + .filter(|(f, _)| *f == file) + .copied() + .collect(); + for key in keys { + if self.pins.get(&key).copied().unwrap_or(0) > 0 || self.dirty.contains(&key) { + continue; + } + if let Some(idx) = self.map.remove(&key) { + let (prev, next) = { + let node = self.slab[idx].as_ref().expect("indexed node alive"); + (node.prev, node.next) + }; + if self.hand == Some(idx) { + self.hand = next; + } + self.unlink_node(idx, prev, next); + self.free_node(idx); + } + } + } } #[cfg(test)] @@ -368,6 +396,20 @@ mod tests { assert!(c.get((FileKey::Main, 1)).is_some()); } + #[test] + fn clear_clean_file_preserves_dirty_entries() { + let mut c = PageCache::with_capacity(4); + c.insert((FileKey::Main, 1), page(1)); + c.insert((FileKey::Main, 2), page(2)); + c.mark_dirty((FileKey::Main, 2)); + + c.clear_clean_file(FileKey::Main); + + assert!(c.get((FileKey::Main, 1)).is_none()); + assert!(c.get((FileKey::Main, 2)).is_some()); + assert!(c.is_dirty((FileKey::Main, 2))); + } + #[test] fn dirty_iter_is_sorted_ascending() { let mut c = PageCache::with_capacity(16); @@ -396,4 +438,27 @@ mod tests { c.insert((FileKey::Main, 3), page(3)); assert_eq!(c.slab.len(), 2, "slab reuses freed slot"); } + + #[test] + fn overgrown_cache_still_finds_clean_victim() { + let mut c = PageCache::with_capacity(2); + + for id in 1u8..=4 { + let key = (FileKey::Main, u64::from(id)); + c.insert(key, page(id)); + c.pin(key); + } + c.insert((FileKey::Main, 5), page(5)); + assert_eq!(c.len(), 5, "pinned pages may force temporary growth"); + + let evicted = c.insert((FileKey::Main, 6), page(6)); + assert_eq!( + evicted, + Some((FileKey::Main, 5)), + "eviction must scan the full over-capacity list for a clean victim" + ); + assert_eq!(c.len(), 5); + assert!(c.get((FileKey::Main, 5)).is_none()); + assert!(c.get((FileKey::Main, 6)).is_some()); + } } diff --git a/src/pager/core.rs b/src/pager/core.rs index 4719200..b97a42c 100644 --- a/src/pager/core.rs +++ b/src/pager/core.rs @@ -870,6 +870,15 @@ impl Pager { } } + /// Evict unpinned clean main.db pages so later reads fetch and authenticate + /// durable bytes without discarding in-flight writes. + pub fn evict_clean_main_pages(&self, _realm_id: crate::RealmId) { + self.inner + .buffer_pool + .lock() + .clear_clean_file(FileKey::Main); + } + fn write_page( &self, file: FileKey, diff --git a/src/txn/db/catalog.rs b/src/txn/db/catalog.rs index 030d5f9..f22184a 100644 --- a/src/txn/db/catalog.rs +++ b/src/txn/db/catalog.rs @@ -151,8 +151,8 @@ impl Db { counter_anchor, commit_id: state.latest_commit_id, catalog_root: catalog_root_bytes, - commit_history_root_page_id: 0, - commit_history_root_version: 0, + commit_history_root_page_id: state.commit_history_root_page_id, + commit_history_root_version: state.commit_history_root_version, free_list_root_page_id: state.free_list_root_page_id, next_page_id: new_next, })?; diff --git a/src/txn/db/misc.rs b/src/txn/db/misc.rs index 2e6776f..ee55715 100644 --- a/src/txn/db/misc.rs +++ b/src/txn/db/misc.rs @@ -40,7 +40,8 @@ impl Db { .applies_incremental_snapshots() } - /// Drop every cached page of `realm` so the next read goes to storage. + /// Drop unpinned clean cached pages of `realm` so the next read goes to + /// storage, leaving dirty and pinned entries untouched. /// /// Compiled only for the crate's own tests: nothing an embedder does is /// supposed to depend on whether a page is warm, and publishing the lever @@ -48,7 +49,7 @@ impl Db { /// outside the crate reopens the handle instead. #[cfg(test)] pub(crate) fn evict_main_pages(&self, realm: crate::RealmId) { - self.pager.discard_dirty_main(realm); + self.pager.evict_clean_main_pages(realm); } /// Current size of `main.db` in bytes, read from the file rather than from diff --git a/src/txn/write/txn.rs b/src/txn/write/txn.rs index a4bc94d..00809d4 100644 --- a/src/txn/write/txn.rs +++ b/src/txn/write/txn.rs @@ -183,6 +183,14 @@ impl<'db, V: Vfs + Clone> WriteTxn<'db, V> { // Assign a txn_seq starting from 1: fetch_add returns the old value (0 // for the first call), so we add 1 to produce 1-based ids. let txn_seq = db.txn_seq.fetch_add(1, Ordering::Relaxed) + 1; + + // Drop cannot await VFS removal. Sweep the only scratch file a dropped + // predecessor could have left before this transaction can allocate its + // own path. Abort and commit already remove their files eagerly. + if txn_seq > 1 { + let _ = db.vfs.remove(&format!("tmp/scratch-{}", txn_seq - 1)).await; + } + Ok(Self { db, guard, @@ -407,6 +415,9 @@ impl Drop for WriteTxn<'_, V> { fn drop(&mut self) { if !self.committed_or_aborted { self.db.pager.discard_dirty_main(self.db.realm_id); + self.db + .spill_bytes_in_use + .store(0, std::sync::atomic::Ordering::Relaxed); } } } diff --git a/tests/commit_history.rs b/tests/commit_history.rs index 011c2aa..1748a21 100644 --- a/tests/commit_history.rs +++ b/tests/commit_history.rs @@ -1,7 +1,7 @@ use std::time::Duration; use pagedb::vfs::memory::MemVfs; -use pagedb::{CommitId, Db, OpenOptions, PagedbError, RealmId, RetainPolicy}; +use pagedb::{CommitId, Db, OpenOptions, PagedbError, RealmId, RealmQuotas, RetainPolicy}; const PAGE: usize = 4096; const KEK: [u8; 32] = [7u8; 32]; @@ -140,6 +140,38 @@ async fn history_persists_across_reopen() { .unwrap_or_else(|e| panic!("expected commit {cid3:?} to survive reopen, got {e:?}")); } +#[tokio::test(flavor = "current_thread")] +async fn quota_update_preserves_history_across_reopen() { + let vfs = MemVfs::new(); + let opts = OpenOptions::default().with_commit_history_retain(RetainPolicy::Unbounded); + + let first_commit = { + let db = Db::open_internal_with_options(vfs.clone(), KEK, PAGE, REALM, opts.clone()) + .await + .unwrap(); + let ids = write_n(&db, 2).await; + db.set_realm_quotas(REALM, RealmQuotas::default()) + .await + .unwrap(); + db.begin_read_at(ids[0]) + .await + .expect("quota publication must preserve live retained history"); + ids[0] + }; + + let reopened = Db::open_existing_with_options(vfs, KEK, PAGE, REALM, opts) + .await + .unwrap(); + let historical = reopened + .begin_read_at(first_commit) + .await + .expect("quota publication must preserve retained history after reopen"); + assert_eq!( + historical.get(b"k").await.unwrap().as_deref(), + Some(0_u64.to_le_bytes().as_slice()) + ); +} + #[tokio::test(flavor = "current_thread")] async fn retain_policy_age_prunes_old() { // Age(0) means threshold = now - 0 = now, so every entry with diff --git a/tests/spill_basic.rs b/tests/spill_basic.rs index 231a5b7..db92ab1 100644 --- a/tests/spill_basic.rs +++ b/tests/spill_basic.rs @@ -105,6 +105,38 @@ async fn spill_cleanup_on_abort() { assert!(res.is_err(), "tmp file should be cleaned up after abort"); } +#[tokio::test(flavor = "current_thread")] +async fn spill_drop_resets_stats_and_next_write_sweeps_stale_tmp() { + let vfs = MemVfs::new(); + let opts = OpenOptions::default().with_scratch_bytes(1024 * 1024); + let db = Db::open_internal_with_options(vfs.clone(), [9u8; 32], PAGE, REALM, opts) + .await + .unwrap(); + { + let mut w = db.begin_write().await.unwrap(); + let mut s = w.spill_scope(); + s.append(b"drop-cleanup").await.unwrap(); + drop(s); + drop(w); + } + + assert_eq!( + db.stats().await.unwrap().spill_bytes_in_use, + 0, + "dropping an uncommitted transaction must clear observable spill accounting" + ); + + { + let w = db.begin_write().await.unwrap(); + let res = vfs.open("tmp/scratch-1", OpenMode::Read).await; + assert!( + res.is_err(), + "the next write transaction must sweep stale spill tmp files from dropped transactions" + ); + w.abort().await; + } +} + #[tokio::test(flavor = "current_thread")] async fn spill_aead_protects_payload() { // Confirm bytes on disk are NOT plaintext. diff --git a/tests/stats_basic.rs b/tests/stats_basic.rs index b68e3d9..aa1bcd9 100644 --- a/tests/stats_basic.rs +++ b/tests/stats_basic.rs @@ -117,6 +117,38 @@ async fn stats_reports_buffer_pool() { ); } +#[tokio::test] +async fn evict_main_pages_forces_next_read_to_miss_cache() { + let db = fresh_db().await; + { + let mut txn = db.begin_write().await.unwrap(); + txn.put(b"hello", b"world").await.unwrap(); + txn.commit().await.unwrap(); + } + + { + let reader = db.begin_read().await.unwrap(); + assert_eq!( + reader.get(b"hello").await.unwrap().as_deref(), + Some(b"world".as_slice()) + ); + } + let before = db.stats().await.unwrap(); + + db.evict_main_pages(realm()); + + let reader = db.begin_read().await.unwrap(); + assert_eq!( + reader.get(b"hello").await.unwrap().as_deref(), + Some(b"world".as_slice()) + ); + let after = db.stats().await.unwrap(); + assert!( + after.buffer_pool_misses > before.buffer_pool_misses, + "eviction must force a VFS-backed cache miss: before={before:?}, after={after:?}" + ); +} + #[tokio::test] async fn stats_reports_mode() { let vfs = MemVfs::new(); From 1b98b9c05d7075a8565c0215a89ab58ca48c9d47 Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Mon, 27 Jul 2026 20:38:38 +0800 Subject: [PATCH 2/3] fix(txn): scope spill keys to the open handle and reclaim orphaned scratch The transaction sequence restarts at 1 on every open and doubles as the spill nonce, so a spill key derived only from the durable file id and that sequence would repeat across two opens of one store at the same sequence, and the nonce built from it would repeat with it. Mix a random per-open epoch into the derivation so the key is distinct per open, leaving the sequence free to serve as the nonce. Pair this with reclaiming the scratch such keys ever protected: a transaction dropped without commit or abort cannot await the removal, so it now hands its exact path to the next begin_write, and whatever still survives at the next open is swept outright, since a spill file is never read back across handles. --- src/crypto/kdf.rs | 65 ++++++++++++++++++++++++++---- src/crypto/random/identity.rs | 18 +++++++++ src/crypto/random/mod.rs | 2 +- src/recovery/mod.rs | 3 +- src/recovery/scratch.rs | 74 +++++++++++++++++++++++++++++++++++ src/txn/db/core.rs | 11 ++++++ src/txn/db/open/create.rs | 12 +++--- src/txn/db/open/existing.rs | 2 + src/txn/db/open/recovery.rs | 8 ++++ src/txn/write/spill.rs | 30 +++++++++----- src/txn/write/txn.rs | 27 ++++++++++--- tests/commit_history.rs | 6 +-- tests/spill_basic.rs | 33 +++++++++++++++- 13 files changed, 257 insertions(+), 34 deletions(-) create mode 100644 src/recovery/scratch.rs diff --git a/src/crypto/kdf.rs b/src/crypto/kdf.rs index cb99e8e..2edd8df 100644 --- a/src/crypto/kdf.rs +++ b/src/crypto/kdf.rs @@ -57,14 +57,35 @@ pub fn derive_hk(mk: &MasterKey) -> Result { /// Derive a transient per-WriteTxn spill key from the master key. /// -/// `info` = `"pagedb/spill/" ‖ file_id[16] ‖ txn_seq.to_le_bytes()[8]` (13 + 16 + 8 = 37 bytes). -/// The key is transient: the tmp file it protects is discarded at commit/abort. -pub fn derive_spill_key(mk: &MasterKey, file_id: &[u8; 16], txn_seq: u64) -> Result { +/// `info` = `"pagedb/spill/" ‖ file_id[16] ‖ spill_epoch[16] ‖ txn_seq.to_le_bytes()[8]` +/// (13 + 16 + 16 + 8 = 53 bytes). +/// +/// `spill_epoch` is fresh per open handle. `file_id` and the master key are +/// durable and `txn_seq` restarts at 1 on every open, so those three alone +/// would repeat the same key across two opens of one store — and the spill +/// nonce, built from that same restarting sequence, would repeat with it. The +/// epoch is what keeps each open's spill key distinct, and therefore what makes +/// the sequence a sound nonce source without a durable anchor. +/// +/// Nothing reads a spill file written by an earlier open, so there is no +/// durability cost to a per-open key: the scratch a crash leaves behind is +/// swept at the next open, never decrypted. +pub fn derive_spill_key( + mk: &MasterKey, + file_id: &[u8; 16], + spill_epoch: &[u8; 16], + txn_seq: u64, +) -> Result { const PREFIX: &[u8] = b"pagedb/spill/"; - let mut info = [0u8; PREFIX.len() + 16 + 8]; - info[..PREFIX.len()].copy_from_slice(PREFIX); - info[PREFIX.len()..PREFIX.len() + 16].copy_from_slice(file_id); - info[PREFIX.len() + 16..].copy_from_slice(&txn_seq.to_le_bytes()); + let mut info = [0u8; PREFIX.len() + 16 + 16 + 8]; + let mut off = 0; + info[off..off + PREFIX.len()].copy_from_slice(PREFIX); + off += PREFIX.len(); + info[off..off + 16].copy_from_slice(file_id); + off += 16; + info[off..off + 16].copy_from_slice(spill_epoch); + off += 16; + info[off..off + 8].copy_from_slice(&txn_seq.to_le_bytes()); expand(mk.as_bytes(), &info) } @@ -114,6 +135,36 @@ mod tests { assert_ne!(a.as_bytes(), b.as_bytes()); } + #[test] + fn spill_key_isolates_per_open_at_the_same_sequence() { + // The transaction sequence restarts at 1 on every open and also builds + // the spill nonce, so two opens of one store must not share a key at + // one sequence. + let mk = derive_mk(&fixed_kek(), &[0; 16], 0).unwrap(); + let file_id = [0x5A; 16]; + let a = derive_spill_key(&mk, &file_id, &[0x01; 16], 1).unwrap(); + let b = derive_spill_key(&mk, &file_id, &[0x02; 16], 1).unwrap(); + assert_ne!(a.as_bytes(), b.as_bytes()); + } + + #[test] + fn spill_key_isolates_per_sequence_within_one_open() { + let mk = derive_mk(&fixed_kek(), &[0; 16], 0).unwrap(); + let epoch = [0x01; 16]; + let a = derive_spill_key(&mk, &[0x5A; 16], &epoch, 1).unwrap(); + let b = derive_spill_key(&mk, &[0x5A; 16], &epoch, 2).unwrap(); + assert_ne!(a.as_bytes(), b.as_bytes()); + } + + #[test] + fn spill_key_isolates_per_store() { + let mk = derive_mk(&fixed_kek(), &[0; 16], 0).unwrap(); + let epoch = [0x01; 16]; + let a = derive_spill_key(&mk, &[0x5A; 16], &epoch, 1).unwrap(); + let b = derive_spill_key(&mk, &[0xA5; 16], &epoch, 1).unwrap(); + assert_ne!(a.as_bytes(), b.as_bytes()); + } + #[test] fn dek_isolates_per_realm() { let mk = derive_mk(&fixed_kek(), &[0; 16], 0).unwrap(); diff --git a/src/crypto/random/identity.rs b/src/crypto/random/identity.rs index ace9e74..a76844a 100644 --- a/src/crypto/random/identity.rs +++ b/src/crypto/random/identity.rs @@ -25,6 +25,18 @@ pub(crate) fn segment_id() -> Result<[u8; 16]> { random_nonzero_identity() } +/// Identity partitioning the spill nonce space of one open handle. +/// +/// The per-transaction spill key is derived from the durable `file_id` and the +/// transaction sequence, and the spill nonce is built from that same sequence. +/// The sequence restarts at 1 on every open, so without a per-open component +/// two handles over the same store would encrypt different scratch payloads +/// under one key at one nonce. Mixing this in makes the key distinct per open, +/// which is what leaves the sequence free to serve as the nonce. +pub(crate) fn spill_epoch() -> Result<[u8; 16]> { + random_nonzero_identity() +} + // Sole caller is `txn::db::snapshot::apply_incremental`, which is native-only // (`#![cfg(not(target_arch = "wasm32"))]`); gate the function the same way. #[cfg(not(target_arch = "wasm32"))] @@ -41,5 +53,11 @@ mod tests { assert_ne!(database_identity().unwrap().0, [0u8; 16]); assert_ne!(segment_id().unwrap(), [0u8; 16]); assert_ne!(journal_id().unwrap(), [0u8; 16]); + assert_ne!(spill_epoch().unwrap(), [0u8; 16]); + } + + #[test] + fn spill_epochs_do_not_repeat_across_opens() { + assert_ne!(spill_epoch().unwrap(), spill_epoch().unwrap()); } } diff --git a/src/crypto/random/mod.rs b/src/crypto/random/mod.rs index 08169f4..a96160f 100644 --- a/src/crypto/random/mod.rs +++ b/src/crypto/random/mod.rs @@ -4,4 +4,4 @@ mod identity; #[cfg(not(target_arch = "wasm32"))] pub(crate) use identity::journal_id; -pub(crate) use identity::{database_identity, segment_id}; +pub(crate) use identity::{database_identity, segment_id, spill_epoch}; diff --git a/src/recovery/mod.rs b/src/recovery/mod.rs index 200b495..9ebfbd2 100644 --- a/src/recovery/mod.rs +++ b/src/recovery/mod.rs @@ -1,10 +1,11 @@ //! Open-flow recovery: apply-journal replay, catalog reconciliation, -//! tombstone GC. +//! tombstone GC, spill-scratch reclamation. pub(crate) mod deep_walk; pub(crate) mod gc; pub(crate) mod journal; pub(crate) mod reconcile; +pub(crate) mod scratch; #[cfg(test)] mod tests; diff --git a/src/recovery/scratch.rs b/src/recovery/scratch.rs new file mode 100644 index 0000000..f2fee03 --- /dev/null +++ b/src/recovery/scratch.rs @@ -0,0 +1,74 @@ +//! Reclamation of write-transaction spill scratch. +//! +//! A spill file is scratch for one live write transaction and is never read +//! back by a later one: its key is scoped to the open handle that wrote it. So +//! every scratch file present at open belongs to a handle that is already gone, +//! and is reclaimable without inspection. + +use crate::Result; +use crate::vfs::Vfs; + +/// Directory holding per-transaction spill scratch. +const SCRATCH_DIR: &str = "tmp"; + +/// Prefix of a spill scratch file name, completed by the transaction sequence. +const SCRATCH_PREFIX: &str = "scratch-"; + +/// Delete the spill scratch left by handles that closed without cleaning up. +/// +/// Callers must hold write authority over the store: this removes files, and a +/// handle that only observes must not. Returns the number of files deleted. +pub async fn delete_orphaned_scratch(vfs: &V) -> Result { + let entries = vfs.list_dir(SCRATCH_DIR).await?; + let mut count: u64 = 0; + for name in entries { + // Only the names this crate writes. `tmp/` is a shared directory in the + // store layout, and a sweep that removed everything under it would + // reclaim files belonging to whatever is added there next. + if !name.starts_with(SCRATCH_PREFIX) { + continue; + } + vfs.remove(&format!("{SCRATCH_DIR}/{name}")).await?; + count += 1; + } + if count > 0 { + // Durable directory entries: a crash between here and the next open + // would otherwise present the same scratch to be swept again. + vfs.sync_dir(SCRATCH_DIR).await?; + } + Ok(count) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::vfs::VfsFile; + use crate::vfs::memory::MemVfs; + use crate::vfs::types::OpenMode; + + async fn write(vfs: &MemVfs, path: &str) { + vfs.mkdir_all(SCRATCH_DIR).await.unwrap(); + let mut f = vfs.open(path, OpenMode::CreateNew).await.unwrap(); + f.write_at(0, b"scratch").await.unwrap(); + } + + #[tokio::test(flavor = "current_thread")] + async fn sweeps_scratch_and_spares_unrelated_names() { + let vfs = MemVfs::new(); + write(&vfs, "tmp/scratch-1").await; + write(&vfs, "tmp/scratch-9").await; + write(&vfs, "tmp/unrelated").await; + + assert_eq!(delete_orphaned_scratch(&vfs).await.unwrap(), 2); + + assert!(vfs.open("tmp/scratch-1", OpenMode::Read).await.is_err()); + assert!(vfs.open("tmp/scratch-9", OpenMode::Read).await.is_err()); + assert!(vfs.open("tmp/unrelated", OpenMode::Read).await.is_ok()); + } + + #[tokio::test(flavor = "current_thread")] + async fn absent_scratch_directory_is_not_an_error() { + let vfs = MemVfs::new(); + assert_eq!(delete_orphaned_scratch(&vfs).await.unwrap(), 0); + } +} diff --git a/src/txn/db/core.rs b/src/txn/db/core.rs index efbd3f5..759a276 100644 --- a/src/txn/db/core.rs +++ b/src/txn/db/core.rs @@ -177,6 +177,17 @@ pub struct Db { /// Each txn gets `txn_seq.fetch_add(1, Relaxed)` (first txn gets 1 since /// we start at 0 and add-then-use the pre-increment value + 1). pub(crate) txn_seq: AtomicU64, + /// Random per-open identity mixed into every spill key derivation. The + /// counter above restarts at 1 on each open and also supplies the spill + /// nonce, so this is what stops two opens of one store from encrypting + /// different scratch under the same key at the same nonce. + pub(crate) spill_epoch: [u8; 16], + /// Scratch paths belonging to write transactions that were dropped without + /// `commit` or `abort`. `Drop` cannot await the VFS removal, so it records + /// the exact path here and the next `begin_write` — which holds the writer + /// lock and can await — removes it. Anything still listed when the handle + /// closes is swept by the next open. + pub(crate) orphaned_spill_paths: parking_lot::Mutex>, /// The mode this handle was opened with (Standalone, Follower, `ReadOnly`, Observer). pub(crate) mode: DbMode, /// Set of reader `entry_id`s that have been aborted by `AbortOldest` stall diff --git a/src/txn/db/open/create.rs b/src/txn/db/open/create.rs index f2bebd4..dcb5d01 100644 --- a/src/txn/db/open/create.rs +++ b/src/txn/db/open/create.rs @@ -42,7 +42,7 @@ struct FreshDbState { } impl Db { - fn assemble_fresh(state: FreshDbState) -> Self { + fn assemble_fresh(state: FreshDbState) -> Result { let FreshDbState { pager, realm_id, @@ -83,7 +83,7 @@ impl Db { seq: initial.seq, }, ); - Self { + Ok(Self { pager: Arc::new(pager), realm_id, page_size, @@ -109,6 +109,8 @@ impl Db { mmap_bytes_in_use: Arc::new(AtomicU64::new(0)), spill_bytes_in_use: AtomicU64::new(0), txn_seq: AtomicU64::new(0), + spill_epoch: crate::crypto::random::spill_epoch()?, + orphaned_spill_paths: parking_lot::Mutex::new(Vec::new()), mode: DbMode::Standalone, aborted_readers: parking_lot::Mutex::new(std::collections::HashSet::new()), sentinel_locks: Vec::new(), @@ -132,7 +134,7 @@ impl Db { visibility_test_hook: parking_lot::Mutex::new(None), #[cfg(test)] rekey_test_fault: parking_lot::Mutex::new(None), - } + }) } /// Bootstrap a fresh database. Creates `main.db`, writes an initial A/B @@ -278,7 +280,7 @@ impl Db { let vfs_for_pager = V::clone(&*vfs_arc); let pager = Pager::open(vfs_for_pager, mk, cfg).await?; - Ok(Self::assemble_fresh(FreshDbState { + Self::assemble_fresh(FreshDbState { pager, realm_id: realm, page_size, @@ -291,6 +293,6 @@ impl Db { file_id, kek_salt, initial, - })) + }) } } diff --git a/src/txn/db/open/existing.rs b/src/txn/db/open/existing.rs index 5fb69a4..3cdc587 100644 --- a/src/txn/db/open/existing.rs +++ b/src/txn/db/open/existing.rs @@ -274,6 +274,8 @@ impl Db { mmap_bytes_in_use: Arc::new(AtomicU64::new(0)), spill_bytes_in_use: AtomicU64::new(0), txn_seq: AtomicU64::new(0), + spill_epoch: crate::crypto::random::spill_epoch()?, + orphaned_spill_paths: parking_lot::Mutex::new(Vec::new()), mode, aborted_readers: parking_lot::Mutex::new(std::collections::HashSet::new()), sentinel_locks: Vec::new(), diff --git a/src/txn/db/open/recovery.rs b/src/txn/db/open/recovery.rs index 4d75688..0cb80a9 100644 --- a/src/txn/db/open/recovery.rs +++ b/src/txn/db/open/recovery.rs @@ -32,6 +32,14 @@ pub(super) async fn recover_open_state( Err(PagedbError::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => {} Err(error) => return Err(error), } + + // Write-transaction spill scratch is keyed to the handle that wrote it, + // so anything still present belongs to a handle that is gone and is + // reclaimable outright. Without this the scratch of every transaction + // that died with its process would accumulate for the life of the + // store. Gated with the removals above because it needs write + // authority. + crate::recovery::scratch::delete_orphaned_scratch(&*db.vfs).await?; } // An incremental apply assembles its target beside `main.db` and renames it diff --git a/src/txn/write/spill.rs b/src/txn/write/spill.rs index 9da2ce7..d39be1c 100644 --- a/src/txn/write/spill.rs +++ b/src/txn/write/spill.rs @@ -39,17 +39,22 @@ pub(crate) struct SpillSegmentMeta { /// storage in a per-transaction tmp file. /// /// The tmp file lives at `tmp/scratch-`. It is created lazily on the -/// first `append` call and removed at `commit` or `abort` (best-effort). +/// first `append` call and removed at `commit` or `abort`. A transaction +/// dropped without either records its path for the next `begin_write` to +/// remove, and whatever a crash leaves behind is swept at the next open. /// /// Nonce scheme: `txn_seq_le6 ‖ segment_index_le6`. /// - First 6 bytes: the low 6 bytes of `txn_seq` in little-endian order. /// - Last 6 bytes: the low 6 bytes of the per-append segment index in /// little-endian order. /// -/// This guarantees uniqueness across all appends within a txn and across -/// independent txns (different `txn_seq`), without requiring a durable -/// nonce anchor (the tmp file is discarded before any subsequent txn can -/// reuse the same `txn_seq`). +/// Uniqueness under one key comes from `txn_seq` being strictly increasing +/// within an open handle and `segment_index` within a transaction. It does not +/// come from the file being deleted: `txn_seq` restarts at 1 on every open, so +/// deletion alone would leave two opens writing one nonce. The spill key is +/// instead derived per open, from a random epoch this handle generates and +/// never persists, so a repeated sequence lands under a different key and no +/// durable nonce anchor is needed. pub struct SpillScope<'scope, 'db, V: Vfs + Clone> { pub(super) txn: &'scope mut WriteTxn<'db, V>, } @@ -212,7 +217,12 @@ impl<'db, V: Vfs + Clone> WriteTxn<'db, V> { pub(crate) fn ensure_spill_cipher(&mut self) -> Result<&Cipher> { if self.spill_cipher.is_none() { let pager_mk = self.db.pager.mk()?; - let key = derive_spill_key(&pager_mk, &self.db.file_id, self.txn_seq)?; + let key = derive_spill_key( + &pager_mk, + &self.db.file_id, + &self.db.spill_epoch, + self.txn_seq, + )?; self.spill_cipher = Some(Cipher::new_aes_gcm(&key)); } self.spill_cipher.as_ref().ok_or(PagedbError::NotFound) @@ -238,10 +248,10 @@ impl<'db, V: Vfs + Clone> WriteTxn<'db, V> { /// Build the per-append nonce deterministically from `(txn_seq, segment_index)`. /// /// Layout: `txn_seq_le6 ‖ segment_index_le6` (6 + 6 = 12 bytes). - /// Uniqueness: nonces are unique per-txn (distinct `segment_index`) and - /// across txns (distinct `txn_seq` in first 6 bytes). No durable anchor - /// is needed because the tmp file is discarded before any key reuse - /// could matter. + /// Uniqueness within one key: distinct `segment_index` within a + /// transaction, distinct `txn_seq` across the transactions of one open + /// handle. `txn_seq` repeats across opens, which is why the key is scoped + /// to an open rather than to the store; see the type-level docs. pub(crate) fn next_spill_nonce(&self, segment_index: u64) -> Nonce { let mut bytes = [0u8; 12]; let seq_le = self.txn_seq.to_le_bytes(); diff --git a/src/txn/write/txn.rs b/src/txn/write/txn.rs index 00809d4..b6ae2fe 100644 --- a/src/txn/write/txn.rs +++ b/src/txn/write/txn.rs @@ -184,11 +184,18 @@ impl<'db, V: Vfs + Clone> WriteTxn<'db, V> { // for the first call), so we add 1 to produce 1-based ids. let txn_seq = db.txn_seq.fetch_add(1, Ordering::Relaxed) + 1; - // Drop cannot await VFS removal. Sweep the only scratch file a dropped - // predecessor could have left before this transaction can allocate its - // own path. Abort and commit already remove their files eagerly. - if txn_seq > 1 { - let _ = db.vfs.remove(&format!("tmp/scratch-{}", txn_seq - 1)).await; + // `Drop` cannot await VFS removal, so it leaves the exact paths of the + // scratch files it could not remove. This is the first point after + // those drops that both holds the writer lock and can await, so it is + // where they are collected. Removal is best effort for the same reason + // it is in abort and commit: a scratch file nothing references cannot + // make the durable store unopenable, and the next open sweeps whatever + // survives. Paths that fail to remove are not requeued — a path that + // cannot be removed now will not become removable by being retried on + // every subsequent transaction. + let orphaned = std::mem::take(&mut *db.orphaned_spill_paths.lock()); + for path in orphaned { + let _ = db.vfs.remove(&path).await; } Ok(Self { @@ -415,9 +422,19 @@ impl Drop for WriteTxn<'_, V> { fn drop(&mut self) { if !self.committed_or_aborted { self.db.pager.discard_dirty_main(self.db.realm_id); + // The gauge describes the live writer's scratch, and the writer + // lock this drop is about to release means there is no other one to + // account for. Leaving it set would report bytes of a transaction + // that no longer exists. self.db .spill_bytes_in_use .store(0, std::sync::atomic::Ordering::Relaxed); + // Removal has to await, which a drop cannot. Hand the exact path to + // the next `begin_write`; if the handle closes first, the next open + // sweeps it. + if let Some(path) = self.spill_path.take() { + self.db.orphaned_spill_paths.lock().push(path); + } } } } diff --git a/tests/commit_history.rs b/tests/commit_history.rs index 1748a21..4584580 100644 --- a/tests/commit_history.rs +++ b/tests/commit_history.rs @@ -146,7 +146,7 @@ async fn quota_update_preserves_history_across_reopen() { let opts = OpenOptions::default().with_commit_history_retain(RetainPolicy::Unbounded); let first_commit = { - let db = Db::open_internal_with_options(vfs.clone(), KEK, PAGE, REALM, opts.clone()) + let db = Db::open(vfs.clone(), KEK, PAGE, REALM, opts.clone()) .await .unwrap(); let ids = write_n(&db, 2).await; @@ -159,9 +159,7 @@ async fn quota_update_preserves_history_across_reopen() { ids[0] }; - let reopened = Db::open_existing_with_options(vfs, KEK, PAGE, REALM, opts) - .await - .unwrap(); + let reopened = Db::open(vfs, KEK, PAGE, REALM, opts).await.unwrap(); let historical = reopened .begin_read_at(first_commit) .await diff --git a/tests/spill_basic.rs b/tests/spill_basic.rs index db92ab1..b88df6e 100644 --- a/tests/spill_basic.rs +++ b/tests/spill_basic.rs @@ -109,7 +109,7 @@ async fn spill_cleanup_on_abort() { async fn spill_drop_resets_stats_and_next_write_sweeps_stale_tmp() { let vfs = MemVfs::new(); let opts = OpenOptions::default().with_scratch_bytes(1024 * 1024); - let db = Db::open_internal_with_options(vfs.clone(), [9u8; 32], PAGE, REALM, opts) + let db = Db::open(vfs.clone(), [9u8; 32], PAGE, REALM, opts) .await .unwrap(); { @@ -137,6 +137,37 @@ async fn spill_drop_resets_stats_and_next_write_sweeps_stale_tmp() { } } +#[tokio::test(flavor = "current_thread")] +async fn spill_scratch_outliving_a_handle_is_swept_at_the_next_open() { + let vfs = MemVfs::new(); + let opts = OpenOptions::default().with_scratch_bytes(1024 * 1024); + { + let db = Db::open(vfs.clone(), [9u8; 32], PAGE, REALM, opts.clone()) + .await + .unwrap(); + let mut w = db.begin_write().await.unwrap(); + let mut s = w.spill_scope(); + s.append(b"outlives-the-handle").await.unwrap(); + drop(s); + // Dropped with no further transaction to collect it, then the handle + // closes: the in-process path cannot reclaim this one. + drop(w); + drop(db); + } + assert!( + vfs.open("tmp/scratch-1", OpenMode::Read).await.is_ok(), + "the scratch file must genuinely survive the handle for this to test the open sweep" + ); + + let _db = Db::open(vfs.clone(), [9u8; 32], PAGE, REALM, opts) + .await + .unwrap(); + assert!( + vfs.open("tmp/scratch-1", OpenMode::Read).await.is_err(), + "open must reclaim spill scratch left by a handle that is gone" + ); +} + #[tokio::test(flavor = "current_thread")] async fn spill_aead_protects_payload() { // Confirm bytes on disk are NOT plaintext. From 1bb5ccc8fbfb7f0fb0b83faaab9a39d87d94cca5 Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Mon, 27 Jul 2026 20:38:52 +0800 Subject: [PATCH 3/3] refactor(pager): gate cache-eviction test helpers to test builds evict_clean_main_pages and its cache-layer counterpart are exercised only by tests, so mark them #[cfg(test)] rather than shipping them in the crate's public build surface. Move the test that exercises them into the unit tests beside Db, per this crate's convention that implementation-detail coverage lives with the code it tests. --- src/pager/cache.rs | 44 +++++++++++++++++++------------------------- src/pager/core.rs | 5 +++++ src/txn/db/misc.rs | 33 +++++++++++++++++++++++++++++++++ tests/stats_basic.rs | 32 -------------------------------- 4 files changed, 57 insertions(+), 57 deletions(-) diff --git a/src/pager/cache.rs b/src/pager/cache.rs index d030f2d..b9b1955 100644 --- a/src/pager/cache.rs +++ b/src/pager/cache.rs @@ -270,34 +270,22 @@ impl PageCache { /// when compaction replaces the backing file (the cached pages no longer /// match what is on disk). Pinned entries are left untouched. pub fn clear_file(&mut self, file: FileKey) { - let keys: Vec<(FileKey, u64)> = self - .map - .keys() - .filter(|(f, _)| *f == file) - .copied() - .collect(); - for key in keys { - if self.pins.get(&key).copied().unwrap_or(0) > 0 { - continue; - } - if let Some(idx) = self.map.remove(&key) { - let (prev, next) = { - let node = self.slab[idx].as_ref().expect("indexed node alive"); - (node.prev, node.next) - }; - if self.hand == Some(idx) { - self.hand = next; - } - self.unlink_node(idx, prev, next); - self.free_node(idx); - } - self.dirty.remove(&key); - } + self.clear_file_entries(file, false); } /// Drop every unpinned clean entry for `file`, leaving dirty entries /// available for a later flush and pinned entries available to readers. + /// Used to force a re-read of the durable bytes without discarding the + /// in-flight writes that have not reached them yet. + #[cfg(test)] pub fn clear_clean_file(&mut self, file: FileKey) { + self.clear_file_entries(file, true); + } + + /// Shared body of the two `clear_*` entry points. Pinned entries are always + /// spared: a reader holds them. `keep_dirty` decides whether an unflushed + /// entry is spared as well, or dropped along with its dirty marker. + fn clear_file_entries(&mut self, file: FileKey, keep_dirty: bool) { let keys: Vec<(FileKey, u64)> = self .map .keys() @@ -305,7 +293,10 @@ impl PageCache { .copied() .collect(); for key in keys { - if self.pins.get(&key).copied().unwrap_or(0) > 0 || self.dirty.contains(&key) { + if self.pins.get(&key).copied().unwrap_or(0) > 0 { + continue; + } + if keep_dirty && self.dirty.contains(&key) { continue; } if let Some(idx) = self.map.remove(&key) { @@ -319,6 +310,9 @@ impl PageCache { self.unlink_node(idx, prev, next); self.free_node(idx); } + if !keep_dirty { + self.dirty.remove(&key); + } } } } @@ -407,7 +401,7 @@ mod tests { assert!(c.get((FileKey::Main, 1)).is_none()); assert!(c.get((FileKey::Main, 2)).is_some()); - assert!(c.is_dirty((FileKey::Main, 2))); + assert_eq!(c.dirty_for_file(FileKey::Main), vec![2]); } #[test] diff --git a/src/pager/core.rs b/src/pager/core.rs index b97a42c..3eb7e3a 100644 --- a/src/pager/core.rs +++ b/src/pager/core.rs @@ -872,6 +872,11 @@ impl Pager { /// Evict unpinned clean main.db pages so later reads fetch and authenticate /// durable bytes without discarding in-flight writes. + /// + /// Compiled only for the crate's own tests, matching the handle-level + /// accessor that is its sole caller: correctness never depends on whether a + /// page is warm, so nothing in the shipped crate should be steering that. + #[cfg(test)] pub fn evict_clean_main_pages(&self, _realm_id: crate::RealmId) { self.inner .buffer_pool diff --git a/src/txn/db/misc.rs b/src/txn/db/misc.rs index ee55715..262b91c 100644 --- a/src/txn/db/misc.rs +++ b/src/txn/db/misc.rs @@ -219,6 +219,39 @@ mod tests { const PAGE: usize = 4096; const REALM: RealmId = RealmId::new([0xA5; 16]); + #[tokio::test(flavor = "current_thread")] + async fn evict_main_pages_forces_next_read_to_miss_cache() { + let db = Db::open_internal(MemVfs::new(), [9u8; 32], PAGE, REALM) + .await + .unwrap(); + { + let mut txn = db.begin_write().await.unwrap(); + txn.put(b"hello", b"world").await.unwrap(); + txn.commit().await.unwrap(); + } + { + let reader = db.begin_read().await.unwrap(); + assert_eq!( + reader.get(b"hello").await.unwrap().as_deref(), + Some(b"world".as_slice()) + ); + } + let before = db.stats().await.unwrap(); + + db.evict_main_pages(REALM); + + let reader = db.begin_read().await.unwrap(); + assert_eq!( + reader.get(b"hello").await.unwrap().as_deref(), + Some(b"world".as_slice()) + ); + let after = db.stats().await.unwrap(); + assert!( + after.buffer_pool_misses > before.buffer_pool_misses, + "eviction must force a VFS-backed cache miss: before={before:?}, after={after:?}" + ); + } + #[tokio::test(flavor = "current_thread")] async fn stats_surfaces_malformed_segment_catalog_row() { let db = Db::open_internal(MemVfs::new(), [9u8; 32], PAGE, REALM) diff --git a/tests/stats_basic.rs b/tests/stats_basic.rs index aa1bcd9..b68e3d9 100644 --- a/tests/stats_basic.rs +++ b/tests/stats_basic.rs @@ -117,38 +117,6 @@ async fn stats_reports_buffer_pool() { ); } -#[tokio::test] -async fn evict_main_pages_forces_next_read_to_miss_cache() { - let db = fresh_db().await; - { - let mut txn = db.begin_write().await.unwrap(); - txn.put(b"hello", b"world").await.unwrap(); - txn.commit().await.unwrap(); - } - - { - let reader = db.begin_read().await.unwrap(); - assert_eq!( - reader.get(b"hello").await.unwrap().as_deref(), - Some(b"world".as_slice()) - ); - } - let before = db.stats().await.unwrap(); - - db.evict_main_pages(realm()); - - let reader = db.begin_read().await.unwrap(); - assert_eq!( - reader.get(b"hello").await.unwrap().as_deref(), - Some(b"world".as_slice()) - ); - let after = db.stats().await.unwrap(); - assert!( - after.buffer_pool_misses > before.buffer_pool_misses, - "eviction must force a VFS-backed cache miss: before={before:?}, after={after:?}" - ); -} - #[tokio::test] async fn stats_reports_mode() { let vfs = MemVfs::new();