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.

9 files changed, 83 insertions, 58 deletions

diff --git a/src/remake/api.rs b/src/remake/api.rs
index 0428a43..24fb7e8 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -69,7 +69,7 @@ use super::coordinator;
 use super::local::LocalStatus;
 use super::records::{
     self, AbortRequest, ContentRef, Decision, EpochRecord, Phase, Policy, Proposal, Record,
-    Retention, Vote, unix_millis,
+    Retention, Vote,
 };
 use super::{Host, kind, vote};
 use crate::document::store::StoreError;
@@ -299,7 +299,7 @@ fn grace<H: Host>(host: &H) -> u64 {
 
 fn report<H: Host>(host: &H, id: String, record: Record) -> Report {
     Report {
-        phase: record.phase(unix_millis(), grace(host)),
+        phase: record.phase(host.now_millis(), grace(host)),
         local: host.remake().status(&id),
         id,
         record,
@@ -502,7 +502,7 @@ async fn start_kind<H: Host>(
         _ => {}
     }
 
-    let now = unix_millis();
+    let now = host.now_millis();
     let settle_ms = settle.as_millis() as u64;
     let id = Uuid::new_v4().to_string();
     let grace = grace(&host);
@@ -887,7 +887,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, query: AbortQuery) -> Respons
     if current.record.abort_requested.is_none() {
         let request = AbortRequest {
             by: host.node_id(),
-            at: unix_millis(),
+            at: host.now_millis(),
         };
         let record_id = id.clone();
         if let Err(error) = host
@@ -926,7 +926,7 @@ async fn abort_expired<H: Host>(host: &H, current: Report, force: &Force) -> Res
         &host.node_id(),
         &active_peers(host).await,
         force,
-        unix_millis(),
+        host.now_millis(),
         grace,
     ) {
         return error_response(
@@ -946,11 +946,12 @@ async fn abort_expired<H: Host>(host: &H, current: Report, force: &Force) -> Res
             "aborted by operator via {} after it expired",
             host.node_id()
         ),
-        decided_at: unix_millis(),
+        decided_at: host.now_millis(),
     };
+    let clock = host.clock();
     let written = host
         .commit_doc(move |doc| -> Result<bool, StoreError> {
-            let now = unix_millis();
+            let now = clock.now_millis();
             let expired = records::load(doc, &record_id).is_some_and(|record| {
                 record.phase(now, grace) == Phase::Expired && now >= abortable_at(&record, grace)
             });
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(_)
         ));
     }
diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs
index 427de08..20de712 100644
--- a/src/remake/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -106,7 +106,7 @@ async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
 
     let tally = vote::tally(id, &record, &all, &register);
     // The deadline first: a majority counted after it is too late.
-    if records::unix_millis() >= record.proposal.settle_until {
+    if host.now_millis() >= record.proposal.settle_until {
         decide_abort(
             host,
             id,
@@ -181,7 +181,7 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor
         to_epoch: to_epoch.into(),
         base_heads,
         snapshot_bytes,
-        decided_at: records::unix_millis(),
+        decided_at: host.now_millis(),
     };
     match decide(host, id, decision).await? {
         Some(Decision::Committed { .. }) => {
@@ -200,7 +200,7 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor
 async fn decide_abort<H: Host>(host: &H, id: &str, reason: String) -> Result<(), StoreError> {
     let decision = Decision::Aborted {
         reason: reason.clone(),
-        decided_at: records::unix_millis(),
+        decided_at: host.now_millis(),
     };
     if decide(host, id, decision).await?.is_some() {
         warn!(remake = %id, reason = %reason, "Remake aborted");
@@ -220,6 +220,7 @@ async fn decide<H: Host>(
 ) -> Result<Option<Decision>, StoreError> {
     let id = id.to_owned();
     let grace = host.dormant_after().as_millis() as u64;
+    let clock = host.clock();
     host.commit_doc(move |doc| -> Result<Option<Decision>, StoreError> {
         let all = records::load_all(doc);
         let Some((_, record)) = all.iter().find(|(other, _)| *other == id) else {
@@ -232,7 +233,7 @@ async fn decide<H: Host>(
             Decision::Committed { decided_at, .. } => {
                 let register = records::load_register(doc);
                 let tally = vote::tally(&id, record, &all, &register);
-                let expired = record.phase(records::unix_millis(), grace) == Phase::Expired;
+                let expired = record.phase(clock.now_millis(), grace) == Phase::Expired;
                 match obstacle(&id, record, &all) {
                     Some(reason) => Decision::Aborted { reason, decided_at },
                     // Others may abort it by now (`api`): too late to commit.
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);
     }
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index 9085475..3980d38 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -58,6 +58,15 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
     /// This node's in-memory remake state.
     fn remake(&self) -> &RemakeState;
 
+    /// This node's wall clock (`clock`): deadlines, votes and decisions are stamped and
+    /// judged with it. Shared, so closures run under the document lock can read it.
+    fn clock(&self) -> Arc<crate::clock::Clock>;
+
+    /// Now, in Unix milliseconds, by [`clock`](Self::clock).
+    fn now_millis(&self) -> u64 {
+        self.clock().now_millis()
+    }
+
     /// This node's ID (its peers know it by this).
     fn node_id(&self) -> String;
 
diff --git a/src/remake/quarantine.rs b/src/remake/quarantine.rs
index 1090e2c..e7ef136 100644
--- a/src/remake/quarantine.rs
+++ b/src/remake/quarantine.rs
@@ -31,7 +31,6 @@ use super::api::error_response;
 use super::delta;
 use super::kind::Translator;
 use super::local::Quarantined;
-use super::records::unix_millis;
 use crate::document::replica::Epoch;
 use crate::document::store;
 
@@ -90,7 +89,7 @@ pub(super) fn add<H: Host>(
         id: id.clone(),
         epoch,
         target,
-        created_at: unix_millis(),
+        created_at: host.now_millis(),
         reason,
         paths,
         document: document.clone(),
diff --git a/src/remake/records.rs b/src/remake/records.rs
index 283f612..82c61a1 100644
--- a/src/remake/records.rs
+++ b/src/remake/records.rs
@@ -41,13 +41,6 @@ use crate::document::store::{self, StoreError};
 /// One day, for retention defaults.
 const DAY_SECS: u64 = 24 * 60 * 60;
 
-pub(crate) fn unix_millis() -> u64 {
-    std::time::SystemTime::now()
-        .duration_since(std::time::UNIX_EPOCH)
-        .map(|d| d.as_millis() as u64)
-        .unwrap_or(0)
-}
-
 // --- record types ------------------------------------------------------------
 
 /// An epoch in a record: generation plus hex-encoded snapshot hash.
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index 1a4dbf9..9cc1ca0 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -32,7 +32,7 @@ use super::delta::{self, ReplayStats};
 use super::kind::Translator;
 use super::local::LocalStatus;
 use super::quarantine;
-use super::records::{self, EpochRecord, Record, unix_millis};
+use super::records::{self, EpochRecord, Record};
 use super::snapshot::{self, META_ROOT, REMAKES};
 use super::vote;
 use crate::document::replica::Epoch;
@@ -73,7 +73,8 @@ pub(crate) async fn run_worker<H: Host>(host: H) {
 
 async fn check<H: Host>(host: &H) {
     let local = host.current_epoch();
-    host.remake().prune_archive(local.generation);
+    host.remake()
+        .prune_archive(local.generation, host.now_millis());
 
     // While rejoining, the document is being filled from peers: its records are not
     // complete enough to act on.
@@ -87,7 +88,7 @@ async fn check<H: Host>(host: &H) {
     content::prune(
         host,
         &all,
-        unix_millis(),
+        host.now_millis(),
         host.dormant_after().as_millis() as u64,
     );
     let local_record = EpochRecord::from(local);
@@ -144,7 +145,7 @@ async fn check<H: Host>(host: &H) {
             host.remake().set_status(
                 &id,
                 LocalStatus {
-                    switched_at: Some(unix_millis()),
+                    switched_at: Some(host.now_millis()),
                     replayed_paths: Some(stats.replayed),
                     failed_paths: Some(stats.failed),
                     quarantine,
@@ -257,8 +258,13 @@ async fn switch<H: Host>(
         &old,
         std::mem::take(&mut stats.quarantined),
     );
-    host.remake()
-        .archive(previous, old, retention, expected.generation);
+    host.remake().archive(
+        previous,
+        old,
+        retention,
+        expected.generation,
+        host.now_millis(),
+    );
 
     info!(
         remake = %id,
diff --git a/src/remake/vote.rs b/src/remake/vote.rs
index b4936dc..9e28cde 100644
--- a/src/remake/vote.rs
+++ b/src/remake/vote.rs
@@ -28,7 +28,7 @@ use tracing::{info, warn};
 
 use super::Host;
 use super::content;
-use super::records::{self, Decision, EpochRecord, Phase, Record, Register, Vote, unix_millis};
+use super::records::{self, Decision, EpochRecord, Phase, Record, Register, Vote};
 use crate::document::store::StoreError;
 
 /// A majority of `electorate` members.
@@ -235,19 +235,12 @@ pub(super) async fn cast<H: Host>(host: &H) {
 
     // Usually there is nothing to vote on: look first, without writing.
     let (voter, wanted) = (me.clone(), epoch.clone());
+    let now = host.now_millis();
     let due = host
         .read_doc(move |doc| {
             let all = records::load_all(doc);
             let register = records::load_register(doc);
-            let id = choice(
-                &voter,
-                &wanted,
-                &all,
-                &register,
-                started,
-                unix_millis(),
-                grace,
-            )?;
+            let id = choice(&voter, &wanted, &all, &register, started, now, grace)?;
             let content = all
                 .into_iter()
                 .find(|(other, _)| *other == id)
@@ -267,9 +260,10 @@ pub(super) async fn cast<H: Host>(host: &H) {
 
     // Decided again under the write lock, so two passes can't both vote.
     let voter = me.clone();
+    let clock = host.clock();
     let result = host
         .commit_doc(move |doc| -> Result<Option<String>, StoreError> {
-            let now = unix_millis();
+            let now = clock.now_millis();
             let all = records::load_all(doc);
             let register = records::load_register(doc);
             let Some(id) = choice(&voter, &epoch, &all, &register, started, now, grace)