aboutsummaryrefslogtreecommitdiffstats
path: root/src/remake/catchup.rs
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-10-25 23:00:56 +0100
committertv <tv@krebsco.de>2026-10-25 23:00:56 +0100
commit05d1694622a7ef181dbbb4b9c0aeb2f8c07a5b48 (patch)
treec71b14e64257a700771c7af5ed0a86c0fb0f7cbd /src/remake/catchup.rs
parenta76e50a0a97de28fd22a0c6c366b31f1b40859f8 (diff)
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.
Diffstat (limited to 'src/remake/catchup.rs')
-rw-r--r--src/remake/catchup.rs34
1 files changed, 25 insertions, 9 deletions
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(_)
));
}