From c52d4b5ae43c202569c8801c312c2b56954ff8a0 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 25 Oct 2026 23:00:56 +0100 Subject: 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. --- src/api/peers.rs | 4 ++-- src/clock.rs | 53 +++++++++++++++++++++++++++++++++++++++++++++++ src/main.rs | 2 ++ src/node.rs | 6 +++++- src/remake/api.rs | 15 +++++++------- src/remake/catchup.rs | 34 ++++++++++++++++++++++-------- src/remake/coordinator.rs | 9 ++++---- src/remake/local.rs | 30 ++++++++++++++++----------- src/remake/mod.rs | 9 ++++++++ src/remake/quarantine.rs | 3 +-- src/remake/records.rs | 7 ------- src/remake/switch.rs | 18 ++++++++++------ src/remake/vote.rs | 16 +++++--------- src/simtest/mod.rs | 1 + src/state.rs | 15 +++++++------- src/swarm/members.rs | 4 ++-- src/swarm/state_file.rs | 10 ++++++--- src/swarm/sync.rs | 3 ++- 18 files changed, 164 insertions(+), 75 deletions(-) create mode 100644 src/clock.rs 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>, /// For `GET /admin/config`. pub(crate) config_report: serde_json::Value, + /// The node's wall clock; `Clock::default()` outside tests. + pub(crate) clock: Arc, } /// 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(host: &H) -> u64 { fn report(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( _ => {} } - 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(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(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(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 { - 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(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(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(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, + 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(host: &H, id: &str) -> Result { let tally = vote::tally(id, &record, &all, ®ister); // 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(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(host: &H, id: &str, record: &Record) -> Result<(), Stor async fn decide_abort(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( ) -> Result, 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, StoreError> { let all = records::load_all(doc); let Some((_, record)) = all.iter().find(|(other, _)| *other == id) else { @@ -232,7 +233,7 @@ async fn decide( Decision::Committed { decided_at, .. } => { let register = records::load_register(doc); let tally = vote::tally(&id, record, &all, ®ister); - 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, } -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 { @@ -369,8 +372,7 @@ impl RemakeState { } /// An archive is kept until every limit of its retention has passed. -fn prune(archive: &mut Vec, current_generation: u64) { - let now = unix_millis(); +fn prune(archive: &mut Vec, 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; + + /// 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( 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(host: H) { async fn check(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(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(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( &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(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, - ®ister, - started, - unix_millis(), - grace, - )?; + let id = choice(&voter, &wanted, &all, ®ister, started, now, grace)?; let content = all .into_iter() .find(|(other, _)| *other == id) @@ -267,9 +260,10 @@ pub(super) async fn cast(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, 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, ®ister, 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>>, + /// The node's wall clock (`clock`). + pub(crate) clock: Arc, } /// 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 { + 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 { - let now = crate::state::unix_millis(); +fn seeds(doc: &automerge::AutoCommit, me: &str, window: Duration, now: u64) -> Vec { let (mut stable, transient): (Vec, Vec) = 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 = replica .doc -- cgit v1.3.1