replace snapshots by quarantine & replay

1 files changed, 28 insertions, 11 deletions

diff --git a/src/compaction/mod.rs b/src/compaction/mod.rs
index e97d9c8..4685b86 100644
--- a/src/compaction/mod.rs
+++ b/src/compaction/mod.rs
@@ -4,7 +4,9 @@
 //! * `records`: the replicated compaction records under `_continuum/compactions/<id>`.
 //! * `coordinator`: proposal, settle period and decision on the node that was asked.
 //! * `switch`: switching this node to a committed epoch (worker).
-//! * `catchup`: fetching a newer epoch from a peer (late, new or failed nodes).
+//! * `catchup`: rejoining a newer epoch from peers (late, new or failed nodes), then
+//!   replaying or quarantining this node's own missing changes.
+//! * `quarantine`: inspecting, replaying or discarding quarantined changes.
 //! * `delta`: replaying this node's changes since the base into the new epoch.
 //! * `local`: per-node in-memory state (prebuilt snapshots, switch status, archive).
 //! * `api`: `/admin/compaction(s)` handlers.
@@ -17,11 +19,11 @@ mod catchup;
 mod coordinator;
 mod delta;
 mod local;
+mod quarantine;
 mod records;
 pub(crate) mod snapshot;
 mod switch;
 
-pub(crate) use catchup::{receive_snapshot_reply, serve_snapshot};
 pub(crate) use local::Compaction;
 pub(crate) use switch::run_worker;
 
@@ -38,7 +40,6 @@ use tokio::sync::broadcast;
 
 use crate::replica::Epoch;
 use crate::sync::ChangeSet;
-use crate::transport::ParsedMessage;
 
 /// What compaction needs from the node. Implemented by `AppState`.
 ///
@@ -61,15 +62,18 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
     /// Known (not forgotten) peers.
     fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send;
 
-    /// True if `peer` completed the handshake as a member of our cluster.
-    fn is_cluster_member(&self, peer: &str) -> impl Future<Output = bool> + Send;
+    /// Stops serving requests (back to `Joining`, keeping the cluster ID), as while a
+    /// rejoin syncs a fresh document. Returns whether the node was serving.
+    fn suspend(&self) -> impl Future<Output = bool> + Send;
 
-    /// Sends a message to a peer over the transport.
-    fn send_message(
-        &self,
-        peer: &str,
-        message: ParsedMessage,
-    ) -> impl Future<Output = anyhow::Result<()>> + Send;
+    /// Serves requests again after [`Host::suspend`].
+    fn resume(&self) -> impl Future<Output = ()> + Send;
+
+    /// Whether the document caught up with a peer since the last reset.
+    fn initial_sync_done(&self) -> bool;
+
+    /// Makes the next sync count as initial again.
+    fn reset_initial_sync(&self);
 
     /// Store paths touched by each published document change.
     fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>>;
@@ -130,4 +134,17 @@ pub(crate) fn router<H: Host>() -> Router<H> {
             get(api::get::<H>).delete(api::abort::<H>),
         )
         .route("/admin/epochs", get(api::epochs::<H>))
+        .route("/admin/quarantine", get(quarantine::list::<H>))
+        .route(
+            "/admin/quarantine/{id}",
+            get(quarantine::get::<H>).delete(quarantine::discard::<H>),
+        )
+        .route(
+            "/admin/quarantine/{id}/document",
+            get(quarantine::document::<H>),
+        )
+        .route(
+            "/admin/quarantine/{id}/replay",
+            post(quarantine::replay::<H>),
+        )
 }