aboutsummaryrefslogtreecommitdiffstats
path: root/src/remake/catchup.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/remake/catchup.rs')
-rw-r--r--src/remake/catchup.rs388
1 files changed, 210 insertions, 178 deletions
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 319037d..1ee9508 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -33,8 +33,6 @@
//! its own (`sideways`), its writes on the epoch it leaves going to the quarantine.
use std::collections::{BTreeMap, HashMap};
-use std::convert::Infallible;
-use std::sync::Arc;
use std::time::Duration;
use crate::document::{ActorId, ChangeHash, Document, ReadDoc};
@@ -231,27 +229,12 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
error: None,
}));
- // Keep answering reads, from a copy of the current document marked stale, before
- // the node stops being ready: no window in which reads get 503. Taken as writes
- // stop, so it holds every write answered.
- if let Some(copy) = host.refuse_writes(host.is_node_ready().await).await {
- host.set_stale_view(Some((local, copy)));
- }
- let was_ready = host.suspend().await;
let actor = ActorId::random();
host.remake().add_own_actor(actor.clone());
- // The fresh document takes no writes either until the rejoin is done
- // (`Replica::writable`): one checked as ready before the suspension above would
- // land in it.
- let old = match host
- .switch_epoch(target, false, move |_live| {
- Ok::<_, Infallible>(Document::new().with_actor(actor))
- })
- .await
- {
- Ok(old) => old,
- Err(never) => match never {},
- };
+ // At once: the fresh document takes no writes until the rejoin is done; a ready
+ // node's document so far answers reads, marked stale, and takes its writes
+ // meanwhile (`Host::begin_rejoin`).
+ let begun = host.begin_rejoin(target, actor).await;
// After the swap: messages of the old epoch are dropped from now on, so nothing
// can mark the new, empty document as caught up by mistake. A stale "done" from
// just before is caught in `finish` (the snapshot change is missing then).
@@ -261,9 +244,10 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
host.remake().start_rejoin(Rejoin {
previous: local,
target,
- old,
+ old: begun.old,
+ heads: begun.heads,
started: Instant::now(),
- was_ready,
+ was_ready: begun.was_ready,
});
// Sync waits until the snapshot is here (or its download failed).
@@ -304,13 +288,13 @@ async fn fetch_snapshot<H: Host>(host: H, target: Epoch) {
host.request_sync();
}
-/// What to do with this node's own changes the new epoch lacks.
+/// What to do with this node's own changes the new epoch lacks, those made before the
+/// rejoin.
enum Plan {
Nothing,
- /// Replay these paths (what changed from `before` to `after` in the old
- /// document), carried through every remake since the node's epoch.
+ /// Replay what changed from `before` to `after` in the old document, carried
+ /// through every remake since the node's epoch.
Replay {
- paths: Vec<Vec<String>>,
before: Vec<ChangeHash>,
after: Vec<ChangeHash>,
translator: Translator,
@@ -321,19 +305,18 @@ enum Plan {
},
}
-// Every variant holds a whole document; one value lives for one step of a rejoin, so
-// boxing would only add an indirection.
-#[allow(clippy::large_enum_variant)]
enum Prepared {
/// Caught up with someone, but not with the target epoch yet: keep waiting.
- NotYet(Document),
- Failed(Document, String),
+ NotYet,
+ Failed(String),
Ready {
- old: Document,
metadata: EpochMetadata,
plan: Plan,
+ /// Where the writes the old document took during the rejoin go: fresh, they
+ /// are replayed whatever the replay window says, unless there is no telling
+ /// where (the reason).
+ during: Result<Translator, String>,
own_missing: usize,
- had_history: bool,
},
}
@@ -345,118 +328,139 @@ async fn finish<H: Host>(host: &H) {
previous,
target,
old,
+ heads: began,
started,
was_ready,
} = rejoin;
- // Everything heavy runs on a copy, without a lock.
+ // Everything heavy runs on copies, without a lock; the old document takes writes
+ // meanwhile, and those after `began` are replayed under the lock in the end.
let copy = host.read_doc(|doc| doc.clone()).await;
+ let kept = host.rejoin_document().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, now)).await;
+ let prepared = {
+ let began = began.clone();
+ tokio::task::spawn_blocking(move || {
+ let empty = Document::new();
+ let reference = kept.as_ref().or(old.as_ref()).unwrap_or(&empty);
+ let prepared = prepare(&copy, reference, &began, previous, target, own, now);
+ (prepared, old)
+ })
+ .await
+ };
- let (old, metadata, plan, own_missing, had_history) = match prepared {
- Ok(Prepared::Ready {
+ let (metadata, plan, during, own_missing, old) = match prepared {
+ Ok((
+ Prepared::Ready {
+ metadata,
+ plan,
+ during,
+ own_missing,
+ },
old,
- metadata,
- plan,
- own_missing,
- had_history,
- }) => (old, metadata, plan, own_missing, had_history),
- Ok(Prepared::NotYet(old)) => {
+ )) => (metadata, plan, during, own_missing, old),
+ Ok((Prepared::NotYet, old)) => {
host.reset_initial_sync();
host.remake().start_rejoin(Rejoin {
previous,
target,
old,
+ heads: began,
started,
was_ready,
});
return;
}
- Ok(Prepared::Failed(old, reason)) => {
+ Ok((Prepared::Failed(reason), old)) => {
let rejoin = Rejoin {
previous,
target,
old,
+ heads: began,
started,
was_ready,
};
return abandon(host, rejoin, reason).await;
}
Err(join_error) => {
- // The old document is gone with the panicked task; nothing to restore.
+ // A document of a node that wasn't ready is gone with the panicked task; a
+ // ready node's writes during the rejoin wait in the quarantine.
error!(error = %join_error, "Rejoin task failed; staying on the new epoch");
- host.accept_writes().await;
- if was_ready {
- host.resume().await;
- }
- host.set_stale_view(None);
+ let (unplaced, old) = host
+ .finish_rejoin(None, move |_live, old| {
+ let reason = "written during a rejoin whose check failed".to_owned();
+ delta::changed_paths(old, &began, &old.get_heads())
+ .into_iter()
+ .map(|path| (path, reason.clone()))
+ .collect::<Vec<_>>()
+ })
+ .await;
+ quarantine::add(host, previous, target, &old, unplaced);
return;
}
};
- let mut stats = ReplayStats::default();
- let mut quarantine_id = None;
- let mut keep_archive = true;
- let mut old = Some(old);
-
- // The new epoch is verified: writes go in again, the replay first.
- host.accept_writes().await;
-
- match plan {
- Plan::Nothing => {}
- Plan::Replay {
- paths,
- before,
- after,
- translator,
- } => {
- let document = Arc::new(old.take().expect("old document"));
- let superseded = host.superseded();
- let replayed = {
- let document = Arc::clone(&document);
- host.commit_changes(move |doc| {
- delta::replay_newest(&document, &before, &after, doc, &translator, superseded)
- })
- .await
- };
- // The commit's task held the only other reference, and it is done.
- let document = Arc::into_inner(document).expect("the replay's reference is gone");
- let unplaced = match replayed {
- Ok(replayed) => {
- stats = replayed;
- // Writes the remakes since had no place for.
- std::mem::take(&mut stats.quarantined)
+ // The writes, under the lock, with the switch back to serving requests.
+ let quarantined_whole = matches!(plan, Plan::Quarantine { .. });
+ let superseded = host.superseded();
+ let ((stats, unplaced), old) = host
+ .finish_rejoin(old, move |live, old| {
+ let mut stats = ReplayStats::default();
+ let mut unplaced = Vec::new();
+ match plan {
+ Plan::Nothing => {}
+ Plan::Replay {
+ before,
+ after,
+ translator,
+ } => {
+ stats.absorb(delta::replay_newest(
+ old,
+ &before,
+ &after,
+ live,
+ &translator,
+ superseded,
+ ));
}
- // Not after `accept_writes`; but if so, nothing is lost.
- Err(error) => {
- let reason = format!("couldn't replay: {error}");
- paths
- .into_iter()
- .map(|path| (path, reason.clone()))
- .collect()
+ Plan::Quarantine { paths, reason } => {
+ unplaced.extend(paths.into_iter().map(|path| (path, reason.clone())));
}
- };
- quarantine_id = quarantine::add(host, previous, target, &document, unplaced);
- old = Some(document);
- }
- Plan::Quarantine { paths, reason } => {
- let document = old.take().expect("old document");
- let entries = paths
- .into_iter()
- .map(|path| (path, reason.clone()))
- .collect();
- quarantine_id = quarantine::add(host, previous, target, &document, entries);
- keep_archive = false; // the quarantine holds the document
- }
- }
+ }
+ let heads = old.get_heads();
+ match during {
+ Ok(translator) => {
+ stats.absorb(delta::replay_newest(
+ old,
+ &began,
+ &heads,
+ live,
+ &translator,
+ superseded,
+ ));
+ }
+ Err(reason) => {
+ let reason = format!("written during the rejoin, but {reason}");
+ unplaced.extend(
+ delta::changed_paths(old, &began, &heads)
+ .into_iter()
+ .map(|path| (path, reason.clone())),
+ );
+ }
+ }
+ // Writes the remakes since had no place for.
+ unplaced.append(&mut stats.quarantined);
+ (stats, unplaced)
+ })
+ .await;
+ let quarantine_id = quarantine::add(host, previous, target, &old, unplaced);
- if let (true, true, Some(document)) = (had_history, keep_archive, old) {
+ // Unless the quarantine holds the document.
+ if !old.get_heads().is_empty() && !quarantined_whole {
host.remake().archive(
previous,
- document,
+ old,
metadata.policy.archive_retention.clone(),
target.generation,
host.now_millis(),
@@ -476,11 +480,6 @@ async fn finish<H: Host>(host: &H) {
host.remake().set_catch_up(None);
let actor = host.read_doc(|doc| doc.get_actor().clone()).await;
host.remake().reset_own_actors(actor);
- if was_ready {
- host.resume().await;
- }
- // After resuming, so reads move from the stale view to the live document directly.
- host.set_stale_view(None);
info!(
from = %previous,
@@ -495,65 +494,40 @@ 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. `now`: Unix milliseconds, by the node's clock.
+/// Verifies the synced document `copy` and decides about the old document's own
+/// changes: those up to `began` (when the rejoin began) by the replay window, those
+/// after it (made during the rejoin) whatever it says. `own`: this node's actors
+/// (`RemakeState::own_actors`); the old document's own actor counts in any case.
+/// `now`: Unix milliseconds, by the node's clock.
fn prepare(
- copy: Document,
- old: Document,
+ copy: &Document,
+ old: &Document,
+ began: &[ChangeHash],
previous: Epoch,
target: Epoch,
mut own: Vec<ActorId>,
now: UnixMillis,
) -> Prepared {
if copy.get_change_by_hash(&ChangeHash(target.id)).is_none() {
- return Prepared::NotYet(old);
+ return Prepared::NotYet;
}
- let metadata = match snapshot::verify_document(&copy, target) {
+ let metadata = match snapshot::verify_document(copy, target) {
Ok(metadata) => metadata,
- Err(reason) => return Prepared::Failed(old, reason),
+ Err(reason) => return Prepared::Failed(reason),
};
- let old_heads = old.get_heads();
- if old_heads.is_empty() {
- return Prepared::Ready {
- old,
- metadata,
- plan: Plan::Nothing,
- own_missing: 0,
- had_history: false,
- };
- }
-
// The epoch that followed ours records (per actor) which of our changes it took.
// Snapshots carry earlier epochs' metadata forward, so this works across any gap.
- let next = snapshot::read_epoch_metadata(&copy, previous.generation + 1)
+ let next = snapshot::read_epoch_metadata(copy, previous.generation + 1)
.ok()
.filter(|next| next.previous == previous);
-
- own.push(old.get_actor().clone());
- let clock = next
- .as_ref()
- .map_or_else(BTreeMap::new, |next| next.clock.clone());
- let missing = delta::missing_from_snapshot(&old, &clock, &own);
- if missing.own == 0 {
- return Prepared::Ready {
- old,
- metadata,
- plan: Plan::Nothing,
- own_missing: 0,
- had_history: true,
- };
- }
-
- let paths = delta::changed_paths(&old, &missing.before, &old_heads);
let gap = target.generation - previous.generation;
// Writes made in our epoch go through every remake since, in order.
let translator = (previous.generation + 1..=target.generation).try_fold(
Translator::default(),
|mut translator, generation| {
- let epoch = snapshot::read_epoch_metadata(&copy, generation)?;
+ let epoch = snapshot::read_epoch_metadata(copy, generation)?;
let kind = epoch.kind().ok_or_else(|| {
format!(
"epoch generation {generation} was made by unknown remake kind {:?}",
@@ -564,7 +538,41 @@ fn prepare(
Ok::<_, String>(translator)
},
);
+ let during = match (&next, &translator) {
+ // Sideways, to another epoch of our generation: the same paths.
+ _ if gap == 0 => Ok(Translator::default()),
+ (Some(_), Ok(translator)) => Ok(translator.clone()),
+ (Some(_), Err(reason)) => Err(format!("cannot be translated: {reason}")),
+ (None, _) => Err(format!(
+ "the metadata of the epoch after {previous} is missing"
+ )),
+ };
+
+ if began.is_empty() {
+ return Prepared::Ready {
+ metadata,
+ plan: Plan::Nothing,
+ during,
+ own_missing: 0,
+ };
+ }
+
+ own.push(old.get_actor().clone());
+ let clock = next
+ .as_ref()
+ .map_or_else(BTreeMap::new, |next| next.clock.clone());
+ let missing = delta::missing_from_snapshot(old, &clock, &own);
+ if missing.own == 0 {
+ return Prepared::Ready {
+ metadata,
+ plan: Plan::Nothing,
+ during,
+ own_missing: 0,
+ };
+ }
+ // Up to when the rejoin began; what came after is `during`'s.
+ let paths = delta::changed_paths(old, &missing.before, began);
let plan = match (&next, translator) {
// Rejoined sideways (`sideways`): whatever was written on our epoch, the swarm
// went another way; nothing says where it would go.
@@ -590,9 +598,8 @@ fn prepare(
let away_ms = now.ms_since(next.proposed_at);
if next.policy.replay_window.covers(gap, away_ms) {
Plan::Replay {
- paths,
before: missing.before.clone(),
- after: old_heads.clone(),
+ after: began.to_vec(),
translator,
}
} else {
@@ -609,37 +616,27 @@ fn prepare(
};
Prepared::Ready {
- old,
metadata,
plan,
+ during,
own_missing: missing.own,
- had_history: true,
}
}
-/// Restores the old document and epoch, and retries later.
+/// Restores the old document (with the writes it took meanwhile) and epoch, and
+/// retries later.
async fn abandon<H: Host>(host: &H, rejoin: Rejoin, reason: String) {
let Rejoin {
previous,
target,
old,
- was_ready,
..
} = rejoin;
warn!(target = %target, reason = %reason, "Rejoin failed; restoring the previous epoch");
- if let Err(never) = host
- .switch_epoch(previous, true, move |_live| Ok::<_, Infallible>(old))
- .await
- {
- match never {}
- }
+ host.abandon_rejoin(previous, old).await;
// A node that was never ready must sync again before bootstrap may finish.
host.reset_initial_sync();
- if was_ready {
- host.resume().await;
- }
- host.set_stale_view(None);
let attempts = host
.remake()
@@ -740,32 +737,65 @@ mod tests {
#[test]
fn replays_within_the_window() {
let (old, swarm, target) = scenario(&json!({}));
+ let began = old.get_heads();
match prepare(
- swarm,
- old,
+ &swarm,
+ &old,
+ &began,
Epoch::GENESIS,
target,
Vec::new(),
UnixMillis(0),
) {
Prepared::Ready {
- plan: Plan::Replay { paths, .. },
+ plan: Plan::Replay { before, after, .. },
own_missing,
..
} => {
assert_eq!(own_missing, 1);
- assert_eq!(paths, vec![vec!["mine".to_owned()]]);
+ assert_eq!(
+ delta::changed_paths(&old, &before, &after),
+ vec![vec!["mine".to_owned()]]
+ );
}
_ => panic!("expected a replay"),
}
}
+ /// Writes made during the rejoin aren't the window's: those after `began` go
+ /// wherever the remakes since put them, even outside it.
+ #[test]
+ fn writes_during_the_rejoin_are_replayed_outside_the_window() {
+ let (old, swarm, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } }));
+ let began = old.get_heads();
+ let mut during = old.clone();
+ change(&mut during, |doc| doc.put(ROOT, "later", 1_i64).unwrap());
+ match prepare(
+ &swarm,
+ &during,
+ &began,
+ Epoch::GENESIS,
+ target,
+ Vec::new(),
+ UnixMillis(0),
+ ) {
+ Prepared::Ready {
+ plan: Plan::Quarantine { paths, .. },
+ during: Ok(_),
+ ..
+ } => assert_eq!(paths, vec![vec!["mine".to_owned()]]),
+ _ => panic!("expected a quarantine before, and a replay during the rejoin"),
+ }
+ }
+
#[test]
fn quarantines_outside_the_window() {
let (old, swarm, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } }));
+ let began = old.get_heads();
match prepare(
- swarm,
- old,
+ &swarm,
+ &old,
+ &began,
Epoch::GENESIS,
target,
Vec::new(),
@@ -784,8 +814,9 @@ mod tests {
let (_, swarm, target) = scenario(&json!({}));
assert!(matches!(
prepare(
- swarm,
- Document::new(),
+ &swarm,
+ &Document::new(),
+ &[],
Epoch::GENESIS,
target,
Vec::new(),
@@ -793,7 +824,7 @@ mod tests {
),
Prepared::Ready {
plan: Plan::Nothing,
- had_history: false,
+ own_missing: 0,
..
}
));
@@ -804,14 +835,15 @@ mod tests {
let (old, _, target) = scenario(&json!({}));
assert!(matches!(
prepare(
- Document::new(),
- old,
+ &Document::new(),
+ &old,
+ &old.get_heads(),
Epoch::GENESIS,
target,
Vec::new(),
UnixMillis(0)
),
- Prepared::NotYet(_)
+ Prepared::NotYet
));
}
}