nodes keep list of every actor ID they've used

5 files changed, 54 insertions, 20 deletions

diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index cad5a22..4feb70e 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -31,7 +31,7 @@ use std::collections::{BTreeMap, HashMap};
 use std::convert::Infallible;
 use std::time::Duration;
 
-use automerge::{AutoCommit, ChangeHash};
+use automerge::{ActorId, AutoCommit, ChangeHash};
 use tokio::time::Instant;
 use tracing::{debug, error, info, warn};
 
@@ -177,8 +177,12 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
         host.set_stale_view(Some((local, copy)));
     }
     let was_ready = host.suspend().await;
+    let actor = ActorId::random();
+    host.remake().add_own_actor(actor.clone());
     let old = match host
-        .switch_epoch(target, |_live| Ok::<_, Infallible>(AutoCommit::new()))
+        .switch_epoch(target, move |_live| {
+            Ok::<_, Infallible>(AutoCommit::new().with_actor(actor))
+        })
         .await
     {
         Ok(old) => old,
@@ -276,7 +280,9 @@ async fn finish<H: Host>(host: &H) {
 
     // Everything heavy runs on a copy, without a lock.
     let copy = host.read_doc(|doc| doc.clone()).await;
-    let prepared = tokio::task::spawn_blocking(move || prepare(copy, old, previous, target)).await;
+    let own = host.remake().own_actors();
+    let prepared =
+        tokio::task::spawn_blocking(move || prepare(copy, old, previous, target, own)).await;
 
     let (old, metadata, plan, own_missing, had_history) = match prepared {
         Ok(Prepared::Ready {
@@ -395,7 +401,15 @@ async fn finish<H: Host>(host: &H) {
 }
 
 /// Verifies the synced document and decides about the old document's own changes.
-fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: Epoch) -> Prepared {
+/// `own`: this node's actors (`RemakeState::own_actors`); the old document's own actor
+/// counts in any case.
+fn prepare(
+    mut copy: AutoCommit,
+    mut old: AutoCommit,
+    previous: Epoch,
+    target: Epoch,
+    mut own: Vec<ActorId>,
+) -> Prepared {
     if copy.get_change_by_hash(&ChangeHash(target.id)).is_none() {
         return Prepared::NotYet(old);
     }
@@ -421,7 +435,7 @@ fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: E
         .ok()
         .filter(|next| next.previous == previous);
 
-    let own = old.get_actor().clone();
+    own.push(old.get_actor().clone());
     let clock = next
         .as_ref()
         .map_or_else(BTreeMap::new, |next| next.clock.clone());
@@ -573,7 +587,7 @@ mod tests {
     #[test]
     fn replays_within_the_window() {
         let (old, swarm, target) = scenario(&json!({}));
-        match prepare(swarm, old, Epoch::GENESIS, target) {
+        match prepare(swarm, old, Epoch::GENESIS, target, Vec::new()) {
             Prepared::Ready {
                 plan: Plan::Replay(paths, _),
                 own_missing,
@@ -589,7 +603,7 @@ mod tests {
     #[test]
     fn quarantines_outside_the_window() {
         let (old, swarm, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } }));
-        match prepare(swarm, old, Epoch::GENESIS, target) {
+        match prepare(swarm, old, Epoch::GENESIS, target, Vec::new()) {
             Prepared::Ready {
                 plan: Plan::Quarantine { paths, .. },
                 ..
@@ -602,7 +616,7 @@ mod tests {
     fn new_node_has_nothing_to_replay() {
         let (_, swarm, target) = scenario(&json!({}));
         assert!(matches!(
-            prepare(swarm, AutoCommit::new(), Epoch::GENESIS, target),
+            prepare(swarm, AutoCommit::new(), Epoch::GENESIS, target, Vec::new()),
             Prepared::Ready {
                 plan: Plan::Nothing,
                 had_history: false,
@@ -615,7 +629,7 @@ mod tests {
     fn waits_for_the_target_epoch() {
         let (old, _, target) = scenario(&json!({}));
         assert!(matches!(
-            prepare(AutoCommit::new(), old, Epoch::GENESIS, target),
+            prepare(AutoCommit::new(), old, Epoch::GENESIS, target, Vec::new()),
             Prepared::NotYet(_)
         ));
     }
diff --git a/src/remake/delta.rs b/src/remake/delta.rs
index c8f6b7c..51cc409 100644
--- a/src/remake/delta.rs
+++ b/src/remake/delta.rs
@@ -159,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();
@@ -370,7 +371,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);
@@ -381,7 +382,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();
diff --git a/src/remake/local.rs b/src/remake/local.rs
index 04ac0b0..09c99bd 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -9,7 +9,7 @@ use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, Mutex, MutexGuard};
 use std::time::Duration;
 
-use automerge::AutoCommit;
+use automerge::{ActorId, AutoCommit};
 use serde::Serialize;
 use tokio::time::Instant;
 
@@ -55,6 +55,9 @@ struct Inner {
     content_fetches: HashSet<[u8; 32]>,
     /// Remakes this node coordinates now (`coordinator`).
     coordinating: HashSet<String>,
+    /// Every actor this node writes or wrote with: one per run and epoch, and those of
+    /// earlier runs it knows of (`--document-file`). Its own writes, for a rejoin.
+    own_actors: Vec<ActorId>,
 }
 
 /// Content no remake refers to any more is kept this long after it was stored, for
@@ -147,6 +150,18 @@ impl RemakeState {
         self.started_at
     }
 
+    /// This node writes with `actor` from now on (or did, in an earlier run).
+    pub(crate) fn add_own_actor(&self, actor: ActorId) {
+        let mut inner = self.lock();
+        if !inner.own_actors.contains(&actor) {
+            inner.own_actors.push(actor);
+        }
+    }
+
+    pub(crate) fn own_actors(&self) -> Vec<ActorId> {
+        self.lock().own_actors.clone()
+    }
+
     /// Whether this node starts coordinating `id` now (false if it already does).
     pub(crate) fn start_coordinating(&self, id: &str) -> bool {
         self.lock().coordinating.insert(id.to_owned())
diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs
index 3c180ba..20569e0 100644
--- a/src/remake/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -750,7 +750,8 @@ mod tests {
         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);
+        let missing =
+            delta::missing_from_snapshot(&mut late, &metadata.clock, std::slice::from_ref(&own));
         assert_eq!(missing.own, 1);
         let heads = late.get_heads();
         let mut adopted = fetched.with_actor(ActorId::from(&b"late-node-next"[..]));
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index cf5ea75..e78643a 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -205,6 +205,9 @@ async fn switch<H: Host>(
         .take_prebuilt(id)
         .filter(|snapshot| snapshot.epoch == expected);
 
+    let actor = ActorId::random();
+    host.remake().add_own_actor(actor.clone());
+
     // 2. Build (or reuse), verify and replay without holding any lock.
     let (new_doc, copy_heads, mut stats) = tokio::task::spawn_blocking(
         move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> {
@@ -222,7 +225,7 @@ async fn switch<H: Host>(
             }
 
             // A fresh actor: this node's writes in the new epoch.
-            let mut doc = snapshot.doc.with_actor(ActorId::random());
+            let mut doc = snapshot.doc.with_actor(actor);
 
             // Keep this remake's record visible in the new epoch, for its history.
             if carry_record {