1 files changed, 79 insertions, 10 deletions

diff --git a/src/remake/delta.rs b/src/remake/delta.rs
index d318c00..59acc8c 100644
--- a/src/remake/delta.rs
+++ b/src/remake/delta.rs
@@ -27,7 +27,7 @@ use automerge::{ActorId, AutoCommit, ChangeHash, PatchAction, Prop};
 use tracing::warn;
 
 use super::kind::{Translation, Translator};
-use super::snapshot::{EPOCHS, META_ROOT, REMAKES};
+use super::snapshot::{EPOCHS, META_ROOT, REMAKES, VOTES};
 use crate::document::store::{self, StoreError};
 
 #[derive(Debug, Default, Clone)]
@@ -135,7 +135,10 @@ fn log_failure(path: &[String], error: &StoreError) {
 }
 
 fn is_protocol_path(path: &[String]) -> bool {
-    matches!(path, [root, part, ..] if root == META_ROOT && (part == REMAKES || part == EPOCHS))
+    let [root, part, ..] = path else {
+        return false;
+    };
+    root == META_ROOT && [REMAKES, VOTES, EPOCHS].contains(&part.as_str())
 }
 
 /// True for the root and for `_continuum` itself.
@@ -156,22 +159,23 @@ pub(crate) struct Missing {
 /// 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.
+/// Only changes this node wrote itself (any of the actors in `own`: one per run and
+/// epoch) are worth replaying when it rejoins: other nodes' changes after the base were
+/// replayed by the nodes that 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,
+    own: &[ActorId],
 ) -> Missing {
-    let own_hex = own.to_string();
+    let own_hex: HashSet<String> = own.iter().map(ActorId::to_string).collect();
     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;
+            let own_missing = own_hex.contains(&actor) && !in_snapshot;
             (change.hash(), change.deps().to_vec(), own_missing)
         })
         .collect();
@@ -275,6 +279,71 @@ mod tests {
         store::read_path(doc, &[]).unwrap().unwrap()
     }
 
+    /// A random store write as the API makes it (one change): a scalar or a small map
+    /// at a path up to three keys deep, or a delete. Writes the store refuses (e.g. into
+    /// a scalar) are rolled back.
+    fn random_write(rng: &mut crate::testrand::Rng, doc: &mut AutoCommit) {
+        const KEYS: [&str; 3] = ["a", "b", "c"];
+        let depth = rng.between(1, 3);
+        let path: Vec<String> = (0..depth).map(|_| rng.pick(&KEYS).to_string()).collect();
+        let result = if rng.chance(0.2) {
+            store::delete_path(doc, &path, false)
+        } else {
+            let value = match rng.below(4) {
+                0 => json!(rng.below(100)),
+                1 => json!(format!("s{}", rng.below(100))),
+                2 => json!(rng.chance(0.5)),
+                _ => {
+                    let mut map = serde_json::Map::new();
+                    map.insert(rng.pick(&KEYS).to_string(), json!(rng.below(100)));
+                    serde_json::Value::Object(map)
+                }
+            };
+            store::write_path(doc, &path, &value)
+        };
+        match result {
+            Ok(()) => {
+                doc.commit_with(automerge::transaction::CommitOptions::default().with_time(0));
+            }
+            Err(_) => {
+                doc.rollback();
+            }
+        }
+    }
+
+    /// Compaction's promise: a snapshot of the base plus everything replayed since gives
+    /// the old document's content, whatever several nodes wrote concurrently.
+    #[test]
+    fn replay_reproduces_the_document() {
+        crate::testrand::for_seeds(200, |rng| {
+            let actor = |i: u8| ActorId::from(&[i; 16][..]);
+            let mut old = AutoCommit::new().with_actor(actor(0));
+            for _ in 0..rng.between(0, 8) {
+                random_write(rng, &mut old);
+            }
+            let base = old.get_heads();
+            let mut new = AutoCommit::new().with_actor(actor(100));
+            store::materialize_into(&old, &mut new, &|_| false).unwrap();
+            new.commit();
+
+            let mut nodes: Vec<AutoCommit> = (1..=rng.between(1, 3))
+                .map(|i| old.fork().with_actor(actor(i as u8)))
+                .collect();
+            for _ in 0..rng.between(1, 20) {
+                let node = rng.below(nodes.len());
+                random_write(rng, &mut nodes[node]);
+            }
+            for node in &mut nodes {
+                old.merge(node).unwrap();
+            }
+            let after = old.get_heads();
+
+            let stats = replay(&mut old, &base, &after, &mut new, &Translator::default());
+            assert_eq!(stats.failed, 0);
+            assert_eq!(view(&new), view(&old));
+        });
+    }
+
     #[test]
     fn minimal_paths() {
         let mut doc = AutoCommit::new();
@@ -367,7 +436,7 @@ mod tests {
         doc.commit();
         let after = doc.get_heads();
 
-        let missing = missing_from_snapshot(&mut doc, &clock, &own);
+        let missing = missing_from_snapshot(&mut doc, &clock, std::slice::from_ref(&own));
         assert_eq!(missing.own, 1);
         assert_eq!(missing.before, with_foreign);
         assert_ne!(missing.before, base);
@@ -378,7 +447,7 @@ mod tests {
 
         // 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);
+        let none = missing_from_snapshot(&mut doc, &full, std::slice::from_ref(&own));
         assert_eq!(none.own, 0);
         let mut heads = after.clone();
         heads.sort();