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<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();
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/pager/cache.rs b/src/pager/cache.rs
index d2a6830..b9b1955 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 {
@@ -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());
+    }
 }
diff --git a/src/pager/core.rs b/src/pager/core.rs
index 4719200..3eb7e3a 100644
--- a/src/pager/core.rs
+++ b/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,
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<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);
+    }
+}
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<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,
         })?;
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<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
diff --git a/src/txn/db/misc.rs b/src/txn/db/misc.rs
index 2e6776f..262b91c 100644
--- a/src/txn/db/misc.rs
+++ b/src/txn/db/misc.rs
@@ -40,7 +40,8 @@ impl<V: Vfs + Clone> Db<V> {
             .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<V: Vfs + Clone> Db<V> {
     /// 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
@@ -218,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/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<V: Vfs + Clone> {
 }
 
 impl<V: Vfs + Clone> Db<V> {
-    fn assemble_fresh(state: FreshDbState<V>) -> Self {
+    fn assemble_fresh(state: FreshDbState<V>) -> Result<Self> {
         let FreshDbState {
             pager,
             realm_id,
@@ -83,7 +83,7 @@ impl<V: Vfs + Clone> Db<V> {
                 seq: initial.seq,
             },
         );
-        Self {
+        Ok(Self {
             pager: Arc::new(pager),
             realm_id,
             page_size,
@@ -109,6 +109,8 @@ impl<V: Vfs + Clone> Db<V> {
             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<V: Vfs + Clone> Db<V> {
             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<V: Vfs + Clone> Db<V> {
         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<V: Vfs + Clone> Db<V> {
             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<V: Vfs + Clone> Db<V> {
             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<V: Vfs + Clone>(
             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-<txn_seq>`. 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 a4bc94d..b6ae2fe 100644
--- a/src/txn/write/txn.rs
+++ b/src/txn/write/txn.rs
@@ -183,6 +183,21 @@ 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, 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 {
             db,
             guard,
@@ -407,6 +422,19 @@ 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);
+            // 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 011c2aa..4584580 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,36 @@ 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(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(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..b88df6e 100644
--- a/tests/spill_basic.rs
+++ b/tests/spill_basic.rs
@@ -105,6 +105,69 @@ 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(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_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.