1 files changed, 44 insertions, 12 deletions

diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index cad5a22..405b9f9 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};
 
@@ -40,7 +40,6 @@ use super::delta::{self, ReplayStats};
 use super::kind::Translator;
 use super::local::{CatchUp, LocalStatus, Rejoin};
 use super::quarantine;
-use super::records::unix_millis;
 use super::snapshot::{self, EpochMetadata};
 use crate::document::replica::Epoch;
 
@@ -177,8 +176,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 +279,10 @@ 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 now = host.now_millis();
+    let prepared =
+        tokio::task::spawn_blocking(move || prepare(copy, old, previous, target, own, now)).await;
 
     let (old, metadata, plan, own_missing, had_history) = match prepared {
         Ok(Prepared::Ready {
@@ -362,13 +368,14 @@ async fn finish<H: Host>(host: &H) {
             document,
             metadata.policy.archive_retention.clone(),
             target.generation,
+            host.now_millis(),
         );
     }
 
     host.remake().set_status(
         &metadata.remake_id,
         LocalStatus {
-            switched_at: Some(unix_millis()),
+            switched_at: Some(host.now_millis()),
             replayed_paths: Some(stats.replayed),
             failed_paths: Some(stats.failed),
             quarantine: quarantine_id.clone(),
@@ -376,6 +383,8 @@ async fn finish<H: Host>(host: &H) {
         },
     );
     host.remake().set_catch_up(None);
+    let actor = host.read_doc(|doc| doc.get_actor().clone()).await;
+    host.remake().reset_own_actors(actor);
     if was_ready {
         host.resume().await;
     }
@@ -395,7 +404,16 @@ 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. `now`: Unix milliseconds, by the node's clock.
+fn prepare(
+    mut copy: AutoCommit,
+    mut old: AutoCommit,
+    previous: Epoch,
+    target: Epoch,
+    mut own: Vec<ActorId>,
+    now: u64,
+) -> Prepared {
     if copy.get_change_by_hash(&ChangeHash(target.id)).is_none() {
         return Prepared::NotYet(old);
     }
@@ -421,7 +439,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());
@@ -468,7 +486,7 @@ fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: E
             ),
         },
         (Some(next), Ok(translator)) => {
-            let away_ms = unix_millis().saturating_sub(next.proposed_at);
+            let away_ms = now.saturating_sub(next.proposed_at);
             if next.policy.replay_window.covers(gap, away_ms) {
                 Plan::Replay(paths, translator)
             } else {
@@ -573,7 +591,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(), 0) {
             Prepared::Ready {
                 plan: Plan::Replay(paths, _),
                 own_missing,
@@ -589,7 +607,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(), 0) {
             Prepared::Ready {
                 plan: Plan::Quarantine { paths, .. },
                 ..
@@ -602,7 +620,14 @@ 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(),
+                0
+            ),
             Prepared::Ready {
                 plan: Plan::Nothing,
                 had_history: false,
@@ -615,7 +640,14 @@ 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(),
+                0
+            ),
             Prepared::NotYet(_)
         ));
     }