document: reject writes during rejoin

1 files changed, 9 insertions, 3 deletions

diff --git a/src/state/mod.rs b/src/state/mod.rs
index d8bdac5..9ee08b7 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -38,7 +38,7 @@ use tracing::{debug, error, warn};
 
 use crate::document::lock::DocumentLock;
 use crate::document::replica::{Epoch, Stamp};
-use crate::document::store::ChangeSet;
+use crate::document::store::{ChangeSet, StoreError};
 use crate::document::{Document, Transaction};
 use crate::locks::MutexExt;
 use crate::remake::{Host, RemakeState};
@@ -393,20 +393,26 @@ 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.
+    /// cancelled request can't leave a committed write unannounced. Refused
+    /// (`StoreError::Rejoining`, `f` not called) while the document isn't writable.
     pub(crate) async fn mutate_doc<T, E>(
         &self,
         f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static,
     ) -> Result<T, E>
     where
         T: Send + 'static,
-        E: Send + 'static,
+        E: From<StoreError> + Send + 'static,
     {
         let state = self.clone();
 
         tokio::task::spawn_blocking(move || {
             let result = {
                 let mut replica = state.doc.blocking_write();
+                // 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));
+                }
                 // One API write = one change; on error, its partial writes (e.g. the old
                 // value deleted, the new one not written) are dropped.
                 let result = replica.transact(state.clock.now_millis(), f);