fix(txn): preserve cache and transaction lifecycle state

src/crypto/kdf.rs

@@ -57,14 +57,35 @@ pub fn derive_hk(mk: &MasterKey) -> Result<DerivedKey> {
 
 /// 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<DerivedKey> {
+/// `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<DerivedKey> {
     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();

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());
     }
 }

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};

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 {
@@ -269,6 +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) {
+        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()
@@ -279,6 +296,9 @@ impl PageCache {
             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) {
                 let (prev, next) = {
                     let node = self.slab[idx].as_ref().expect("indexed node alive");
@@ -290,7 +310,9 @@ impl PageCache {
                 self.unlink_node(idx, prev, next);
                 self.free_node(idx);
             }
-            self.dirty.remove(&key);
+            if !keep_dirty {
+                self.dirty.remove(&key);
+            }
         }
     }
 }
@@ -368,6 +390,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_eq!(c.dirty_for_file(FileKey::Main), vec![2]);
+    }
+
     #[test]
     fn dirty_iter_is_sorted_ascending() {
         let mut c = PageCache::with_capacity(16);
@@ -396,4 +432,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());
+    }
 }

src/pager/core.rs

@@ -870,6 +870,20 @@ impl<V: Vfs> Pager<V> {
         }
     }
 
+    /// 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
+            .lock()
+            .clear_clean_file(FileKey::Main);
+    }
+
     fn write_page(
         &self,
         file: FileKey,

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;
 

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<V: Vfs + Clone>(vfs: &V) -> Result<u64> {
+    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);
+    }
+}

src/txn/db/catalog.rs

@@ -151,8 +151,8 @@ impl<V: Vfs + Clone> Db<V> {
             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,
         })?;

src/txn/db/core.rs

@@ -177,6 +177,17 @@ pub struct Db<V: Vfs + Clone> {
     /// 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<Vec<String>>,
     /// 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