From 885da82bb0d169f995598e0ba7c75523b1f98080 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 1 Nov 2026 13:37:04 +0100 Subject: remake: copy stale view when writes stop This ensures that stale reads show every write that was answered. --- src/remake/catchup.rs | 11 ++++++----- src/remake/mod.rs | 5 +++++ src/state/host.rs | 13 +++++++++++++ 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(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 + 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> + 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> + 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 + Send { let state = self.clone(); async move { -- cgit v1.3.1