remake: copy stale view when writes stop

This ensures that stale reads show every write that was answered.

3 files changed, 24 insertions, 5 deletions

diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 43cc390..2da8540 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -233,16 +233,17 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
     }));
 
     // 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.
-    if host.is_node_ready().await {
-        let copy = host.read_doc(|doc| doc.clone()).await;
+    // 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 until the rejoin is done: one checked as
-    // ready just before the suspension above would land in it (`Replica::writable`).
+    // 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))
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index 04e9d8a..01e1e3e 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -197,6 +197,11 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
     /// ([`Host::switch_epoch`] with `writable` false).
     fn accept_writes(&self) -> impl Future<Output = ()> + 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<Output = Option<Document>> + Send;
+
     /// Applies a change downloaded from peers (`fetch_change`), then publishes it like
     /// a change received by sync.
     fn apply_change(
diff --git a/src/state/host.rs b/src/state/host.rs
index 7c3e165..30ebcdb 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -226,6 +226,19 @@ impl Host for AppState {
         async move { state.mutate_doc(f).await }
     }
 
+    fn refuse_writes(&self, copy: bool) -> impl Future<Output = Option<Document>> + 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")
+        }
+    }
+
     fn accept_writes(&self) -> impl Future<Output = ()> + Send {
         let state = self.clone();
         async move {