summaryrefslogtreecommitdiffstats
path: root/src/remake/catchup.rs
diff options
context:
space:
mode:
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(_)
));
}