fix(txn): preserve cache and transaction lifecycle state

Author: presempathy-awb

src/pager/cache.rs

@@ -168,9 +168,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 {
@@ -318,6 +319,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)]
@@ -393,6 +421,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);
@@ -421,4 +463,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

@@ -653,6 +653,15 @@ impl<V: Vfs> Pager<V> {
         }
     }
 
+    /// 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,

src/txn/db/catalog.rs

@@ -146,8 +146,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/misc.rs

@@ -58,11 +58,11 @@ impl<V: Vfs + Clone> Db<V> {
         self.snapshot.read().next_page_id
     }
 
-    /// Evict all clean and dirty pages for the main realm from the buffer
-    /// pool. Intended for integration tests that corrupt pages on disk and
-    /// want subsequent reads to see the disk contents rather than cached data.
+    /// Evict unpinned clean pages for the main realm from the buffer pool.
+    /// Intended for integration tests that corrupt durable pages and want
+    /// subsequent reads to see disk contents without dropping in-flight writes.
     pub fn evict_main_pages(&self, realm: crate::RealmId) {
-        self.pager.discard_dirty_main(realm);
+        self.pager.evict_clean_main_pages(realm);
     }
 
     /// Return the current size of `main.db` in bytes. Useful for tests that

src/txn/write/txn.rs

@@ -154,6 +154,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,
@@ -377,6 +385,9 @@ impl<V: Vfs + Clone> 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);
         }
     }
 }

tests/commit_history.rs

@@ -3,7 +3,7 @@ use std::time::Duration;
 use pagedb::errors::PagedbError;
 use pagedb::options::{OpenOptions, RetainPolicy};
 use pagedb::vfs::memory::MemVfs;
-use pagedb::{CommitId, Db, RealmId};
+use pagedb::{CommitId, Db, RealmId, RealmQuotas};
 
 const PAGE: usize = 4096;
 const KEK: [u8; 32] = [7u8; 32];
@@ -144,6 +144,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

tests/spill_basic.rs

@@ -108,6 +108,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.

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