1 files changed, 18 insertions, 12 deletions

diff --git a/src/remake/local.rs b/src/remake/local.rs
index 65a2735..517367b 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -13,7 +13,7 @@ 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(),
         }
     }
@@ -215,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> {
@@ -369,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),
@@ -385,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],
@@ -400,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.
@@ -414,6 +418,7 @@ mod tests {
                 seconds: Some(3600),
             },
             5,
+            now,
         );
         assert_eq!(remake.archived_epochs().len(), 1);
 
@@ -426,6 +431,7 @@ mod tests {
                 seconds: None,
             },
             3,
+            now,
         );
         assert_eq!(remake.archived_epochs().len(), 1);
     }