summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/api/peers.rs4
-rw-r--r--src/clock.rs53
-rw-r--r--src/main.rs2
-rw-r--r--src/node.rs6
-rw-r--r--src/remake/api.rs15
-rw-r--r--src/remake/catchup.rs34
-rw-r--r--src/remake/coordinator.rs9
-rw-r--r--src/remake/local.rs30
-rw-r--r--src/remake/mod.rs9
-rw-r--r--src/remake/quarantine.rs3
-rw-r--r--src/remake/records.rs7
-rw-r--r--src/remake/switch.rs18
-rw-r--r--src/remake/vote.rs16
-rw-r--r--src/simtest/mod.rs1
-rw-r--r--src/state.rs15
-rw-r--r--src/swarm/members.rs4
-rw-r--r--src/swarm/state_file.rs10
-rw-r--r--src/swarm/sync.rs3
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
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-26 11:35:11 +0000