fix(snapshot): stage incremental apply into a scratch image copy

Writing the target state directly into the live main.db meant a
follower's own free-list chain or commit-history tree could collide
with pages the incoming delta claims, so any such collision had to be
refused outright. Building the target in a scratch copy alongside
main.db and swapping it in with a rename removes that constraint:
colliding structures are relocated onto fresh ids instead, so the
apply only refuses a delta when the collision is genuinely
unrelocatable base-live state.

Splits snapshot::apply into a module directory (delta streaming,
image staging, segment staging) and adds a read-only Pager view plus
an EpochKeyring snapshot so the target image's trees can be
authenticated before it goes live.

Author: farhan-syah

CHANGELOG.md

@@ -53,10 +53,6 @@ version, since none exists yet.
 - Single-writer per database; multi-writer cross-process is not supported.
 - Writes carry per-page AEAD and copy-on-write overhead; for throughput-bound
   plaintext KV workloads a generic store may be faster.
-- An incremental snapshot cannot always be applied: if the producer recycled any
-  page the follower's base state still holds, apply reports
-  `PagedbError::SnapshotBasePageReused` and the follower needs a full snapshot or
-  a nearer base commit. Copy-on-write recycles page ids continuously, so a
-  follower should expect this within a few deltas under ordinary write churn and
-  must be able to fall back to a full snapshot. The refusal is checked before any
-  page is written, so it leaves the follower unchanged.
+- Applying an incremental snapshot stages a full copy of `main.db` alongside the
+  original and commits it with a rename, so a follower needs room for two copies
+  of the database during an apply.

src/errors.rs

@@ -56,14 +56,23 @@ pub enum PagedbError {
     #[error("incremental snapshot is incompatible: {field}")]
     SnapshotIncompatible { field: &'static str },
 
-    /// The target state reuses a page the base commit still hosts, so the delta
-    /// cannot be expressed without overwriting state a follower on that base
-    /// still depends on.
+    /// A delta record names a page the base commit's readers can still reach.
     ///
-    /// Not a corruption: both states are internally sound. They simply cannot be
-    /// related by a page delta, and the correct remedy is a full snapshot or a
-    /// nearer base commit — which is why this is distinct from
-    /// [`CorruptionDetail::SnapshotArtifactInvalid`].
+    /// A delta is defined as target-reachable minus base-reader-visible, so no
+    /// well-formed export produces such a record: the pages both sides of the
+    /// protocol can name are exactly the ones it must not carry. Raised before
+    /// anything is staged.
+    ///
+    /// Recycling ids the *follower's own* free-list chain or commit-history tree
+    /// occupies is not this condition and never raises it. The producer cannot
+    /// see those pages, so it cannot avoid them; the apply relocates both
+    /// structures out of the incoming page space instead of refusing an
+    /// otherwise-healthy delta.
+    ///
+    /// Not a corruption of this store: both states are internally sound, they
+    /// simply cannot be related by this delta — which is why this is distinct
+    /// from [`CorruptionDetail::SnapshotArtifactInvalid`]. The remedy is a full
+    /// snapshot or a nearer base commit.
     #[error("incremental snapshot would overwrite base-live page {page_id}")]
     SnapshotBasePageReused { page_id: u64 },
 

src/pager/core.rs

@@ -224,6 +224,59 @@ impl KindBinding {
     }
 }
 
+impl<V: Vfs + Clone> Pager<V> {
+    /// A read-only Pager over an alternate main.db image at `image_path`.
+    ///
+    /// An incremental apply builds the target image in a scratch file and has to
+    /// authenticate the producer's trees *inside* that image before the rename
+    /// that makes it live. The live Pager cannot serve those reads: it is still
+    /// answering base-image reads for concurrent Follower readers, and
+    /// repointing it would hand them target bytes for a state no durable header
+    /// names yet.
+    ///
+    /// The view carries a snapshot of the epoch keyring, so a page sealed under
+    /// any epoch the live handle can decrypt opens here too. It is read-only, so
+    /// it can never issue a nonce or write to the image — the single nonce
+    /// counter stays with the live Pager, and no page is ever sealed twice.
+    ///
+    /// The view's buffer pool is sized by the same `OpenOptions` budget as the
+    /// live one. Callers drop cached main pages before opening a view and drop
+    /// the view before the swap, so the two pools are not resident at full size
+    /// at the same time.
+    // Only the incremental-apply staging path opens a view, and that path is
+    // native-only.
+    #[cfg(not(target_arch = "wasm32"))]
+    pub(crate) fn open_main_view(&self, image_path: String) -> Self {
+        let mut cfg = self.cfg.clone();
+        cfg.main_db_path = image_path;
+        let inner = Arc::new(PagerInner {
+            buffer_pool: parking_lot::Mutex::new(PageCache::with_capacity(cfg.buffer_pool_pages)),
+            // A view exists to walk main-db trees; no caller routes a segment or
+            // journal page through it, so its second cache class is given the
+            // smallest legal budget rather than a copy of the configured one.
+            segment_cache: parking_lot::Mutex::new(PageCache::with_capacity(1)),
+            buffer_pool_hits: AtomicU64::new(0),
+            buffer_pool_misses: AtomicU64::new(0),
+            metrics_enabled: false,
+        });
+        let main_nonce = MainDbNonceGen::new(&cfg.main_db_file_id, cfg.anchor_budget);
+        Self {
+            dek_lru: parking_lot::Mutex::new(DekLru::with_capacity(cfg.dek_lru_capacity)),
+            main_nonce: parking_lot::Mutex::new(main_nonce),
+            segment_nonces: parking_lot::Mutex::new(BTreeMap::new()),
+            journal_nonces: parking_lot::Mutex::new(BTreeMap::new()),
+            files: AsyncMutex::new(BTreeMap::new()),
+            inner,
+            active_epoch: AtomicU64::new(self.active_epoch.load(AtomOrd::SeqCst)),
+            read_only: AtomicBool::new(true),
+            keyring: self.keyring.duplicate(),
+            observer_retry_count: self.observer_retry_count,
+            vfs: self.vfs.clone(),
+            cfg,
+        }
+    }
+}
+
 impl<V: Vfs> Pager<V> {
     pub(crate) fn page_size(&self) -> usize {
         self.cfg.page_size