1 files changed, 121 insertions, 1 deletions

diff --git a/src/compaction/delta.rs b/src/compaction/delta.rs
index 3bc9456..533a7be 100644
--- a/src/compaction/delta.rs
+++ b/src/compaction/delta.rs
@@ -17,7 +17,9 @@
 //! The protocol's own metadata (`_continuum/compactions`, `_continuum/epochs`) is never
 //! replayed: the snapshot owns it.
 
-use automerge::{AutoCommit, ChangeHash, PatchAction, Prop};
+use std::collections::{BTreeMap, HashSet};
+
+use automerge::{ActorId, AutoCommit, ChangeHash, PatchAction, Prop};
 use tracing::warn;
 
 use super::snapshot::{COMPACTIONS, EPOCHS, META_ROOT};
@@ -92,6 +94,81 @@ fn contains_protocol_path(path: &[String]) -> bool {
     matches!(path, [] | [_]) && path.first().is_none_or(|root| root == META_ROOT)
 }
 
+/// What a late node has that a snapshot lacks.
+#[derive(Debug)]
+pub(crate) struct Missing {
+    /// Heads of everything except `own` changes the snapshot lacks (and changes built on
+    /// them). Replaying from these heads to the document's heads replays exactly those.
+    pub(crate) before: Vec<ChangeHash>,
+    /// Changes by `own` that the snapshot lacks.
+    pub(crate) own: usize,
+}
+
+/// Splits `doc`'s history against a snapshot `clock`: a change (actor, seq) is in the
+/// snapshot iff seq <= clock[actor].
+///
+/// Only changes this node wrote itself (actor `own`) are worth replaying: other nodes'
+/// changes after the base were replayed by those nodes when they switched, and replaying
+/// them again would overwrite newer writes in the new epoch with stale ones.
+pub(crate) fn missing_from_snapshot(
+    doc: &mut AutoCommit,
+    clock: &BTreeMap<String, u64>,
+    own: &ActorId,
+) -> Missing {
+    let own_hex = own.to_string();
+    let changes: Vec<(ChangeHash, Vec<ChangeHash>, bool)> = doc
+        .get_changes(&[])
+        .iter()
+        .map(|change| {
+            let actor = change.actor_id().to_string();
+            let in_snapshot = clock.get(&actor).is_some_and(|&seq| change.seq() <= seq);
+            let own_missing = actor == own_hex && !in_snapshot;
+            (change.hash(), change.deps().to_vec(), own_missing)
+        })
+        .collect();
+
+    // Excluded: own missing changes and everything that (transitively) depends on them.
+    // A fixpoint, so it doesn't rely on the order `get_changes` returns.
+    let mut excluded: HashSet<ChangeHash> = changes
+        .iter()
+        .filter(|(_, _, own_missing)| *own_missing)
+        .map(|(hash, _, _)| *hash)
+        .collect();
+    let own_count = excluded.len();
+    loop {
+        let before = excluded.len();
+        for (hash, deps, _) in &changes {
+            if !excluded.contains(hash) && deps.iter().any(|dep| excluded.contains(dep)) {
+                excluded.insert(*hash);
+            }
+        }
+        if excluded.len() == before {
+            break;
+        }
+    }
+
+    // Heads of the kept part: kept changes no kept change depends on.
+    let kept: Vec<&(ChangeHash, Vec<ChangeHash>, bool)> = changes
+        .iter()
+        .filter(|(hash, _, _)| !excluded.contains(hash))
+        .collect();
+    let depended_on: HashSet<ChangeHash> = kept
+        .iter()
+        .flat_map(|(_, deps, _)| deps.iter().copied())
+        .collect();
+    let mut before: Vec<ChangeHash> = kept
+        .iter()
+        .map(|(hash, _, _)| *hash)
+        .filter(|hash| !depended_on.contains(hash))
+        .collect();
+    before.sort();
+
+    Missing {
+        before,
+        own: own_count,
+    }
+}
+
 /// The minimal set of store paths covering every change between `before` and `after`:
 /// sorted, and no path is a prefix of another.
 pub(crate) fn changed_paths(
@@ -217,6 +294,49 @@ mod tests {
     }
 
     #[test]
+    fn missing_counts_only_own_changes() {
+        let own = ActorId::from(&b"own-actor"[..]);
+        let other = ActorId::from(&b"other-actor"[..]);
+
+        let mut doc = AutoCommit::new().with_actor(own.clone());
+        doc.put(ROOT, "base", 1_i64).unwrap();
+        doc.commit();
+        let base = doc.get_heads();
+
+        // The snapshot contains the base only.
+        let clock: BTreeMap<String, u64> = [(own.to_string(), 1)].into_iter().collect();
+
+        // A foreign change after the base (replayed by its own node) ...
+        let mut foreign = doc.fork().with_actor(other);
+        foreign.put(ROOT, "theirs", 1_i64).unwrap();
+        foreign.commit();
+        doc.merge(&mut foreign).unwrap();
+        let with_foreign = doc.get_heads();
+
+        // ... and one of ours.
+        doc.put(ROOT, "mine", 1_i64).unwrap();
+        doc.commit();
+        let after = doc.get_heads();
+
+        let missing = missing_from_snapshot(&mut doc, &clock, &own);
+        assert_eq!(missing.own, 1);
+        assert_eq!(missing.before, with_foreign);
+        assert_ne!(missing.before, base);
+        assert_eq!(
+            changed_paths(&mut doc, &missing.before, &after),
+            vec![vec!["mine".to_owned()]]
+        );
+
+        // Nothing of ours missing: nothing to replay.
+        let full: BTreeMap<String, u64> = [(own.to_string(), 2)].into_iter().collect();
+        let none = missing_from_snapshot(&mut doc, &full, &own);
+        assert_eq!(none.own, 0);
+        let mut heads = after.clone();
+        heads.sort();
+        assert_eq!(none.before, heads);
+    }
+
+    #[test]
     fn protocol_metadata_is_never_replayed() {
         let mut old = AutoCommit::new();
         old.put(ROOT, "x", 1_i64).unwrap();