1 files changed, 118 insertions, 4 deletions

diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs
index dc61810..d83d49b 100644
--- a/src/remake/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -175,8 +175,9 @@ struct MetadataRecord {
 }
 
 /// Metadata of an earlier generation, as the current document holds it. Snapshots
-/// carry the metadata of all earlier epochs forward (only remake records stay
-/// behind), so a node many epochs behind can still find the epoch that followed its own.
+/// carry earlier epochs' metadata forward as long as a rejoining node can replay
+/// through it, and a strict remake's for good (`left_out`), so a node many epochs
+/// behind can still find the epoch that followed its own.
 pub(crate) fn read_epoch_metadata(
     doc: &Document,
     generation: u64,
@@ -267,6 +268,38 @@ pub(crate) fn epoch_metadata_path(generation: u64) -> Vec<String> {
     meta::path(&[EPOCHS, generation.to_string().as_str()])
 }
 
+/// Of `earlier` epochs' metadata (by generation), the generations a snapshot of
+/// `generation`, proposed at `proposed_at`, leaves out: those no rejoining node replays
+/// through any more. A node on epoch P replays while the replay window of epoch P+1
+/// covers it, by the epochs and the time since (`catchup`), and then needs every
+/// generation after P. So everything from the oldest generation whose window still
+/// covers a node on the epoch before it is kept, or that doesn't read; older ones go,
+/// but for those a strict remake made: by them a rejoining node, however far behind,
+/// tells a strict remake of a swarm strict already (`catchup::prepare`). Judged by the
+/// remake's `proposed_at`, the same on every node, and earlier than any rejoining
+/// node's clock, so nothing a rejoining node can still replay through goes.
+fn left_out(
+    earlier: &BTreeMap<u64, Result<EpochMetadata, String>>,
+    generation: u64,
+    proposed_at: UnixMillis,
+) -> Vec<u64> {
+    let kept_from = earlier
+        .iter()
+        .find(|(g, metadata)| match metadata {
+            Ok(metadata) => metadata.policy.replay_window.covers(
+                generation.saturating_sub(g.saturating_sub(1)),
+                proposed_at.ms_since(metadata.proposed_at),
+            ),
+            Err(_) => true,
+        })
+        .map_or(generation, |(g, _)| *g);
+    earlier
+        .range(..kept_from)
+        .filter(|(_, metadata)| metadata.as_ref().is_ok_and(|m| m.kind != "strict"))
+        .map(|(g, _)| *g)
+        .collect()
+}
+
 /// Builds the snapshot of `source` at `params.base_heads`.
 pub(crate) fn build(
     source: &Document,
@@ -294,8 +327,9 @@ pub(crate) fn build(
     kind.build(&base, params.content, &mut target)?;
 
     // Protocol data carries forward, whatever the kind: forget records, earlier epochs'
-    // metadata (rejoining nodes need it). Remake records and the votes register describe
-    // the old epoch's protocol and stay behind.
+    // metadata (rejoining nodes need it, as long as they can replay, `left_out`).
+    // Remake records and the votes register describe the old epoch's protocol and stay
+    // behind.
     match store::delete_path(&mut target, &meta::path(&[]), false) {
         Ok(()) | Err(StoreError::NotFound) => {}
         Err(error) => return Err(error.into()),
@@ -308,6 +342,16 @@ 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]))?
+            .unwrap_or_default()
+            .iter()
+            .filter_map(|key| key.parse::<u64>().ok())
+            .map(|g| (g, read_epoch_metadata(&base, g)))
+            .collect();
+    for g in left_out(&earlier, generation, params.proposed_at) {
+        store::delete_path(&mut target, &epoch_metadata_path(g), false)?;
+    }
     // What the kind does to it (a strict remake keeps only what its manifest names).
     kind.protocol(&base, params.content, &mut target)?;
 
@@ -622,6 +666,76 @@ mod tests {
         assert_eq!(epoch["baseHeads"].as_array().unwrap().len(), heads.len());
     }
 
+    /// Earlier epochs' metadata goes once no rejoining node replays through it, oldest
+    /// first, but for a strict remake's; what doesn't read stays.
+    #[test]
+    fn earlier_metadata_left_out_past_the_replay_window() {
+        let entry = |proposed_at: u64, kind: &str, window: JsonValue| {
+            Ok(EpochMetadata {
+                remake_id: String::new(),
+                previous: Epoch::GENESIS,
+                clock: BTreeMap::new(),
+                policy: serde_json::from_value(json!({ "replayWindow": window })).unwrap(),
+                proposed_at: UnixMillis(proposed_at),
+                kind: kind.to_owned(),
+                translation: JsonValue::Null,
+            })
+        };
+        let one = || json!({ "epochs": 1 });
+        // Building generation 5: a node on 3 (window of 4) is 2 behind, past one epoch.
+        let earlier = BTreeMap::from([
+            (1, entry(0, "compact", one())),
+            (2, entry(0, "strict", one())),
+            (3, entry(0, "compact", one())),
+            (4, entry(0, "compact", one())),
+        ]);
+        assert_eq!(left_out(&earlier, 5, UnixMillis(0)), vec![1, 3, 4]);
+        // Generation 4's window covers by its age: everything from 4 stays.
+        let mut by_age = earlier.clone();
+        by_age.insert(4, entry(1_000, "compact", json!({ "ageMs": 500 })));
+        assert_eq!(left_out(&by_age, 5, UnixMillis(1_400)), vec![1, 3]);
+        assert_eq!(left_out(&by_age, 5, UnixMillis(1_600)), vec![1, 3, 4]);
+        // An older generation still covering keeps everything after it.
+        let mut wide = earlier.clone();
+        wide.insert(1, entry(0, "compact", json!({ "epochs": 9 })));
+        assert!(left_out(&wide, 5, UnixMillis(0)).is_empty());
+        // Metadata that doesn't read stays, and so does everything after it.
+        let mut unread = earlier;
+        unread.insert(3, Err("invalid".to_owned()));
+        assert_eq!(left_out(&unread, 5, UnixMillis(0)), vec![1]);
+    }
+
+    /// A snapshot of a snapshot leaves out the generation no node replays through.
+    #[test]
+    fn snapshot_of_snapshot_leaves_out_past_metadata() {
+        let mut a = doc(b"actor-a");
+        change(&mut a, |a| a.put(ROOT, "x", 1_i64).unwrap());
+        let policy = json!({ "replayWindow": { "epochs": 2 } });
+        let heads = a.get_heads();
+        let first = build(&a, &params(&heads, &policy)).unwrap();
+        let generations = |doc: &Document| {
+            store::keys_at(doc, &meta::path(&[EPOCHS]))
+                .unwrap()
+                .unwrap_or_default()
+        };
+
+        let heads = first.doc.get_heads();
+        let mut second = params(&heads, &policy);
+        second.remake_id = "c-2";
+        second.previous = first.epoch;
+        let second = build(&first.doc, &second).unwrap();
+        // A node on genesis is two behind generation 2, within the window: 1 stays.
+        assert_eq!(generations(&second.doc), vec!["1", "2"]);
+
+        let heads = second.doc.get_heads();
+        let mut third = params(&heads, &policy);
+        third.remake_id = "c-3";
+        third.previous = second.epoch;
+        let third = build(&second.doc, &third).unwrap();
+        // Now three behind: 1 goes; a node on 1 replays through 2 and 3.
+        assert_eq!(generations(&third.doc), vec!["2", "3"]);
+    }
+
     #[test]
     fn automerge_types_survive() {
         let mut a = doc(b"actor-a");