From 1f3be9d059876f7f3994056b1ac8ce307628a52a 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/document/file.rs | 5 + src/document/replica.rs | 36 +++-- src/remake/catchup.rs | 388 ++++++++++++++++++++++++--------------------- src/remake/local.rs | 11 +- src/remake/mod.rs | 86 ++++++---- src/remake/switch.rs | 2 +- src/simtest/remakes.rs | 16 +- src/state/host.rs | 269 +++++++++++++++++++------------ src/state/mod.rs | 21 ++- src/state/read.rs | 27 +++- src/swarm/document_file.rs | 60 ++++++- 11 files changed, 572 insertions(+), 349 deletions(-) (limited to 'src') diff --git a/src/document/file.rs b/src/document/file.rs index 9efae42..3ea3afd 100644 --- a/src/document/file.rs +++ b/src/document/file.rs @@ -197,6 +197,11 @@ impl DocumentFile { /// What is due for a document of `epoch` at `heads`, written by this node's `actors` /// (see the module docs). + /// How many of this node's actors the file holds; None before it was written. + pub(crate) fn actors(&self) -> Option { + self.lock().as_ref().map(|written| written.actors) + } + pub(crate) fn due(&self, epoch: Epoch, actors: usize, heads: &[ChangeHash]) -> Due { let written = self.lock().clone(); let reason = match &written { diff --git a/src/document/replica.rs b/src/document/replica.rs index 61f2b3b..df2fafd 100644 --- a/src/document/replica.rs +++ b/src/document/replica.rs @@ -182,19 +182,7 @@ impl Replica { at: crate::clock::UnixMillis, f: impl FnOnce(&mut Transaction<'_>) -> Result, ) -> Result { - let mut transaction = self.doc.transaction(); - let out = f(&mut transaction); - match &out { - Ok(_) => { - transaction.commit_with( - automerge::transaction::CommitOptions::default().with_time(at.as_u64() as i64), - ); - } - Err(_) => { - transaction.rollback(); - } - } - out + transact(&mut self.doc, at, f) } } @@ -272,3 +260,25 @@ mod tests { assert_eq!(epoch(2, 0xab).to_string(), "2:abababababab"); } } + +/// The writes `f` makes to `doc`, as one change stamped `at` if it succeeds, dropped if +/// not ([`Replica::transact`], and writes into the document before a rejoin). +pub(crate) fn transact( + doc: &mut Document, + at: crate::clock::UnixMillis, + f: impl FnOnce(&mut Transaction<'_>) -> Result, +) -> Result { + let mut transaction = doc.transaction(); + let out = f(&mut transaction); + match &out { + Ok(_) => { + transaction.commit_with( + automerge::transaction::CommitOptions::default().with_time(at.as_u64() as i64), + ); + } + Err(_) => { + transaction.rollback(); + } + } + out +} 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 )); } } diff --git a/src/remake/local.rs b/src/remake/local.rs index e8fc9eb..ce133ec 100644 --- a/src/remake/local.rs +++ b/src/remake/local.rs @@ -16,7 +16,7 @@ use super::records::Retention; use super::snapshot::Snapshot; use crate::clock::UnixMillis; use crate::document::replica::Epoch; -use crate::document::{ActorId, Document}; +use crate::document::{ActorId, ChangeHash, Document}; use crate::locks::MutexExt; pub(crate) struct RemakeState { @@ -87,8 +87,13 @@ pub(crate) struct CatchUp { pub(crate) struct Rejoin { pub(crate) previous: Epoch, pub(crate) target: Epoch, - /// The document of `previous`, until it is replayed, quarantined or restored. - pub(crate) old: Document, + /// The document of `previous`, until it is replayed, quarantined or restored; of a + /// node that was ready, kept by the node instead, where it takes its writes + /// meanwhile (`Host::begin_rejoin`). + pub(crate) old: Option, + /// That document's heads as the rejoin began: the changes after them it took + /// during the rejoin. + pub(crate) heads: Vec, pub(crate) started: Instant, /// The node served requests before; it does again once the rejoin is over. pub(crate) was_ready: bool, diff --git a/src/remake/mod.rs b/src/remake/mod.rs index 0d3d073..25ca7ab 100644 --- a/src/remake/mod.rs +++ b/src/remake/mod.rs @@ -159,16 +159,44 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { /// Known (not forgotten) peers. fn peer_summaries(&self) -> impl Future> + Send; - /// Stops serving requests (back to `Joining`, keeping the swarm ID), as while a - /// rejoin syncs a fresh document. Returns whether the node was serving. - fn suspend(&self) -> impl Future + Send; - - /// Serves requests again after [`Host::suspend`]. - fn resume(&self) -> impl Future + Send; + /// Starts a rejoin of `target`, under the document's write lock: the live document + /// becomes an empty one of `target`, written by `actor`, that takes no writes until + /// the rejoin ends. A ready node stops being ready (back to `Joining`, keeping the + /// swarm ID), and its document so far answers reads (marked stale) and takes its + /// writes meanwhile; a node that wasn't ready gets its document back. Afterwards + /// the node resets every peer's sync state and publishes the change. + fn begin_rejoin( + &self, + target: Epoch, + actor: crate::document::ActorId, + ) -> impl Future + Send; + + /// A copy of the document before the rejoin, with the writes it took since, if the + /// node keeps it ([`Host::begin_rejoin`]: it was ready). + fn rejoin_document(&self) -> impl Future> + Send; + + /// Ends a rejoin on its target, under the write lock: `f` writes into the live + /// document from the one before the rejoin (the node's, or `old` where it keeps + /// none), as many changes as it likes; then writes go to the live document again, + /// and a node that was ready is again. Returns what `f` returned and the document + /// before the rejoin. + fn finish_rejoin( + &self, + old: Option, + f: F, + ) -> impl Future + Send + where + F: FnOnce(&mut Document, &Document) -> T + Send + 'static, + T: Send + 'static; - /// Serves reads from this document (read-only, marked stale) while the node is not - /// ready; `None` ends it. - fn set_stale_view(&self, view: Option<(Epoch, Document)>); + /// Ends a rejoin by going back, under the write lock: the document before it (the + /// node's, with the writes it took meanwhile, or `old`) and `previous` are live + /// again, and a node that was ready is again. Afterwards as [`Host::begin_rejoin`]. + fn abandon_rejoin( + &self, + previous: Epoch, + old: Option, + ) -> impl Future + Send; /// Whether the document caught up with a peer since the last reset. fn initial_sync_done(&self) -> bool; @@ -185,9 +213,9 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { F: FnOnce(&Document) -> T + Send + 'static, T: Send + 'static; - /// One committed change (rolled back on error), published and replicated. Refused - /// (`StoreError::Rejoining`) while the document isn't writable: from a rejoin's - /// swap until [`Host::accept_writes`]. + /// One committed change (rolled back on error), published and replicated. While a + /// rejoin runs, into the document before it, or refused (`StoreError::Rejoining`) + /// by a node that keeps none (`AppState::mutate_doc`). fn commit_doc(&self, f: F) -> impl Future> + Send where F: FnOnce(&mut Transaction<'_>) -> Result + Send + 'static, @@ -197,23 +225,6 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { /// What a replay does with writes a newer one superseded (`--superseded-writes`). fn superseded(&self) -> Superseded; - /// Changes `f` commits itself into the live document (as many as it likes, each - /// with its own time), under the write lock, then published and replicated. - /// Refused while the document isn't writable, as [`Host::commit_doc`]. - fn commit_changes(&self, f: F) -> impl Future> + Send - where - F: FnOnce(&mut Document) -> T + Send + 'static, - T: Send + 'static; - - /// Lets writes in again, after a rejoin swapped in a document that took none - /// ([`Host::switch_epoch`] with `writable` false). - fn accept_writes(&self) -> impl Future + Send; - - /// Stops taking writes (until [`Host::accept_writes`], or a swap that takes them) - /// and, with `copy`, returns a copy of the document as it is then, in one step: no - /// write is acknowledged that the copy lacks. - fn refuse_writes(&self, copy: bool) -> impl Future> + Send; - /// Applies a change downloaded from peers (`fetch_change`), then publishes it like /// a change received by sync. fn apply_change( @@ -222,13 +233,11 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { ) -> impl Future> + Send; /// Replaces the document and its epoch atomically. Under the write lock, `f` gets - /// the live document and returns the new one, which takes writes if `writable` - /// (a rejoin's fresh document doesn't). Afterwards the node resets every peer's - /// sync state and publishes the change. Returns the old document. + /// the live document and returns the new one. Afterwards the node resets every + /// peer's sync state and publishes the change. Returns the old document. fn switch_epoch( &self, epoch: Epoch, - writable: bool, f: F, ) -> impl Future> + Send where @@ -236,6 +245,17 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { E: Send + 'static; } +/// What [`Host::begin_rejoin`] left the rejoin with. +pub(crate) struct BegunRejoin { + /// The node was ready, and is again once the rejoin ends. + pub(crate) was_ready: bool, + /// The heads of the document before the rejoin as it began: what changes after + /// them, it took during the rejoin. + pub(crate) heads: Vec, + /// The document before the rejoin, where the node doesn't keep it (it wasn't ready). + pub(crate) old: Option, +} + #[derive(Clone, Debug)] pub(crate) struct PeerSummary { pub(crate) id: String, diff --git a/src/remake/switch.rs b/src/remake/switch.rs index 7d6611d..4c72b39 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -306,7 +306,7 @@ pub(super) async fn swap_in( let late_stats = std::sync::Arc::new(std::sync::Mutex::new(ReplayStats::default())); let late_out = std::sync::Arc::clone(&late_stats); let old = host - .switch_epoch(target, true, move |live| { + .switch_epoch(target, move |live| { // Only one worker switches, but be sure the copy still describes the live document. if !crate::document::changes::missing(live, ©_heads).is_empty() { return Err("the document was replaced while preparing the switch".to_owned()); diff --git a/src/simtest/remakes.rs b/src/simtest/remakes.rs index a063e2a..1c76520 100644 --- a/src/simtest/remakes.rs +++ b/src/simtest/remakes.rs @@ -231,11 +231,12 @@ async fn compaction_with_a_skewed_clock() { assert_eq!(read(&b, "data").await, Some(json!(1))); } -/// Writes on a node right as it starts to rejoin (`Replica::writable`): each is either -/// answered and kept, or refused; none lands in the fresh document before the epoch's -/// snapshot, which every change of the new epoch descends from. +/// Writes on a node through its rejoin: every one is answered and kept, those during +/// the rejoin in the document before it (`StaleView`) and carried over at the end; +/// none lands in the fresh document before the epoch's snapshot, which every change of +/// the new epoch descends from. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn writes_while_a_rejoin_starts_are_kept_or_refused() { +async fn writes_through_a_rejoin_are_kept() { use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; @@ -257,9 +258,8 @@ async fn writes_while_a_rejoin_starts_are_kept_or_refused() { let written = c .mutate_doc(move |doc| store::write_path(doc, &at, &json!(n))) .await; - if written.is_ok() { - answered.lock().unwrap().push(n); - } + assert!(written.is_ok(), "write {n} refused: {written:?}"); + answered.lock().unwrap().push(n); n += 1; tokio::time::sleep(Duration::from_millis(2)).await; } @@ -298,7 +298,7 @@ async fn writes_while_a_rejoin_starts_are_kept_or_refused() { assert!(reached, "change {start} doesn't descend from the snapshot"); } } - // Every write answered is there (those before the rejoin replayed into it). + // Every write is there (those before and during the rejoin replayed into it). let answered = answered.lock().unwrap().clone(); assert!(!answered.is_empty()); let c = &c; diff --git a/src/state/host.rs b/src/state/host.rs index eb7db35..e7921e4 100644 --- a/src/state/host.rs +++ b/src/state/host.rs @@ -11,12 +11,78 @@ use tokio::time::Instant; use tracing::debug; use super::{AppState, HelloState, NodeLifecycle, StaleView, Standing}; -use crate::document::replica::Epoch; +use crate::document::replica::{Epoch, Replica}; use crate::document::store::{ChangeSet, StoreError}; use crate::document::{ChangeHash, Document, ReadDoc, Transaction}; -use crate::remake::{Host, PeerSummary, RemakeState}; +use crate::locks::RwLockExt; +use crate::remake::{BegunRejoin, Host, PeerSummary, RemakeState}; impl AppState { + /// Under the document's write lock: `doc` of `epoch` replaces the live document, + /// which is returned. Every peer's sync state is reset before the lock is released, + /// so no message is generated from or applied to a state that describes the old + /// document. (Lock order doc -> peer, as in sync generation.) + fn swap_document(&self, replica: &mut Replica, epoch: Epoch, doc: Document) -> Document { + let old = std::mem::replace(&mut replica.doc, doc); + replica.epoch = epoch; + replica.epoch_seq += 1; + // Publishes the epoch with the heads (`Replica::stamps`), under the same lock, + // so it never disagrees with the document for anyone holding the lock. + replica.refresh_heads(); + let now = Instant::now(); + for peer in self.peer_states.blocking_read().values() { + let mut peer = peer.blocking_lock(); + peer.reset_sync(); + // A fresh start, not an idle sync: no heal re-handshake for it. + peer.last_sync_rx = Some(now); + } + old + } + + /// Stops serving requests (back to `Joining`, keeping the swarm ID), as while a + /// rejoin syncs a fresh document; on the blocking pool. Whether the node served. + fn suspend(&self) -> bool { + let mut lifecycle = self.lifecycle.blocking_write(); + let NodeLifecycle::Ready { swarm_id } = &*lifecycle else { + return false; + }; + let swarm_id = swarm_id.clone(); + debug!(swarm_id = %swarm_id, "Node suspended (rejoining)"); + *lifecycle = NodeLifecycle::Joining { + discovered_swarm_id: Some(swarm_id), + }; + true + } + + /// Serves requests again after [`suspend`](Self::suspend); on the blocking pool. + fn resume(&self) { + let mut lifecycle = self.lifecycle.blocking_write(); + if let NodeLifecycle::Joining { + discovered_swarm_id: Some(swarm_id), + } = &*lifecycle + { + let swarm_id = swarm_id.clone(); + debug!(swarm_id = %swarm_id, "Node resumed"); + *lifecycle = NodeLifecycle::Ready { swarm_id }; + } + } + + /// The document before a rejoin, out of the stale view (which ends) if the node + /// keeps it there, else `old`; and whether it was kept there (the node was ready). + fn take_stale(&self, old: Option) -> (Document, bool) { + let Some(view) = self.stale.write_or_recover().take() else { + return (old.unwrap_or_default(), false); + }; + // A read holding it a moment longer gets a copy of its own. + let doc = Arc::try_unwrap(view.doc) + .map(|doc| { + doc.into_inner() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + }) + .unwrap_or_else(|shared| shared.read_or_recover().clone()); + (doc, true) + } + /// After the document was replaced: every peer's sync state refers to the old /// document, so start over with all of them. Peers still on the old epoch are /// skipped by the sync worker until they switch too. @@ -143,54 +209,6 @@ impl Host for AppState { } } - fn suspend(&self) -> impl Future + Send { - let state = self.clone(); - async move { - let mut lifecycle = state.lifecycle.write().await; - let NodeLifecycle::Ready { swarm_id } = &*lifecycle else { - return false; - }; - let swarm_id = swarm_id.clone(); - debug!(swarm_id = %swarm_id, "Node suspended (rejoining)"); - *lifecycle = NodeLifecycle::Joining { - discovered_swarm_id: Some(swarm_id), - }; - true - } - } - - fn resume(&self) -> impl Future + Send { - let state = self.clone(); - async move { - let resumed = { - let mut lifecycle = state.lifecycle.write().await; - let swarm_id = match &*lifecycle { - NodeLifecycle::Joining { - discovered_swarm_id: Some(swarm_id), - } => Some(swarm_id.clone()), - _ => None, - }; - if let Some(swarm_id) = &swarm_id { - debug!(swarm_id = %swarm_id, "Node resumed"); - *lifecycle = NodeLifecycle::Ready { - swarm_id: swarm_id.clone(), - }; - } - swarm_id.is_some() - }; - if resumed { - crate::swarm::peer::announce(&state); - } - } - } - - fn set_stale_view(&self, view: Option<(Epoch, Document)>) { - self.set_stale(view.map(|(epoch, doc)| StaleView { - epoch, - doc: Arc::new(doc), - })); - } - fn initial_sync_done(&self) -> bool { self.initial_sync_done .load(std::sync::atomic::Ordering::Acquire) @@ -230,51 +248,121 @@ impl Host for AppState { self.superseded_writes } - fn commit_changes(&self, f: F) -> impl Future> + Send - where - F: FnOnce(&mut Document) -> T + Send + 'static, - T: Send + 'static, - { + fn begin_rejoin( + &self, + target: Epoch, + actor: crate::document::ActorId, + ) -> impl Future + Send { let state = self.clone(); async move { - tokio::task::spawn_blocking(move || { - let result = { + let begun = { + let state = state.clone(); + tokio::task::spawn_blocking(move || { let mut replica = state.doc.blocking_write(); - if !replica.writable { - return Err(StoreError::Rejoining); + let previous = replica.epoch; + let fresh = Document::new().with_actor(actor); + let old = state.swap_document(&mut replica, target, fresh); + replica.writable = false; + let heads = old.get_heads(); + // Under the document's lock: no read or write in between finds the + // node ready on the fresh document. + let was_ready = state.suspend(); + let old = if was_ready { + state.set_stale(Some(StaleView { + epoch: previous, + doc: Arc::new(std::sync::RwLock::new(old)), + })); + None + } else { + Some(old) + }; + BegunRejoin { + was_ready, + heads, + old, } - let result = f(&mut replica.doc); - replica.refresh_heads(); - result - }; - state.publish_change(); - Ok(result) - }) - .await - .expect("document write task panicked") + }) + .await + .expect("rejoin task panicked") + }; + state.after_epoch_switch().await; + begun } } - fn refuse_writes(&self, copy: bool) -> impl Future> + Send { + fn rejoin_document(&self) -> impl Future> + Send { let state = self.clone(); async move { - tokio::task::spawn_blocking(move || { - let mut replica = state.doc.blocking_write(); - replica.writable = false; - copy.then(|| replica.doc.clone()) - }) - .await - .expect("document write task panicked") + let (_, stale) = state.stale_document()?; + tokio::task::spawn_blocking(move || stale.read_or_recover().clone()) + .await + .ok() } } - fn accept_writes(&self) -> impl Future + Send { + fn finish_rejoin( + &self, + old: Option, + f: F, + ) -> impl Future + Send + where + F: FnOnce(&mut Document, &Document) -> T + Send + 'static, + T: Send + 'static, + { let state = self.clone(); async move { - // On the blocking pool, as every write lock is taken. - tokio::task::spawn_blocking(move || state.doc.blocking_write().writable = true) + let (out, old, resumed) = { + let state = state.clone(); + tokio::task::spawn_blocking(move || { + let mut replica = state.doc.blocking_write(); + // No write goes into the document before the rejoin any more: they + // take the document's lock first (`mutate_doc`). + let (old, resumed) = state.take_stale(old); + let out = f(&mut replica.doc, &old); + replica.writable = true; + replica.refresh_heads(); + if resumed { + state.resume(); + } + (out, old, resumed) + }) .await - .expect("document write task panicked"); + .expect("rejoin task panicked") + }; + state.publish_change(); + if resumed { + crate::swarm::peer::announce(&state); + } + (out, old) + } + } + + fn abandon_rejoin( + &self, + previous: Epoch, + old: Option, + ) -> impl Future + Send { + let state = self.clone(); + async move { + let resumed = { + let state = state.clone(); + tokio::task::spawn_blocking(move || { + let mut replica = state.doc.blocking_write(); + let (old, resumed) = state.take_stale(old); + state.swap_document(&mut replica, previous, old); + replica.writable = true; + if resumed { + state.resume(); + } + resumed + }) + .await + .expect("rejoin task panicked") + }; + state.after_epoch_switch().await; + if resumed { + crate::swarm::peer::announce(&state); + } } } @@ -307,7 +395,6 @@ impl Host for AppState { fn switch_epoch( &self, epoch: Epoch, - writable: bool, f: F, ) -> impl Future> + Send where @@ -321,25 +408,7 @@ impl Host for AppState { tokio::task::spawn_blocking(move || -> Result { let mut replica = state.doc.blocking_write(); let new_doc = f(&replica.doc)?; - let old_doc = std::mem::replace(&mut replica.doc, new_doc); - replica.epoch = epoch; - replica.epoch_seq += 1; - replica.writable = writable; - // Publishes the epoch with the heads (`Replica::stamps`), under the - // same lock, so it never disagrees with the document for anyone - // holding the lock. - replica.refresh_heads(); - // Reset every peer's sync state before the lock is released, so no - // message is generated from or applied to a state that describes the - // old document. (Lock order doc -> peer, as in sync generation.) - let now = Instant::now(); - for peer in state.peer_states.blocking_read().values() { - let mut peer = peer.blocking_lock(); - peer.reset_sync(); - // A fresh start, not an idle sync: no heal re-handshake for it. - peer.last_sync_rx = Some(now); - } - Ok(old_doc) + Ok(state.swap_document(&mut replica, epoch, new_doc)) }) .await .expect("epoch switch task panicked") diff --git a/src/state/mod.rs b/src/state/mod.rs index 6e6e48c..7c1299f 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -40,7 +40,7 @@ use crate::document::lock::DocumentLock; use crate::document::replica::{Epoch, Stamp}; use crate::document::store::{ChangeSet, StoreError}; use crate::document::{Document, Transaction}; -use crate::locks::MutexExt; +use crate::locks::{MutexExt, RwLockExt}; use crate::remake::{Host, RemakeState}; use crate::swarm::sync::SyncEvent; use crate::swarm::throughput::ThroughputTable; @@ -394,8 +394,10 @@ impl AppState { /// Mutates the document on the blocking pool as one change, then /// publishes it. Write and notification run in the same blocking task, so a - /// cancelled request can't leave a committed write unannounced. Refused - /// (`StoreError::Rejoining`, `f` not called) while the document isn't writable. + /// cancelled request can't leave a committed write unannounced. While a rejoin runs, + /// the write goes into the document before it (`StaleView`), replayed when the + /// rejoin ends; on a node that keeps none, it is refused (`StoreError::Rejoining`, + /// `f` not called). pub(crate) async fn mutate_doc( &self, f: impl FnOnce(&mut Transaction<'_>) -> Result + Send + 'static, @@ -412,7 +414,18 @@ impl AppState { // Checked here, under the lock: the caller's readiness check (if any) // was before a rejoin could swap in a fresh document. if !replica.writable { - return Err(E::from(StoreError::Rejoining)); + // Rejoining: into the document before the rejoin, which the rejoin + // replays when it ends (`StaleView`); refused by a node that keeps + // none (it wasn't ready). + let Some((_, stale)) = state.stale_document() else { + return Err(E::from(StoreError::Rejoining)); + }; + let mut doc = stale.write_or_recover(); + return crate::document::replica::transact( + &mut doc, + state.clock.now_millis(), + f, + ); } // One API write = one change; on error, its partial writes (e.g. the old // value deleted, the new one not written) are dropped. diff --git a/src/state/read.rs b/src/state/read.rs index a9b3116..e325e61 100644 --- a/src/state/read.rs +++ b/src/state/read.rs @@ -1,4 +1,5 @@ -//! Serving reads: from the live document, or from the pre-rejoin copy while rejoining. +//! Serving reads: from the live document, or from the pre-rejoin document while +//! rejoining. use std::sync::Arc; @@ -7,10 +8,13 @@ use crate::document::replica::Epoch; use crate::document::{ChangeHash, Document}; use crate::locks::RwLockExt; -/// A read-only view of the document the node served before it started rejoining. +/// The document the node served before it started rejoining, of the epoch it left: +/// reads are answered from it (marked stale), and writes go into it +/// (`AppState::mutate_doc`), until the rejoin replays them into the new epoch or +/// restores this document (`Host::finish_rejoin`, `Host::abandon_rejoin`). pub(crate) struct StaleView { pub(crate) epoch: Epoch, - pub(crate) doc: Arc, + pub(crate) doc: Arc>, } /// Where a read was served from. @@ -57,9 +61,12 @@ impl AppState { return Err(axum::http::StatusCode::SERVICE_UNAVAILABLE); }; - let rendered = tokio::task::spawn_blocking(move || finish(read(&doc))) - .await - .expect("document read task panicked"); + let rendered = tokio::task::spawn_blocking(move || { + let data = read(&doc.read_or_recover()); + finish(data) + }) + .await + .expect("document read task panicked"); Ok((rendered, ReadSource::Stale(epoch), None)) } @@ -93,4 +100,12 @@ impl AppState { pub(super) fn set_stale(&self, view: Option) { *self.stale.write_or_recover() = view; } + + /// The stale view's document and epoch, if there is one. + pub(crate) fn stale_document(&self) -> Option<(Epoch, Arc>)> { + self.stale + .read_or_recover() + .as_ref() + .map(|view| (view.epoch, Arc::clone(&view.doc))) + } } diff --git a/src/swarm/document_file.rs b/src/swarm/document_file.rs index 6a516a1..e9deca0 100644 --- a/src/swarm/document_file.rs +++ b/src/swarm/document_file.rs @@ -1,8 +1,15 @@ //! The worker that keeps the document file (`document::file`) up to date: it looks at -//! the node (ready, not rejoining, the document's epoch and heads, this node's actors), -//! and the file says what is due. +//! the node (ready, the document's epoch and heads, this node's actors), and the file +//! says what is due. While a ready node rejoins, the file follows the document before +//! the rejoin, which takes the node's writes meanwhile (`state::StaleView`): a restart +//! then resumes on it, and rejoins again. +use std::sync::Arc; + +use crate::document::Document; use crate::document::file::Due; +use crate::document::replica::Epoch; +use crate::locks::RwLockExt; use crate::state::AppState; /// Keeps the document file up to date. Runs for as long as the node does. @@ -23,7 +30,13 @@ pub(crate) async fn flush(state: &AppState) { let Some(file) = state.document_file.clone() else { return; }; - if !state.is_ready().await || state.remake.is_rejoining() { + if state.remake.is_rejoining() { + if let Some((epoch, stale)) = state.stale_document() { + flush_before_rejoin(&file, epoch, stale).await; + } + return; + } + if !state.is_ready().await { return; } let own = state.remake.own_actors(); @@ -54,3 +67,44 @@ pub(crate) async fn flush(state: &AppState) { Due::Nothing => {} } } + +/// [`flush`] of the document before a rejoin (`epoch`'s), while it takes this node's +/// writes. Only appends: a rewrite (another epoch, or the increments outgrew the full +/// save) waits for the end of the rejoin, so the file keeps the actors it had. +async fn flush_before_rejoin( + file: &Arc, + epoch: Epoch, + stale: Arc>, +) { + let heads = { + let stale = Arc::clone(&stale); + // Sorted, as the live document's (`Replica::heads`) the file compares with. + tokio::task::spawn_blocking(move || { + let mut heads = stale.read_or_recover().get_heads(); + heads.sort_unstable(); + heads + }) + .await + .expect("document read task panicked") + }; + let Some(actors) = file.actors() else { + return; + }; + if let Due::Append { + epoch: written, + since, + } = file.due(epoch, actors, &heads) + && written == epoch + { + let since_then = since.clone(); + let (heads, increment) = tokio::task::spawn_blocking(move || { + let doc = stale.read_or_recover(); + let mut heads = doc.get_heads(); + heads.sort_unstable(); + (heads, crate::document::save_since(&doc, &since_then)) + }) + .await + .expect("document read task panicked"); + file.append(since, heads, increment).await; + } +} -- cgit v1.3.1