1 files changed, 117 insertions, 1 deletions

diff --git a/src/compaction/snapshot.rs b/src/compaction/snapshot.rs
index 1f57352..86caba9 100644
--- a/src/compaction/snapshot.rs
+++ b/src/compaction/snapshot.rs
@@ -18,7 +18,8 @@
 //! at the snapshot change itself, so edits made afterwards through the store API can't
 //! change what the protocol sees.
 
-// `Snapshot::clock` is used by returning nodes (step 7).
+// `Snapshot::clock` is kept for returning nodes (step 7); late nodes read the clock from
+// the metadata instead.
 #![cfg_attr(not(test), allow(dead_code))]
 
 use std::collections::BTreeMap;
@@ -27,6 +28,7 @@ use automerge::{transaction::CommitOptions, ActorId, AutoCommit, ChangeHash};
 use serde_json::{json, Map as JsonMap, Value as JsonValue};
 use sha2::{Digest, Sha256};
 
+use super::records::{EpochRecord, Policy};
 use crate::replica::Epoch;
 use crate::store::{self, StoreError};
 
@@ -105,6 +107,67 @@ pub(crate) async fn build_detached(
     .map_err(|error| SnapshotError::Task(error.to_string()))?
 }
 
+/// The protocol's view of an epoch, read at its snapshot change.
+#[derive(Clone, Debug)]
+pub(crate) struct EpochMetadata {
+    pub(crate) compaction_id: String,
+    pub(crate) previous: Epoch,
+    pub(crate) clock: BTreeMap<String, u64>,
+    pub(crate) policy: Policy,
+}
+
+#[derive(serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct MetadataRecord {
+    generation: u64,
+    compaction_id: String,
+    previous: EpochRecord,
+    clock: BTreeMap<String, u64>,
+    policy: Policy,
+}
+
+/// Checks that `doc` is a document of `epoch`: its history starts with exactly the
+/// snapshot change `epoch.id` (no dependencies), and the metadata at that change names
+/// the same generation. Returns the metadata.
+///
+/// This does not prove the snapshot's content matches the old epoch; the hash commits
+/// to it, and the peer is a cluster member trusted as much as for ordinary sync.
+pub(crate) fn verify_document(doc: &mut AutoCommit, epoch: Epoch) -> Result<EpochMetadata, String> {
+    let root = ChangeHash(epoch.id);
+    let Some(change) = doc.get_change_by_hash(&root) else {
+        return Err(format!(
+            "document does not contain the snapshot change of {epoch}"
+        ));
+    };
+    if !change.deps().is_empty() {
+        return Err(format!("snapshot change of {epoch} has dependencies"));
+    }
+
+    let at_snapshot = doc.fork_at(&[root]).map_err(|error| error.to_string())?;
+    let value = store::read_path(&at_snapshot, &epoch_metadata_path(epoch.generation))
+        .map_err(|error| error.to_string())?
+        .ok_or_else(|| format!("no epoch metadata for generation {}", epoch.generation))?;
+    let record: MetadataRecord = serde_json::from_value(value)
+        .map_err(|error| format!("invalid epoch metadata: {error}"))?;
+    if record.generation != epoch.generation {
+        return Err(format!(
+            "epoch metadata names generation {} instead of {}",
+            record.generation, epoch.generation
+        ));
+    }
+    let previous = record
+        .previous
+        .to_epoch()
+        .ok_or_else(|| "invalid previous epoch in metadata".to_owned())?;
+
+    Ok(EpochMetadata {
+        compaction_id: record.compaction_id,
+        previous,
+        clock: record.clock,
+        policy: record.policy,
+    })
+}
+
 /// Path of the metadata record for an epoch generation.
 pub(crate) fn epoch_metadata_path(generation: u64) -> Vec<String> {
     vec![
@@ -493,4 +556,57 @@ mod tests {
         assert!(snap.clock.is_empty());
         assert_eq!(user_view(&snap.doc), json!({}));
     }
+
+    #[test]
+    fn fetched_document_is_verified_and_late_changes_replayed() {
+        use crate::compaction::delta;
+
+        // Old epoch: the base, then a late write by this node.
+        let mut late = doc(b"late-node");
+        late.put(ROOT, "shared", "base").unwrap();
+        late.commit();
+        let base = late.get_heads();
+        let policy = json!({});
+        let mut snap = build(&mut late, &params(&base, &policy)).unwrap();
+
+        // Meanwhile the cluster moved on in the new epoch.
+        let mut cluster = snap
+            .doc
+            .fork()
+            .with_actor(ActorId::from(&b"cluster-node"[..]));
+        cluster.put(ROOT, "new", 1_i64).unwrap();
+        cluster.commit();
+        let bytes = cluster.save();
+
+        late.put(ROOT, "mine", "late").unwrap();
+        late.commit();
+
+        let mut fetched = AutoCommit::load(&bytes).unwrap();
+        let metadata = verify_document(&mut fetched, snap.epoch).unwrap();
+        assert_eq!(metadata.previous, Epoch::GENESIS);
+        assert_eq!(metadata.compaction_id, "c-1");
+        assert_eq!(
+            metadata.policy.replay_window,
+            Policy::default_replay_window()
+        );
+
+        // Documents of other epochs are refused.
+        let mut wrong = snap.epoch;
+        wrong.generation = 2;
+        assert!(verify_document(&mut fetched, wrong).is_err());
+        let mut wrong = snap.epoch;
+        wrong.id = [0; 32];
+        assert!(verify_document(&mut fetched, wrong).is_err());
+
+        let own = late.get_actor().clone();
+        let missing = delta::missing_from_snapshot(&mut late, &metadata.clock, &own);
+        assert_eq!(missing.own, 1);
+        let heads = late.get_heads();
+        let mut adopted = fetched.with_actor(ActorId::from(&b"late-node-next"[..]));
+        delta::replay(&mut late, &missing.before, &heads, &mut adopted);
+        assert_eq!(
+            user_view(&adopted),
+            json!({ "shared": "base", "new": 1, "mine": "late" })
+        );
+    }
 }