document: reject writes during rejoin

1 files changed, 23 insertions, 8 deletions

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<H: Host>(State(host): State<H>, Path(id): Path<String
         return not_found();
     };
 
-    let (stats, entry) = host
-        .commit_doc(move |doc| {
+    let entry = Arc::new(entry);
+    let replayed = {
+        let entry = Arc::clone(&entry);
+        host.commit_doc(move |doc| {
             // As they are: the operator decided they belong at their old paths.
-            let stats =
-                delta::replay_paths(&entry.document, doc, &entry.paths, &Translator::default());
-            Ok::<_, Infallible>((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.