add clock module

The Clock module is the system clock that every unix timestamp new goes
through.  It comes with an offset that only tests can set.

1 files changed, 25 insertions, 9 deletions

diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 631fb25..405b9f9 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -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;
 
@@ -281,8 +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 own = host.remake().own_actors();
+    let now = host.now_millis();
     let prepared =
-        tokio::task::spawn_blocking(move || prepare(copy, old, previous, target, own)).await;
+        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 {
@@ -368,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(),
@@ -404,13 +405,14 @@ async fn finish<H: Host>(host: &H) {
 
 /// Verifies the synced document and decides about the old document's own changes.
 /// `own`: this node's actors (`RemakeState::own_actors`); the old document's own actor
-/// counts in any case.
+/// 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);
@@ -484,7 +486,7 @@ fn prepare(
             ),
         },
         (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 {
@@ -589,7 +591,7 @@ mod tests {
     #[test]
     fn replays_within_the_window() {
         let (old, swarm, target) = scenario(&json!({}));
-        match prepare(swarm, old, Epoch::GENESIS, target, Vec::new()) {
+        match prepare(swarm, old, Epoch::GENESIS, target, Vec::new(), 0) {
             Prepared::Ready {
                 plan: Plan::Replay(paths, _),
                 own_missing,
@@ -605,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, Vec::new()) {
+        match prepare(swarm, old, Epoch::GENESIS, target, Vec::new(), 0) {
             Prepared::Ready {
                 plan: Plan::Quarantine { paths, .. },
                 ..
@@ -618,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, Vec::new()),
+            prepare(
+                swarm,
+                AutoCommit::new(),
+                Epoch::GENESIS,
+                target,
+                Vec::new(),
+                0
+            ),
             Prepared::Ready {
                 plan: Plan::Nothing,
                 had_history: false,
@@ -631,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, Vec::new()),
+            prepare(
+                AutoCommit::new(),
+                old,
+                Epoch::GENESIS,
+                target,
+                Vec::new(),
+                0
+            ),
             Prepared::NotYet(_)
         ));
     }