diff options
Diffstat (limited to 'src/remake/catchup.rs')
| -rw-r--r-- | src/remake/catchup.rs | 34 |
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(_) )); } |
