summaryrefslogtreecommitdiffstats
path: root/src/remake
diff options
context:
space:
mode:
Diffstat (limited to 'src/remake')
-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
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)
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-26 14:39:14 +0000