document: allow writes during rejoin

4 files changed, 271 insertions, 214 deletions

diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 319037d..1ee9508 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -33,8 +33,6 @@
 //! its own (`sideways`), its writes on the epoch it leaves going to the quarantine.
 
 use std::collections::{BTreeMap, HashMap};
-use std::convert::Infallible;
-use std::sync::Arc;
 use std::time::Duration;
 
 use crate::document::{ActorId, ChangeHash, Document, ReadDoc};
@@ -231,27 +229,12 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
         error: None,
     }));
 
-    // 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. 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 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))
-        })
-        .await
-    {
-        Ok(old) => old,
-        Err(never) => match never {},
-    };
+    // At once: the fresh document takes no writes until the rejoin is done; a ready
+    // node's document so far answers reads, marked stale, and takes its writes
+    // meanwhile (`Host::begin_rejoin`).
+    let begun = host.begin_rejoin(target, actor).await;
     // After the swap: messages of the old epoch are dropped from now on, so nothing
     // can mark the new, empty document as caught up by mistake. A stale "done" from
     // just before is caught in `finish` (the snapshot change is missing then).
@@ -261,9 +244,10 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
     host.remake().start_rejoin(Rejoin {
         previous: local,
         target,
-        old,
+        old: begun.old,
+        heads: begun.heads,
         started: Instant::now(),
-        was_ready,
+        was_ready: begun.was_ready,
     });
 
     // Sync waits until the snapshot is here (or its download failed).
@@ -304,13 +288,13 @@ async fn fetch_snapshot<H: Host>(host: H, target: Epoch) {
     host.request_sync();
 }
 
-/// What to do with this node's own changes the new epoch lacks.
+/// What to do with this node's own changes the new epoch lacks, those made before the
+/// rejoin.
 enum Plan {
     Nothing,
-    /// Replay these paths (what changed from `before` to `after` in the old
-    /// document), carried through every remake since the node's epoch.
+    /// Replay what changed from `before` to `after` in the old document, carried
+    /// through every remake since the node's epoch.
     Replay {
-        paths: Vec<Vec<String>>,
         before: Vec<ChangeHash>,
         after: Vec<ChangeHash>,
         translator: Translator,
@@ -321,19 +305,18 @@ enum Plan {
     },
 }
 
-// Every variant holds a whole document; one value lives for one step of a rejoin, so
-// boxing would only add an indirection.
-#[allow(clippy::large_enum_variant)]
 enum Prepared {
     /// Caught up with someone, but not with the target epoch yet: keep waiting.
-    NotYet(Document),
-    Failed(Document, String),
+    NotYet,
+    Failed(String),
     Ready {
-        old: Document,
         metadata: EpochMetadata,
         plan: Plan,
+        /// Where the writes the old document took during the rejoin go: fresh, they
+        /// are replayed whatever the replay window says, unless there is no telling
+        /// where (the reason).
+        during: Result<Translator, String>,
         own_missing: usize,
-        had_history: bool,
     },
 }
 
@@ -345,118 +328,139 @@ async fn finish<H: Host>(host: &H) {
         previous,
         target,
         old,
+        heads: began,
         started,
         was_ready,
     } = rejoin;
 
-    // Everything heavy runs on a copy, without a lock.
+    // Everything heavy runs on copies, without a lock; the old document takes writes
+    // meanwhile, and those after `began` are replayed under the lock in the end.
     let copy = host.read_doc(|doc| doc.clone()).await;
+    let kept = host.rejoin_document().await;
     let own = host.remake().own_actors();
     let now = host.now_millis();
-    let prepared =
-        tokio::task::spawn_blocking(move || prepare(copy, old, previous, target, own, now)).await;
+    let prepared = {
+        let began = began.clone();
+        tokio::task::spawn_blocking(move || {
+            let empty = Document::new();
+            let reference = kept.as_ref().or(old.as_ref()).unwrap_or(&empty);
+            let prepared = prepare(&copy, reference, &began, previous, target, own, now);
+            (prepared, old)
+        })
+        .await
+    };
 
-    let (old, metadata, plan, own_missing, had_history) = match prepared {
-        Ok(Prepared::Ready {
+    let (metadata, plan, during, own_missing, old) = match prepared {
+        Ok((
+            Prepared::Ready {
+                metadata,
+                plan,
+                during,
+                own_missing,
+            },
             old,
-            metadata,
-            plan,
-            own_missing,
-            had_history,
-        }) => (old, metadata, plan, own_missing, had_history),
-        Ok(Prepared::NotYet(old)) => {
+        )) => (metadata, plan, during, own_missing, old),
+        Ok((Prepared::NotYet, old)) => {
             host.reset_initial_sync();
             host.remake().start_rejoin(Rejoin {
                 previous,
                 target,
                 old,
+                heads: began,
                 started,
                 was_ready,
             });
             return;
         }
-        Ok(Prepared::Failed(old, reason)) => {
+        Ok((Prepared::Failed(reason), old)) => {
             let rejoin = Rejoin {
                 previous,
                 target,
                 old,
+                heads: began,
                 started,
                 was_ready,
             };
             return abandon(host, rejoin, reason).await;
         }
         Err(join_error) => {
-            // The old document is gone with the panicked task; nothing to restore.
+            // A document of a node that wasn't ready is gone with the panicked task; a
+            // ready node's writes during the rejoin wait in the quarantine.
             error!(error = %join_error, "Rejoin task failed; staying on the new epoch");
-            host.accept_writes().await;
-            if was_ready {
-                host.resume().await;
-            }
-            host.set_stale_view(None);
+            let (unplaced, old) = host
+                .finish_rejoin(None, move |_live, old| {
+                    let reason = "written during a rejoin whose check failed".to_owned();
+                    delta::changed_paths(old, &began, &old.get_heads())
+                        .into_iter()
+                        .map(|path| (path, reason.clone()))
+                        .collect::<Vec<_>>()
+                })
+                .await;
+            quarantine::add(host, previous, target, &old, unplaced);
             return;
         }
     };
 
-    let mut stats = ReplayStats::default();
-    let mut quarantine_id = None;
-    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,
-            before,
-            after,
-            translator,
-        } => {
-            let document = Arc::new(old.take().expect("old document"));
-            let superseded = host.superseded();
-            let replayed = {
-                let document = Arc::clone(&document);
-                host.commit_changes(move |doc| {
-                    delta::replay_newest(&document, &before, &after, doc, &translator, superseded)
-                })
-                .await
-            };
-            // 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)
+    // The writes, under the lock, with the switch back to serving requests.
+    let quarantined_whole = matches!(plan, Plan::Quarantine { .. });
+    let superseded = host.superseded();
+    let ((stats, unplaced), old) = host
+        .finish_rejoin(old, move |live, old| {
+            let mut stats = ReplayStats::default();
+            let mut unplaced = Vec::new();
+            match plan {
+                Plan::Nothing => {}
+                Plan::Replay {
+                    before,
+                    after,
+                    translator,
+                } => {
+                    stats.absorb(delta::replay_newest(
+                        old,
+                        &before,
+                        &after,
+                        live,
+                        &translator,
+                        superseded,
+                    ));
                 }
-                // 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()
+                Plan::Quarantine { paths, reason } => {
+                    unplaced.extend(paths.into_iter().map(|path| (path, reason.clone())));
                 }
-            };
-            quarantine_id = quarantine::add(host, previous, target, &document, unplaced);
-            old = Some(document);
-        }
-        Plan::Quarantine { paths, reason } => {
-            let document = old.take().expect("old document");
-            let entries = paths
-                .into_iter()
-                .map(|path| (path, reason.clone()))
-                .collect();
-            quarantine_id = quarantine::add(host, previous, target, &document, entries);
-            keep_archive = false; // the quarantine holds the document
-        }
-    }
+            }
+            let heads = old.get_heads();
+            match during {
+                Ok(translator) => {
+                    stats.absorb(delta::replay_newest(
+                        old,
+                        &began,
+                        &heads,
+                        live,
+                        &translator,
+                        superseded,
+                    ));
+                }
+                Err(reason) => {
+                    let reason = format!("written during the rejoin, but {reason}");
+                    unplaced.extend(
+                        delta::changed_paths(old, &began, &heads)
+                            .into_iter()
+                            .map(|path| (path, reason.clone())),
+                    );
+                }
+            }
+            // Writes the remakes since had no place for.
+            unplaced.append(&mut stats.quarantined);
+            (stats, unplaced)
+        })
+        .await;
+    let quarantine_id = quarantine::add(host, previous, target, &old, unplaced);
 
-    if let (true, true, Some(document)) = (had_history, keep_archive, old) {
+    // Unless the quarantine holds the document.
+    if !old.get_heads().is_empty() && !quarantined_whole {
         host.remake().archive(
             previous,
-            document,
+            old,
             metadata.policy.archive_retention.clone(),
             target.generation,
             host.now_millis(),
@@ -476,11 +480,6 @@ async fn finish<H: Host>(host: &H) {
     host.remake().set_catch_up(None);
     let actor = host.read_doc(|doc| doc.get_actor().clone()).await;
     host.remake().reset_own_actors(actor);
-    if was_ready {
-        host.resume().await;
-    }
-    // After resuming, so reads move from the stale view to the live document directly.
-    host.set_stale_view(None);
 
     info!(
         from = %previous,
@@ -495,65 +494,40 @@ async fn finish<H: Host>(host: &H) {
     );
 }
 
-/// Verifies the synced document and decides about the old document's own changes.
-/// `own`: this node's actors (`RemakeState::own_actors`); the old document's own actor
-/// counts in any case. `now`: Unix milliseconds, by the node's clock.
+/// Verifies the synced document `copy` and decides about the old document's own
+/// changes: those up to `began` (when the rejoin began) by the replay window, those
+/// after it (made during the rejoin) whatever it says. `own`: this node's actors
+/// (`RemakeState::own_actors`); the old document's own actor counts in any case.
+/// `now`: Unix milliseconds, by the node's clock.
 fn prepare(
-    copy: Document,
-    old: Document,
+    copy: &Document,
+    old: &Document,
+    began: &[ChangeHash],
     previous: Epoch,
     target: Epoch,
     mut own: Vec<ActorId>,
     now: UnixMillis,
 ) -> Prepared {
     if copy.get_change_by_hash(&ChangeHash(target.id)).is_none() {
-        return Prepared::NotYet(old);
+        return Prepared::NotYet;
     }
-    let metadata = match snapshot::verify_document(&copy, target) {
+    let metadata = match snapshot::verify_document(copy, target) {
         Ok(metadata) => metadata,
-        Err(reason) => return Prepared::Failed(old, reason),
+        Err(reason) => return Prepared::Failed(reason),
     };
 
-    let old_heads = old.get_heads();
-    if old_heads.is_empty() {
-        return Prepared::Ready {
-            old,
-            metadata,
-            plan: Plan::Nothing,
-            own_missing: 0,
-            had_history: false,
-        };
-    }
-
     // The epoch that followed ours records (per actor) which of our changes it took.
     // Snapshots carry earlier epochs' metadata forward, so this works across any gap.
-    let next = snapshot::read_epoch_metadata(&copy, previous.generation + 1)
+    let next = snapshot::read_epoch_metadata(copy, previous.generation + 1)
         .ok()
         .filter(|next| next.previous == previous);
-
-    own.push(old.get_actor().clone());
-    let clock = next
-        .as_ref()
-        .map_or_else(BTreeMap::new, |next| next.clock.clone());
-    let missing = delta::missing_from_snapshot(&old, &clock, &own);
-    if missing.own == 0 {
-        return Prepared::Ready {
-            old,
-            metadata,
-            plan: Plan::Nothing,
-            own_missing: 0,
-            had_history: true,
-        };
-    }
-
-    let paths = delta::changed_paths(&old, &missing.before, &old_heads);
     let gap = target.generation - previous.generation;
 
     // Writes made in our epoch go through every remake since, in order.
     let translator = (previous.generation + 1..=target.generation).try_fold(
         Translator::default(),
         |mut translator, generation| {
-            let epoch = snapshot::read_epoch_metadata(&copy, generation)?;
+            let epoch = snapshot::read_epoch_metadata(copy, generation)?;
             let kind = epoch.kind().ok_or_else(|| {
                 format!(
                     "epoch generation {generation} was made by unknown remake kind {:?}",
@@ -564,7 +538,41 @@ fn prepare(
             Ok::<_, String>(translator)
         },
     );
+    let during = match (&next, &translator) {
+        // Sideways, to another epoch of our generation: the same paths.
+        _ if gap == 0 => Ok(Translator::default()),
+        (Some(_), Ok(translator)) => Ok(translator.clone()),
+        (Some(_), Err(reason)) => Err(format!("cannot be translated: {reason}")),
+        (None, _) => Err(format!(
+            "the metadata of the epoch after {previous} is missing"
+        )),
+    };
+
+    if began.is_empty() {
+        return Prepared::Ready {
+            metadata,
+            plan: Plan::Nothing,
+            during,
+            own_missing: 0,
+        };
+    }
+
+    own.push(old.get_actor().clone());
+    let clock = next
+        .as_ref()
+        .map_or_else(BTreeMap::new, |next| next.clock.clone());
+    let missing = delta::missing_from_snapshot(old, &clock, &own);
+    if missing.own == 0 {
+        return Prepared::Ready {
+            metadata,
+            plan: Plan::Nothing,
+            during,
+            own_missing: 0,
+        };
+    }
 
+    // Up to when the rejoin began; what came after is `during`'s.
+    let paths = delta::changed_paths(old, &missing.before, began);
     let plan = match (&next, translator) {
         // Rejoined sideways (`sideways`): whatever was written on our epoch, the swarm
         // went another way; nothing says where it would go.
@@ -590,9 +598,8 @@ fn prepare(
             let away_ms = now.ms_since(next.proposed_at);
             if next.policy.replay_window.covers(gap, away_ms) {
                 Plan::Replay {
-                    paths,
                     before: missing.before.clone(),
-                    after: old_heads.clone(),
+                    after: began.to_vec(),
                     translator,
                 }
             } else {
@@ -609,37 +616,27 @@ fn prepare(
     };
 
     Prepared::Ready {
-        old,
         metadata,
         plan,
+        during,
         own_missing: missing.own,
-        had_history: true,
     }
 }
 
-/// Restores the old document and epoch, and retries later.
+/// Restores the old document (with the writes it took meanwhile) and epoch, and
+/// retries later.
 async fn abandon<H: Host>(host: &H, rejoin: Rejoin, reason: String) {
     let Rejoin {
         previous,
         target,
         old,
-        was_ready,
         ..
     } = rejoin;
     warn!(target = %target, reason = %reason, "Rejoin failed; restoring the previous epoch");
 
-    if let Err(never) = host
-        .switch_epoch(previous, true, move |_live| Ok::<_, Infallible>(old))
-        .await
-    {
-        match never {}
-    }
+    host.abandon_rejoin(previous, old).await;
     // A node that was never ready must sync again before bootstrap may finish.
     host.reset_initial_sync();
-    if was_ready {
-        host.resume().await;
-    }
-    host.set_stale_view(None);
 
     let attempts = host
         .remake()
@@ -740,32 +737,65 @@ mod tests {
     #[test]
     fn replays_within_the_window() {
         let (old, swarm, target) = scenario(&json!({}));
+        let began = old.get_heads();
         match prepare(
-            swarm,
-            old,
+            &swarm,
+            &old,
+            &began,
             Epoch::GENESIS,
             target,
             Vec::new(),
             UnixMillis(0),
         ) {
             Prepared::Ready {
-                plan: Plan::Replay { paths, .. },
+                plan: Plan::Replay { before, after, .. },
                 own_missing,
                 ..
             } => {
                 assert_eq!(own_missing, 1);
-                assert_eq!(paths, vec![vec!["mine".to_owned()]]);
+                assert_eq!(
+                    delta::changed_paths(&old, &before, &after),
+                    vec![vec!["mine".to_owned()]]
+                );
             }
             _ => panic!("expected a replay"),
         }
     }
 
+    /// Writes made during the rejoin aren't the window's: those after `began` go
+    /// wherever the remakes since put them, even outside it.
+    #[test]
+    fn writes_during_the_rejoin_are_replayed_outside_the_window() {
+        let (old, swarm, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } }));
+        let began = old.get_heads();
+        let mut during = old.clone();
+        change(&mut during, |doc| doc.put(ROOT, "later", 1_i64).unwrap());
+        match prepare(
+            &swarm,
+            &during,
+            &began,
+            Epoch::GENESIS,
+            target,
+            Vec::new(),
+            UnixMillis(0),
+        ) {
+            Prepared::Ready {
+                plan: Plan::Quarantine { paths, .. },
+                during: Ok(_),
+                ..
+            } => assert_eq!(paths, vec![vec!["mine".to_owned()]]),
+            _ => panic!("expected a quarantine before, and a replay during the rejoin"),
+        }
+    }
+
     #[test]
     fn quarantines_outside_the_window() {
         let (old, swarm, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } }));
+        let began = old.get_heads();
         match prepare(
-            swarm,
-            old,
+            &swarm,
+            &old,
+            &began,
             Epoch::GENESIS,
             target,
             Vec::new(),
@@ -784,8 +814,9 @@ mod tests {
         let (_, swarm, target) = scenario(&json!({}));
         assert!(matches!(
             prepare(
-                swarm,
-                Document::new(),
+                &swarm,
+                &Document::new(),
+                &[],
                 Epoch::GENESIS,
                 target,
                 Vec::new(),
@@ -793,7 +824,7 @@ mod tests {
             ),
             Prepared::Ready {
                 plan: Plan::Nothing,
-                had_history: false,
+                own_missing: 0,
                 ..
             }
         ));
@@ -804,14 +835,15 @@ mod tests {
         let (old, _, target) = scenario(&json!({}));
         assert!(matches!(
             prepare(
-                Document::new(),
-                old,
+                &Document::new(),
+                &old,
+                &old.get_heads(),
                 Epoch::GENESIS,
                 target,
                 Vec::new(),
                 UnixMillis(0)
             ),
-            Prepared::NotYet(_)
+            Prepared::NotYet
         ));
     }
 }
diff --git a/src/remake/local.rs b/src/remake/local.rs
index e8fc9eb..ce133ec 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -16,7 +16,7 @@ use super::records::Retention;
 use super::snapshot::Snapshot;
 use crate::clock::UnixMillis;
 use crate::document::replica::Epoch;
-use crate::document::{ActorId, Document};
+use crate::document::{ActorId, ChangeHash, Document};
 use crate::locks::MutexExt;
 
 pub(crate) struct RemakeState {
@@ -87,8 +87,13 @@ pub(crate) struct CatchUp {
 pub(crate) struct Rejoin {
     pub(crate) previous: Epoch,
     pub(crate) target: Epoch,
-    /// The document of `previous`, until it is replayed, quarantined or restored.
-    pub(crate) old: Document,
+    /// The document of `previous`, until it is replayed, quarantined or restored; of a
+    /// node that was ready, kept by the node instead, where it takes its writes
+    /// meanwhile (`Host::begin_rejoin`).
+    pub(crate) old: Option<Document>,
+    /// That document's heads as the rejoin began: the changes after them it took
+    /// during the rejoin.
+    pub(crate) heads: Vec<ChangeHash>,
     pub(crate) started: Instant,
     /// The node served requests before; it does again once the rejoin is over.
     pub(crate) was_ready: bool,
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index 0d3d073..25ca7ab 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -159,16 +159,44 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
     /// Known (not forgotten) peers.
     fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send;
 
-    /// Stops serving requests (back to `Joining`, keeping the swarm ID), as while a
-    /// rejoin syncs a fresh document. Returns whether the node was serving.
-    fn suspend(&self) -> impl Future<Output = bool> + Send;
+    /// Starts a rejoin of `target`, under the document's write lock: the live document
+    /// becomes an empty one of `target`, written by `actor`, that takes no writes until
+    /// the rejoin ends. A ready node stops being ready (back to `Joining`, keeping the
+    /// swarm ID), and its document so far answers reads (marked stale) and takes its
+    /// writes meanwhile; a node that wasn't ready gets its document back. Afterwards
+    /// the node resets every peer's sync state and publishes the change.
+    fn begin_rejoin(
+        &self,
+        target: Epoch,
+        actor: crate::document::ActorId,
+    ) -> impl Future<Output = BegunRejoin> + Send;
 
-    /// Serves requests again after [`Host::suspend`].
-    fn resume(&self) -> impl Future<Output = ()> + Send;
+    /// A copy of the document before the rejoin, with the writes it took since, if the
+    /// node keeps it ([`Host::begin_rejoin`]: it was ready).
+    fn rejoin_document(&self) -> impl Future<Output = Option<Document>> + Send;
 
-    /// Serves reads from this document (read-only, marked stale) while the node is not
-    /// ready; `None` ends it.
-    fn set_stale_view(&self, view: Option<(Epoch, Document)>);
+    /// Ends a rejoin on its target, under the write lock: `f` writes into the live
+    /// document from the one before the rejoin (the node's, or `old` where it keeps
+    /// none), as many changes as it likes; then writes go to the live document again,
+    /// and a node that was ready is again. Returns what `f` returned and the document
+    /// before the rejoin.
+    fn finish_rejoin<T, F>(
+        &self,
+        old: Option<Document>,
+        f: F,
+    ) -> impl Future<Output = (T, Document)> + Send
+    where
+        F: FnOnce(&mut Document, &Document) -> T + Send + 'static,
+        T: Send + 'static;
+
+    /// Ends a rejoin by going back, under the write lock: the document before it (the
+    /// node's, with the writes it took meanwhile, or `old`) and `previous` are live
+    /// again, and a node that was ready is again. Afterwards as [`Host::begin_rejoin`].
+    fn abandon_rejoin(
+        &self,
+        previous: Epoch,
+        old: Option<Document>,
+    ) -> impl Future<Output = ()> + Send;
 
     /// Whether the document caught up with a peer since the last reset.
     fn initial_sync_done(&self) -> bool;
@@ -185,9 +213,9 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
         F: FnOnce(&Document) -> T + Send + 'static,
         T: Send + 'static;
 
-    /// One committed change (rolled back on error), published and replicated. Refused
-    /// (`StoreError::Rejoining`) while the document isn't writable: from a rejoin's
-    /// swap until [`Host::accept_writes`].
+    /// One committed change (rolled back on error), published and replicated. While a
+    /// rejoin runs, into the document before it, or refused (`StoreError::Rejoining`)
+    /// by a node that keeps none (`AppState::mutate_doc`).
     fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send
     where
         F: FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static,
@@ -197,23 +225,6 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
     /// What a replay does with writes a newer one superseded (`--superseded-writes`).
     fn superseded(&self) -> Superseded;
 
-    /// Changes `f` commits itself into the live document (as many as it likes, each
-    /// with its own time), under the write lock, then published and replicated.
-    /// Refused while the document isn't writable, as [`Host::commit_doc`].
-    fn commit_changes<T, F>(&self, f: F) -> impl Future<Output = Result<T, StoreError>> + Send
-    where
-        F: FnOnce(&mut Document) -> T + Send + 'static,
-        T: Send + 'static;
-
-    /// Lets writes in again, after a rejoin swapped in a document that took none
-    /// ([`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(
@@ -222,13 +233,11 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
     ) -> impl Future<Output = Result<(), String>> + Send;
 
     /// Replaces the document and its epoch atomically. Under the write lock, `f` gets
-    /// the live document and returns the new one, which takes writes if `writable`
-    /// (a rejoin's fresh document doesn't). Afterwards the node resets every peer's
-    /// sync state and publishes the change. Returns the old document.
+    /// the live document and returns the new one. Afterwards the node resets every
+    /// peer's sync state and publishes the change. Returns the old document.
     fn switch_epoch<E, F>(
         &self,
         epoch: Epoch,
-        writable: bool,
         f: F,
     ) -> impl Future<Output = Result<Document, E>> + Send
     where
@@ -236,6 +245,17 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
         E: Send + 'static;
 }
 
+/// What [`Host::begin_rejoin`] left the rejoin with.
+pub(crate) struct BegunRejoin {
+    /// The node was ready, and is again once the rejoin ends.
+    pub(crate) was_ready: bool,
+    /// The heads of the document before the rejoin as it began: what changes after
+    /// them, it took during the rejoin.
+    pub(crate) heads: Vec<crate::document::ChangeHash>,
+    /// The document before the rejoin, where the node doesn't keep it (it wasn't ready).
+    pub(crate) old: Option<Document>,
+}
+
 #[derive(Clone, Debug)]
 pub(crate) struct PeerSummary {
     pub(crate) id: String,
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index 7d6611d..4c72b39 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -306,7 +306,7 @@ pub(super) async fn swap_in<H: Host>(
     let late_stats = std::sync::Arc::new(std::sync::Mutex::new(ReplayStats::default()));
     let late_out = std::sync::Arc::clone(&late_stats);
     let old = host
-        .switch_epoch(target, true, move |live| {
+        .switch_epoch(target, move |live| {
             // Only one worker switches, but be sure the copy still describes the live document.
             if !crate::document::changes::missing(live, &copy_heads).is_empty() {
                 return Err("the document was replaced while preparing the switch".to_owned());