From 7a5a7408179c5fe9943bd2ce60295e86f1afaa40 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 1 Nov 2026 13:19:04 +0100 Subject: document: reject writes during rejoin --- src/remake/catchup.rs | 52 ++++++++++++++++++++++++++++++++---------------- src/remake/mod.rs | 18 ++++++++++++----- src/remake/quarantine.rs | 31 +++++++++++++++++++++-------- src/remake/switch.rs | 2 +- 4 files changed, 72 insertions(+), 31 deletions(-) (limited to 'src/remake') diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index 695e4d7..7269d8a 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -29,6 +29,7 @@ use std::collections::{BTreeMap, HashMap}; use std::convert::Infallible; +use std::sync::Arc; use std::time::Duration; use crate::document::{ActorId, ChangeHash, Document, ReadDoc}; @@ -43,6 +44,7 @@ use super::quarantine; use super::snapshot::{self, EpochMetadata}; use crate::clock::UnixMillis; use crate::document::replica::Epoch; +use crate::document::store::StoreError; /// Pause before trying the same target again after a failed rejoin. const RETRY: Duration = Duration::from_secs(30); @@ -179,8 +181,10 @@ async fn start(host: &H, local: Epoch, target: Epoch, attempts: u32) { let was_ready = host.suspend().await; let actor = ActorId::random(); host.remake().add_own_actor(actor.clone()); + // The fresh document takes no writes until the rejoin is done: one checked as + // ready just before the suspension above would land in it (`Replica::writable`). let old = match host - .switch_epoch(target, move |_live| { + .switch_epoch(target, false, move |_live| { Ok::<_, Infallible>(Document::new().with_actor(actor)) }) .await @@ -318,6 +322,7 @@ async fn finish(host: &H) { Err(join_error) => { // The old document is gone with the panicked task; nothing to restore. error!(error = %join_error, "Rejoin task failed; staying on the new epoch"); + host.accept_writes().await; if was_ready { host.resume().await; } @@ -331,26 +336,39 @@ async fn finish(host: &H) { 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, translator) => { - let document = old.take().expect("old document"); - let (replayed, document) = host - .commit_doc(move |doc| { - let stats = delta::replay_paths(&document, doc, &paths, &translator); - Ok::<_, Infallible>((stats, document)) + let document = Arc::new(old.take().expect("old document")); + let replayed = { + let document = Arc::clone(&document); + let paths = paths.clone(); + host.commit_doc(move |doc| { + Ok::<_, StoreError>(delta::replay_paths(&document, doc, &paths, &translator)) }) .await - .unwrap_or_else(|never| match never {}); - stats = replayed; - // Writes the remakes since had no place for. - quarantine_id = quarantine::add( - host, - previous, - target, - &document, - std::mem::take(&mut stats.quarantined), - ); + }; + // 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) + } + // 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() + } + }; + quarantine_id = quarantine::add(host, previous, target, &document, unplaced); old = Some(document); } Plan::Quarantine { paths, reason } => { @@ -525,7 +543,7 @@ async fn abandon(host: &H, rejoin: Rejoin, reason: String) { warn!(target = %target, reason = %reason, "Rejoin failed; restoring the previous epoch"); if let Err(never) = host - .switch_epoch(previous, move |_live| Ok::<_, Infallible>(old)) + .switch_epoch(previous, true, move |_live| Ok::<_, Infallible>(old)) .await { match never {} diff --git a/src/remake/mod.rs b/src/remake/mod.rs index e9d3021..04e9d8a 100644 --- a/src/remake/mod.rs +++ b/src/remake/mod.rs @@ -53,7 +53,7 @@ use axum::{ use tokio::sync::broadcast; use crate::document::replica::Epoch; -use crate::document::store::ChangeSet; +use crate::document::store::{ChangeSet, StoreError}; use crate::document::{Document, Transaction}; /// Remakes and their aftermath on this node, in numbers (`/admin/stats`). @@ -184,12 +184,18 @@ 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. + /// 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`]. fn commit_doc(&self, f: F) -> impl Future> + Send where F: FnOnce(&mut Transaction<'_>) -> Result + Send + 'static, T: Send + 'static, - E: Send + 'static; + E: From + 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; /// Applies a change downloaded from peers (`fetch_change`), then publishes it like /// a change received by sync. @@ -199,11 +205,13 @@ 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. 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, 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. fn switch_epoch( &self, epoch: Epoch, + writable: bool, f: F, ) -> impl Future> + Send where diff --git a/src/remake/quarantine.rs b/src/remake/quarantine.rs index 3ad5005..bb34d73 100644 --- a/src/remake/quarantine.rs +++ b/src/remake/quarantine.rs @@ -12,7 +12,7 @@ //! //! Quarantine lives in memory on the node that holds it. -use std::convert::Infallible; +use std::sync::Arc; use crate::document::Document; use axum::{ @@ -33,7 +33,7 @@ use super::local::Quarantined; use crate::api::error_response; use crate::clock::UnixMillis; use crate::document::replica::Epoch; -use crate::document::store; +use crate::document::store::{self, StoreError}; #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] @@ -167,15 +167,30 @@ pub(crate) async fn replay(State(host): State, Path(id): Path((stats, entry)) + Ok::<_, StoreError>(delta::replay_paths( + &entry.document, + doc, + &entry.paths, + &Translator::default(), + )) }) .await - .unwrap_or_else(|never| match never {}); + }; + // The commit's task held the only other reference, and it is done. + let entry = Arc::into_inner(entry).expect("the replay's reference is gone"); + let stats = match replayed { + Ok(stats) => stats, + Err(error) => { + // Rejoining since the check above: keep the entry for later. + host.remake().restore_quarantined(entry); + return error_response(StatusCode::SERVICE_UNAVAILABLE, &error.to_string(), None); + } + }; if stats.failed > 0 && stats.replayed == 0 { // Nothing went in; keep the entry for another look. diff --git a/src/remake/switch.rs b/src/remake/switch.rs index e60c942..9b03789 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -303,7 +303,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, move |live| { + .switch_epoch(target, true, 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()); -- cgit v1.3.1