1 files changed, 42 insertions, 13 deletions

diff --git a/src/remake/local.rs b/src/remake/local.rs
index 04ac0b0..517367b 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -9,11 +9,11 @@ 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;
 
-use super::records::{Retention, unix_millis};
+use super::records::Retention;
 use super::snapshot::Snapshot;
 use crate::document::replica::Epoch;
 
@@ -25,10 +25,11 @@ pub(crate) struct RemakeState {
     inner: Mutex<Inner>,
 }
 
-impl Default for RemakeState {
-    fn default() -> Self {
+impl RemakeState {
+    /// `started_at`: now, by the node's clock.
+    pub(crate) fn new(started_at: u64) -> Self {
         Self {
-            started_at: unix_millis(),
+            started_at,
             inner: Mutex::default(),
         }
     }
@@ -55,6 +56,11 @@ struct Inner {
     content_fetches: HashSet<[u8; 32]>,
     /// Remakes this node coordinates now (`coordinator`).
     coordinating: HashSet<String>,
+    /// The actors this node writes or wrote with in the current document: this run's,
+    /// and those of earlier runs it knows of (`--document-file`). Its own writes, for a
+    /// rejoin. A switch or rejoin starts the list over with the new document's actor, so
+    /// it only grows with restarts within one epoch.
+    own_actors: Vec<ActorId>,
 }
 
 /// Content no remake refers to any more is kept this long after it was stored, for
@@ -147,6 +153,24 @@ 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);
+        }
+    }
+
+    /// The current document is a new one, written with `actor` only (a switch or rejoin
+    /// is done).
+    pub(crate) fn reset_own_actors(&self, actor: ActorId) {
+        self.lock().own_actors = vec![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())
@@ -192,25 +216,27 @@ impl RemakeState {
     }
 
     /// Keeps the previous epoch's document, then drops what has outlived its retention.
+    /// `now`: Unix milliseconds, by the node's clock.
     pub(crate) fn archive(
         &self,
         epoch: Epoch,
         doc: AutoCommit,
         retention: Retention,
         current_generation: u64,
+        now: u64,
     ) {
         let mut inner = self.lock();
         inner.archive.push(Archived {
             epoch,
             doc,
-            archived_at: unix_millis(),
+            archived_at: now,
             retention,
         });
-        prune(&mut inner.archive, current_generation);
+        prune(&mut inner.archive, current_generation, now);
     }
 
-    pub(crate) fn prune_archive(&self, current_generation: u64) {
-        prune(&mut self.lock().archive, current_generation);
+    pub(crate) fn prune_archive(&self, current_generation: u64, now: u64) {
+        prune(&mut self.lock().archive, current_generation, now);
     }
 
     pub(crate) fn catch_up(&self) -> Option<CatchUp> {
@@ -346,8 +372,7 @@ impl RemakeState {
 }
 
 /// An archive is kept until every limit of its retention has passed.
-fn prune(archive: &mut Vec<Archived>, current_generation: u64) {
-    let now = unix_millis();
+fn prune(archive: &mut Vec<Archived>, current_generation: u64, now: u64) {
     archive.retain(|archived| {
         archived.retention.covers(
             current_generation.saturating_sub(archived.epoch.generation),
@@ -362,7 +387,8 @@ mod tests {
 
     #[test]
     fn archive_retention() {
-        let remake = RemakeState::default();
+        let remake = RemakeState::new(0);
+        let now = 1_000_000;
         let epoch = |generation| Epoch {
             generation,
             id: [generation as u8; 32],
@@ -377,9 +403,10 @@ mod tests {
                 seconds: None,
             },
             1,
+            now,
         );
         assert_eq!(remake.archived_epochs().len(), 1);
-        remake.prune_archive(2);
+        remake.prune_archive(2, now);
         assert!(remake.archived_epochs().is_empty());
 
         // Kept by time while the epoch count has passed.
@@ -391,6 +418,7 @@ mod tests {
                 seconds: Some(3600),
             },
             5,
+            now,
         );
         assert_eq!(remake.archived_epochs().len(), 1);
 
@@ -403,6 +431,7 @@ mod tests {
                 seconds: None,
             },
             3,
+            now,
         );
         assert_eq!(remake.archived_epochs().len(), 1);
     }