From 5640bacf735c6728acf4bc22217b0c07ac4bf7f9 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 1 Nov 2026 19:35:20 +0100 Subject: document: allow writes during rejoin --- src/remake/catchup.rs | 388 +++++++++++++++++++++++++++----------------------- 1 file changed, 210 insertions(+), 178 deletions(-) (limited to 'src/remake/catchup.rs') 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(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(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(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>, before: Vec, after: Vec, 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, own_missing: usize, - had_history: bool, }, } @@ -345,118 +328,139 @@ async fn finish(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(©, 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::>() + }) + .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(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(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, 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(©, 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(©, 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(©, 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(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 )); } } -- cgit v1.3.1