document: allow writes during rejoin

1 files changed, 52 insertions, 32 deletions

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,