document: reject writes during rejoin

1 files changed, 35 insertions, 17 deletions

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<H: Host>(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<H: Host>(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<H: Host>(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<H: Host>(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 {}