fix(snapshot): authenticate exact incremental state

CHANGELOG.md

@@ -4,9 +4,11 @@ All notable changes to this project are documented here. The format is based on
 [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project
 adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [0.1.0] - 2026-06-07
+## [Unreleased]
 
-Initial release.
+The first stable release. Pre-releases are published as `0.1.0-beta.N`; the
+entries below describe `0.1.0` as a whole rather than deltas against a shipped
+version, since none exists yet.
 
 ### Added
 
@@ -28,7 +30,9 @@ Initial release.
   (Grand Central Dispatch), Android, WASM/OPFS, and WASI backends, plus a
   tokio thread-pool fallback and an in-memory backend, with format-bit identity
   across targets.
-- **Snapshots** — `snapshot_to`, `restore_from`, and incremental apply.
+- **Snapshots** — `snapshot_to`, `restore_from`, and incremental apply, each
+  authenticated against the exact state its manifest describes. Destinations
+  must be empty; malformed or incomplete artifacts fail closed.
 - **Recovery** — open-flow GC, apply-journal replay, deep-walk `fsck`, and the
   `pagedb-fsck` binary.
 - **Online rekey** — rekey the database under a new key with mixed-cipher and
@@ -41,5 +45,7 @@ Initial release.
 - 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.
-
-[0.1.0]: https://github.com/nodedb-lab/pagedb/releases/tag/v0.1.0
+- An incremental snapshot cannot always be applied: if the producer recycled a
+  page the follower's own free-list chain or commit-history tree still hosts,
+  apply reports `PagedbError::SnapshotBasePageReused` and the follower needs a
+  full snapshot or a nearer base commit.

src/errors.rs

@@ -56,6 +56,17 @@ 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.
+    ///
+    /// 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`].
+    #[error("incremental snapshot would overwrite base-live page {page_id}")]
+    SnapshotBasePageReused { page_id: u64 },
+
     #[error("commit {commit:?} is durable but unpublished; reopen required")]
     DurablyCommittedButUnpublished { commit: CommitId },
 
@@ -433,6 +444,12 @@ impl PagedbError {
         Self::SnapshotIncompatible { field }
     }
 
+    /// Canonical constructor for a target state that reuses a base-live page.
+    #[must_use]
+    pub const fn snapshot_base_page_reused(page_id: u64) -> Self {
+        Self::SnapshotBasePageReused { page_id }
+    }
+
     /// Canonical constructor for a handle whose newest durable commit could
     /// not be reconciled into its reader-visible state.
     #[must_use]

src/segment/reader.rs

@@ -85,6 +85,29 @@ impl<V: Vfs + Clone> SegmentReader<V> {
         .await
     }
 
+    #[cfg(not(target_arch = "wasm32"))]
+    pub(crate) async fn open_internal_at_path(
+        pager: Arc<Pager<V>>,
+        catalog_meta: SegmentMeta,
+        path: &str,
+        mmap_budget_used: std::sync::Arc<std::sync::atomic::AtomicU64>,
+        mmap_budget_limit: u64,
+    ) -> Result<Self> {
+        let page_size = pager.page_size();
+        let file = pager.vfs().open(path, OpenMode::Read).await?;
+        Self::finish_open(
+            pager,
+            catalog_meta,
+            page_size,
+            file,
+            MmapBudget {
+                used: mmap_budget_used,
+                limit: mmap_budget_limit,
+            },
+        )
+        .await
+    }
+
     /// Open a sealed replacement from either publication location. The
     /// replacement identity is durable progress, so a missing or malformed
     /// file must fail closed rather than being regenerated.

src/snapshot/apply.rs

@@ -3,13 +3,22 @@
 
 #![cfg(not(target_arch = "wasm32"))]
 
-use std::path::Path;
+use std::{collections::BTreeSet, path::Path};
 
 use tokio::fs;
 use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
 
 use crate::Result;
 use crate::errors::PagedbError;
+use crate::pager::page_space::is_reserved;
+
+/// Pages consumed from one `pages.delta` stream during an incremental apply.
+pub struct AppliedDeltaPages {
+    /// Number of page records written to `main.db`.
+    pub pages_applied: u64,
+    /// Page IDs supplied by this exact delta stream.
+    pub page_ids: BTreeSet<u64>,
+}
 
 /// Apply an incremental snapshot directory (`src_path`) to the Follower's
 /// `main.db` file at `main_db_path` (absolute filesystem path). Returns stats.
@@ -20,54 +29,113 @@ pub async fn apply_delta_pages(
     src_path: &Path,
     dst_main_db_path: &Path,
     page_size: usize,
-) -> Result<u64> {
+    protected_page_ids: &BTreeSet<u64>,
+    target_next_page_id: u64,
+) -> Result<AppliedDeltaPages> {
     let delta_path = src_path.join("pages.delta");
     let mut delta = match fs::File::open(&delta_path).await {
         Ok(f) => f,
-        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(0),
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
+            return Ok(AppliedDeltaPages {
+                pages_applied: 0,
+                page_ids: BTreeSet::new(),
+            });
+        }
         Err(e) => return Err(PagedbError::Io(e)),
     };
 
+    let page_size_u64 = u64::try_from(page_size)
+        .map_err(|_| PagedbError::snapshot_artifact_invalid("pages.delta.page_size"))?;
+    let record_size = 8u64
+        .checked_add(page_size_u64)
+        .ok_or_else(|| PagedbError::snapshot_artifact_invalid("pages.delta.record_size"))?;
+    let delta_len = delta.metadata().await.map_err(PagedbError::Io)?.len();
+    if delta_len % record_size != 0 {
+        return Err(PagedbError::snapshot_artifact_invalid("pages.delta.length"));
+    }
+
+    // Validate every record before opening main.db for writes. A malformed
+    // record late in the stream must not leave a valid prefix written into the
+    // follower's future page range.
+    let record_count = delta_len / record_size;
+    let mut page_ids = BTreeSet::new();
+    let mut id_buf = [0u8; 8];
+    let page_skip = i64::try_from(page_size)
+        .map_err(|_| PagedbError::snapshot_artifact_invalid("pages.delta.page_size"))?;
+    for _ in 0..record_count {
+        delta
+            .read_exact(&mut id_buf)
+            .await
+            .map_err(PagedbError::Io)?;
+        let page_id = u64::from_be_bytes(id_buf);
+        // Reserved pages are the A/B headers and the apply-journal slots. A
+        // delta record naming one would overwrite the very state that makes the
+        // apply recoverable, so it is rejected by identity rather than by
+        // happening to fall under some allocation bound.
+        if is_reserved(page_id) || page_id >= target_next_page_id {
+            return Err(PagedbError::snapshot_artifact_invalid(
+                "pages.delta.page_id",
+            ));
+        }
+        // The follower keeps its own free-list chain and commit-history tree
+        // across an apply, and those pages are invisible to the producer, which
+        // can neither predict nor avoid them. A collision is therefore not a
+        // malformed artifact — both states are internally sound, they simply
+        // cannot be related by a page delta — so it reports as
+        // `SnapshotBasePageReused` and the remedy is a full snapshot or a nearer
+        // base. Caught here, before `main.db` is opened for writing.
+        if protected_page_ids.contains(&page_id) {
+            return Err(PagedbError::snapshot_base_page_reused(page_id));
+        }
+        if !page_ids.insert(page_id) {
+            return Err(PagedbError::snapshot_artifact_invalid(
+                "pages.delta.duplicate_page_id",
+            ));
+        }
+        delta
+            .seek(std::io::SeekFrom::Current(page_skip))
+            .await
+            .map_err(PagedbError::Io)?;
+    }
+    delta
+        .seek(std::io::SeekFrom::Start(0))
+        .await
+        .map_err(PagedbError::Io)?;
+
     let mut dst = fs::OpenOptions::new()
         .read(true)
         .write(true)
         .open(dst_main_db_path)
         .await
         .map_err(PagedbError::Io)?;
-
-    let mut pages_applied: u64 = 0;
-    let mut id_buf = [0u8; 8];
     let mut page_buf = vec![0u8; page_size];
-
-    loop {
-        // Read page_id (8 bytes BE).
-        match delta.read_exact(&mut id_buf).await {
-            Ok(_) => {}
-            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
-            Err(e) => return Err(PagedbError::Io(e)),
-        }
+    for _ in 0..record_count {
+        delta
+            .read_exact(&mut id_buf)
+            .await
+            .map_err(PagedbError::Io)?;
         let page_id = u64::from_be_bytes(id_buf);
-
-        // Read page bytes.
-        match delta.read_exact(&mut page_buf).await {
-            Ok(_) => {}
-            Err(e) => return Err(PagedbError::Io(e)),
-        }
+        delta
+            .read_exact(&mut page_buf)
+            .await
+            .map_err(PagedbError::Io)?;
 
         // Write page to main.db at the correct offset.
         let offset = page_id
-            .checked_mul(page_size as u64)
-            .ok_or_else(|| PagedbError::Io(std::io::Error::other("page offset overflow")))?;
+            .checked_mul(page_size_u64)
+            .ok_or_else(|| PagedbError::snapshot_artifact_invalid("pages.delta.page_offset"))?;
         dst.seek(std::io::SeekFrom::Start(offset))
             .await
             .map_err(PagedbError::Io)?;
         dst.write_all(&page_buf).await.map_err(PagedbError::Io)?;
-        pages_applied += 1;
     }
 
     dst.flush().await.map_err(PagedbError::Io)?;
     dst.sync_all().await.map_err(PagedbError::Io)?;
-    Ok(pages_applied)
+    Ok(AppliedDeltaPages {
+        pages_applied: record_count,
+        page_ids,
+    })
 }
 
 /// Verify that the snapshot's segment directory has exactly the count claimed
@@ -90,9 +158,20 @@ pub(crate) async fn validate_snapshot_segment_count(src_path: &Path, expected: u
 pub async fn stage_snapshot_segments(
     src_path: &Path,
     dst_seg_root: &Path,
+    expected_segment_ids: &BTreeSet<[u8; 16]>,
 ) -> Result<Vec<[u8; 16]>> {
     let entries = snapshot_segment_entries(src_path).await?;
     let seg_src = src_path.join("seg");
+    let actual_segment_ids: BTreeSet<[u8; 16]> = entries
+        .iter()
+        .map(|name| {
+            crate::hex::parse_hex::<16>(name)
+                .ok_or_else(|| PagedbError::snapshot_artifact_invalid("segment_file_name"))
+        })
+        .collect::<Result<_>>()?;
+    if &actual_segment_ids != expected_segment_ids {
+        return Err(PagedbError::snapshot_artifact_invalid("segments"));
+    }
 
     let staging_dir = dst_seg_root.join(".staging");
     fs::create_dir_all(&staging_dir)
@@ -103,7 +182,6 @@ pub async fn stage_snapshot_segments(
     let mut copy_buf = vec![0u8; 64 * 1024];
 
     for name in &entries {
-        // Each name in seg/ is 32 hex chars encoding the 16-byte segment id.
         let segment_id = crate::hex::parse_hex::<16>(name)
             .ok_or_else(|| PagedbError::snapshot_artifact_invalid("segment_file_name"))?;
         let src_file = seg_src.join(name);