remake: earlier epochs' metadata read as the snapshot built it

A snapshot leaving out earlier epochs' metadata (`snapshot::left_out`) and a rejoining
node (`catchup::prepare`) read it from the document as it stands. It is never writable
through the store, and under a policy the rules take a change to it only from an
epoch's first; but with no policy, screening judges only an epoch's root, so a member's
own change could rewrite or remove an entry: a replay window narrowed or widened, a
translation or a clock changed, a strict remake's entry gone.

Now both read it at the epoch's snapshot change (`snapshot::as_built`), as
`verify_document` reads the epoch's own: a snapshot from its base's epoch's, a
rejoining node from the target's. Every earlier entry is a copy in that change, so what
was written there since doesn't count. The first epoch follows none and has none.

2 files changed, 55 insertions, 9 deletions

diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index a3443ba..68fa117 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -520,12 +520,19 @@ fn prepare(
         Ok(metadata) => metadata,
         Err(reason) => return Prepared::Failed(reason),
     };
+    // Earlier epochs' metadata as the target's snapshot built it, whatever was written
+    // there since.
+    let built = match snapshot::as_built(copy, target) {
+        Ok(Some(built)) => built,
+        Ok(None) => return Prepared::Failed(format!("{target} follows no epoch")),
+        Err(reason) => return Prepared::Failed(reason),
+    };
 
     // The epoch that followed ours records (per actor) which of our changes it took.
     // Snapshots carry earlier epochs' metadata forward while a node can replay through
     // it (`snapshot::left_out`), so this works across any gap within the window; beyond
     // it, the metadata missing quarantines as the window would.
-    let next = snapshot::read_epoch_metadata(copy, previous.generation + 1)
+    let next = snapshot::read_epoch_metadata(&built, previous.generation + 1)
         .ok()
         .filter(|next| next.previous == previous);
     let gap = target.generation - previous.generation;
@@ -537,7 +544,7 @@ fn prepare(
     // this sees every one since our epoch, however far behind.
     let mut strict = crate::access::policy::of(old).strict();
     for generation in previous.generation + 1..=target.generation {
-        if let Ok(epoch) = snapshot::read_epoch_metadata(copy, generation)
+        if let Ok(epoch) = snapshot::read_epoch_metadata(&built, generation)
             && epoch.kind == "strict"
         {
             if strict {
@@ -552,7 +559,7 @@ fn prepare(
     let translator = (previous.generation + 1..=target.generation).try_fold(
         Translator::default(),
         |mut translator, generation| {
-            let epoch = snapshot::read_epoch_metadata(copy, generation)?;
+            let epoch = snapshot::read_epoch_metadata(&built, generation)?;
             let kind = epoch.kind().ok_or_else(|| {
                 format!(
                     "epoch generation {generation} was made by unknown remake kind {:?}",
diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs
index d83d49b..e82a2ab 100644
--- a/src/remake/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -80,6 +80,8 @@ pub(crate) enum SnapshotError {
     Task(String),
     #[error("unknown remake kind {0:?}")]
     UnknownKind(String),
+    #[error("earlier epochs' metadata: {0}")]
+    Earlier(String),
 }
 
 /// Owned version of [`SnapshotParams`], for moving into a blocking task.
@@ -209,6 +211,25 @@ pub(crate) fn read_epoch_metadata(
     })
 }
 
+/// `doc`, a document of `epoch`, at that epoch's snapshot change: earlier epochs'
+/// metadata as the snapshot built it, whatever was written there since (with no policy,
+/// screening judges only the root: a member's own change could). None for the first
+/// epoch, which follows none.
+pub(crate) fn as_built(doc: &Document, epoch: Epoch) -> Result<Option<Document>, String> {
+    if epoch.generation == 0 {
+        return Ok(None);
+    }
+    let root = ChangeHash(epoch.id);
+    if !crate::document::changes::holds(doc, &root) {
+        return Err(format!(
+            "document does not contain the snapshot change of {epoch}"
+        ));
+    }
+    doc.fork_at(&[root])
+        .map(Some)
+        .map_err(|error| error.to_string())
+}
+
 /// 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.
@@ -342,13 +363,17 @@ pub(crate) fn build(
     {
         store::copy_path(&base, &mut target, &meta::path(&[key.as_str()]))?;
     }
-    let earlier: BTreeMap<u64, Result<EpochMetadata, String>> =
-        store::keys_at(&base, &meta::path(&[EPOCHS]))?
+    // As the base's epoch built them, whatever was written there since.
+    let built = as_built(&base, params.previous).map_err(SnapshotError::Earlier)?;
+    let earlier: BTreeMap<u64, Result<EpochMetadata, String>> = match &built {
+        Some(built) => store::keys_at(built, &meta::path(&[EPOCHS]))?
             .unwrap_or_default()
             .iter()
             .filter_map(|key| key.parse::<u64>().ok())
-            .map(|g| (g, read_epoch_metadata(&base, g)))
-            .collect();
+            .map(|g| (g, read_epoch_metadata(built, g)))
+            .collect(),
+        None => BTreeMap::new(),
+    };
     for g in left_out(&earlier, generation, params.proposed_at) {
         store::delete_path(&mut target, &epoch_metadata_path(g), false)?;
     }
@@ -719,11 +744,25 @@ mod tests {
                 .unwrap_or_default()
         };
 
-        let heads = first.doc.get_heads();
+        // A change since the snapshot narrowing generation 1's window changes nothing:
+        // the metadata is read as the snapshot built it.
+        let mut edited = first
+            .doc
+            .fork()
+            .with_actor(ActorId::from(b"member".as_slice()));
+        change(&mut edited, |d| {
+            store::write_path(
+                d,
+                &[meta::path(&[EPOCHS, "1"]), vec!["policy".to_owned()]].concat(),
+                &json!({ "replayWindow": { "epochs": 0 } }),
+            )
+            .unwrap();
+        });
+        let heads = edited.get_heads();
         let mut second = params(&heads, &policy);
         second.remake_id = "c-2";
         second.previous = first.epoch;
-        let second = build(&first.doc, &second).unwrap();
+        let second = build(&edited, &second).unwrap();
         // A node on genesis is two behind generation 2, within the window: 1 stays.
         assert_eq!(generations(&second.doc), vec!["1", "2"]);