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.

18 files changed, 164 insertions, 75 deletions

diff --git a/src/api/peers.rs b/src/api/peers.rs
index dc4bd3b..a9694f7 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -16,7 +16,7 @@ use utoipa::ToSchema;
 use super::store_status;
 use crate::document::replica::Epoch;
 use crate::document::store;
-use crate::state::{AppState, HelloState, Incidents, Liveness, PeerState, unix_millis};
+use crate::state::{AppState, HelloState, Incidents, Liveness, PeerState};
 use crate::swarm::peer::forgotten_path;
 use crate::transport::traffic::PeerTraffic;
 
@@ -232,7 +232,7 @@ pub(crate) async fn delete_peer(
         }
     }
 
-    let record = serde_json::json!({ "at": unix_millis(), "by": state.local_id() });
+    let record = serde_json::json!({ "at": state.clock.now_millis(), "by": state.local_id() });
 
     let peer_clone = peer.clone();
     state
diff --git a/src/clock.rs b/src/clock.rs
new file mode 100644
index 0000000..dafd44b
--- /dev/null
+++ b/src/clock.rs
@@ -0,0 +1,53 @@
+//! Wall-clock time as this node sees it, in Unix milliseconds: commit stamps, alive
+//! stamps, remake deadlines and votes, archive and quarantine times.
+//!
+//! Every such reading goes through the node's [`Clock`], so tests can put one node's
+//! clock ahead of or behind the others (`docs/trust.md`, "What depends on clocks").
+//! Elapsed time within a node (timeouts, keepalives) uses tokio's monotonic clock and
+//! isn't affected.
+
+use std::sync::atomic::{AtomicI64, Ordering};
+
+/// The node's wall clock: the system clock plus an offset (zero outside tests).
+#[derive(Debug, Default)]
+pub(crate) struct Clock {
+    offset_ms: AtomicI64,
+}
+
+impl Clock {
+    /// Now, in Unix milliseconds.
+    pub(crate) fn now_millis(&self) -> u64 {
+        let offset = self.offset_ms.load(Ordering::Relaxed);
+        system_millis().saturating_add_signed(offset)
+    }
+
+    /// Puts this clock `offset_ms` ahead of the system clock (behind if negative).
+    #[cfg(test)]
+    pub(crate) fn set_offset_ms(&self, offset_ms: i64) {
+        self.offset_ms.store(offset_ms, Ordering::Relaxed);
+    }
+}
+
+/// The system clock, in Unix milliseconds; 0 if it is before 1970.
+fn system_millis() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn offset_moves_the_clock() {
+        let clock = Clock::default();
+        let before = clock.now_millis();
+        clock.set_offset_ms(3_600_000);
+        let ahead = clock.now_millis();
+        assert!(ahead >= before + 3_600_000);
+        clock.set_offset_ms(-3_600_000);
+        assert!(clock.now_millis() < before);
+    }
+}
diff --git a/src/main.rs b/src/main.rs
index e494497..d1e8053 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -14,6 +14,7 @@ use tracing::{debug, error, info, warn};
 
 mod api;
 mod backoff;
+mod clock;
 mod config;
 mod document;
 mod files;
@@ -178,6 +179,7 @@ async fn main() -> Result<()> {
         state_file,
         document_file,
         config_report: config.report(),
+        clock: Default::default(),
     });
     let state = node.state.clone();
 
diff --git a/src/node.rs b/src/node.rs
index 57d961c..5005d30 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -12,6 +12,7 @@ use automerge::{ActorId, AutoCommit};
 use tokio::sync::{RwLock, broadcast, mpsc, watch};
 use tracing::trace;
 
+use crate::clock::Clock;
 use crate::document::file::DocumentFile;
 use crate::document::replica::{Epoch, Replica};
 use crate::files::Files;
@@ -46,6 +47,8 @@ pub(crate) struct NodeConfig {
     pub(crate) document_file: Option<Arc<DocumentFile>>,
     /// For `GET /admin/config`.
     pub(crate) config_report: serde_json::Value,
+    /// The node's wall clock; `Clock::default()` outside tests.
+    pub(crate) clock: Arc<Clock>,
 }
 
 /// A built node whose workers haven't started yet.
@@ -91,7 +94,7 @@ impl Node {
             initial_sync_done: Arc::new(AtomicBool::new(false)),
             change_events,
             timing: config.timing,
-            remake: Arc::new(remake::RemakeState::default()),
+            remake: Arc::new(remake::RemakeState::new(config.clock.now_millis())),
             swarm_key: config.swarm_key,
             run_id: uuid::Uuid::new_v4().into_bytes(),
             neighbours_k: config.neighbours_k,
@@ -114,6 +117,7 @@ impl Node {
             catch_up_source: Default::default(),
             catch_up_preference: Default::default(),
             catch_up_avoided: Default::default(),
+            clock: config.clock,
         };
         for actor in config.earlier_actors {
             state.remake.add_own_actor(actor);
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)
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index 18975aa..d458338 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -88,6 +88,7 @@ impl Sim {
             state_file: None,
             document_file: None,
             config_report: JsonValue::Null,
+            clock: Default::default(),
         };
         adjust(&mut config);
         let state = Node::new(config).spawn_workers();
diff --git a/src/state.rs b/src/state.rs
index 78b058c..46ebfb9 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -676,6 +676,8 @@ pub(crate) struct AppState {
     /// A catch-up source given up for not answering our sync messages, and until when it
     /// isn't chosen again (`swarm::source`).
     pub(crate) catch_up_avoided: Arc<std::sync::Mutex<Option<(String, Instant)>>>,
+    /// The node's wall clock (`clock`).
+    pub(crate) clock: Arc<crate::clock::Clock>,
 }
 
 /// The peer a catching-up node syncs with first (`swarm::source`).
@@ -913,7 +915,7 @@ impl AppState {
                         // how long the change took to reach them (`swarm::stats`).
                         replica.doc.commit_with(
                             automerge::transaction::CommitOptions::default()
-                                .with_time(unix_millis() as i64),
+                                .with_time(state.clock.now_millis() as i64),
                         );
                     }
                     // Discard partial ops (e.g. old value deleted, new one not written).
@@ -1087,19 +1089,16 @@ impl AppState {
     }
 }
 
-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)
-}
-
 /// The node as seen by the remake module.
 impl Host for AppState {
     fn remake(&self) -> &RemakeState {
         &self.remake
     }
 
+    fn clock(&self) -> Arc<crate::clock::Clock> {
+        Arc::clone(&self.clock)
+    }
+
     fn node_id(&self) -> String {
         self.local_id().to_owned()
     }
diff --git a/src/swarm/members.rs b/src/swarm/members.rs
index 895e0cc..9b0de18 100644
--- a/src/swarm/members.rs
+++ b/src/swarm/members.rs
@@ -21,7 +21,7 @@ use serde_json::{Value as JsonValue, json};
 use tracing::{debug, info, warn};
 
 use crate::document::store::{self, StoreError};
-use crate::state::{AppState, unix_millis};
+use crate::state::AppState;
 
 const META_ROOT: &str = "_continuum";
 
@@ -168,7 +168,7 @@ pub(crate) async fn ensure_registered(state: &AppState) {
         )
     };
     let wanted = state.alias.clone();
-    let now = unix_millis();
+    let now = state.clock.now_millis();
 
     if !registered {
         let mut entry = json!({ "since": now, "alive": now });
diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs
index 62a6dbe..8b883fc 100644
--- a/src/swarm/state_file.rs
+++ b/src/swarm/state_file.rs
@@ -129,7 +129,12 @@ pub(crate) async fn update(state: &AppState, now: bool) {
 
     let seeds = {
         let replica = state.doc.read().await;
-        seeds(&replica.doc, state.local_id(), state.electorate_window)
+        seeds(
+            &replica.doc,
+            state.local_id(),
+            state.electorate_window,
+            state.clock.now_millis(),
+        )
     };
     let remembered = Remembered { swarm_id, seeds };
 
@@ -172,8 +177,7 @@ pub(crate) async fn update(state: &AppState, now: bool) {
 /// Members to rejoin through: those alive within the electorate window, stable ones
 /// first, at most `MAX_SEEDS`, sorted. Changes only when members join, leave or change
 /// tier, so the file rarely does.
-fn seeds(doc: &automerge::AutoCommit, me: &str, window: Duration) -> Vec<String> {
-    let now = crate::state::unix_millis();
+fn seeds(doc: &automerge::AutoCommit, me: &str, window: Duration, now: u64) -> Vec<String> {
     let (mut stable, transient): (Vec<String>, Vec<String>) = members::electorate(doc, now, window)
         .into_iter()
         .filter(|pk| pk != me)
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index d4503bb..a05e346 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -330,6 +330,7 @@ pub(crate) async fn handle_sync(
         let epoch = envelope.epoch;
         let message = envelope.message;
         let peer = peer.to_owned();
+        let clock = Arc::clone(&state.clock);
 
         // Some((caught_up, heads changed)) when applied, None when the message belongs
         // to another epoch.
@@ -366,7 +367,7 @@ pub(crate) async fn handle_sync(
             stats.heads_seen(heads_after.len());
             let changed = heads_after != heads_before;
             if changed && measure_propagation {
-                let now_ms = crate::state::unix_millis() as i64;
+                let now_ms = clock.now_millis() as i64;
                 // Time 0: not stamped (snapshots, older nodes).
                 let delays: Vec<i64> = replica
                     .doc