1 files changed, 158 insertions, 0 deletions
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
new file mode 100644
index 0000000..8cca57c
--- /dev/null
+++ b/src/remake/mod.rs
@@ -0,0 +1,158 @@
+//! Remakes: coordinated moves of the whole cluster to a new document, a new epoch.
+//! Compaction is one: the new document has the old one's content without its history.
+//!
+//! * `snapshot`: deterministic build of the next epoch's initial document.
+//! * `records`: the replicated remake records under `_continuum/remakes/<id>`.
+//! * `coordinator`: proposal, settle period and decision on the node that was asked.
+//! * `switch`: switching this node to a committed epoch (worker).
+//! * `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/remakes` handlers.
+//!
+//! The rest of the node is reached exclusively through [`Host`], so this module never
+//! depends on `AppState`, peers or sync internals.
+
+mod api;
+mod catchup;
+mod coordinator;
+mod delta;
+mod local;
+mod quarantine;
+mod records;
+pub(crate) mod snapshot;
+mod switch;
+
+pub(crate) use local::RemakeState;
+pub(crate) use switch::run_worker;
+
+use std::future::Future;
+use std::sync::Arc;
+use std::time::Duration;
+
+use automerge::AutoCommit;
+use axum::{
+    routing::{get, post},
+    Router,
+};
+use tokio::sync::broadcast;
+
+use crate::document::replica::Epoch;
+use crate::swarm::sync::ChangeSet;
+
+/// What remake needs from the node. Implemented by `AppState`.
+///
+/// Used generically (`H: Host`), never as `dyn Host`, so the generic methods are fine.
+pub(crate) trait Host: Clone + Send + Sync + 'static {
+    /// This node's in-memory remake state.
+    fn remake(&self) -> &RemakeState;
+
+    /// This node's peer ID.
+    fn node_id(&self) -> String;
+
+    /// The epoch of the local replica.
+    fn current_epoch(&self) -> Epoch;
+
+    /// Lower bound and default basis for the settle period.
+    fn dormant_after(&self) -> Duration;
+
+    fn is_node_ready(&self) -> impl Future<Output = bool> + Send;
+
+    /// Known (not forgotten) peers.
+    fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + 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;
+
+    /// Serves requests again after [`Host::suspend`].
+    fn resume(&self) -> impl Future<Output = ()> + 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, AutoCommit)>);
+
+    /// 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>>;
+
+    /// Reads the document on the blocking pool.
+    fn read_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send
+    where
+        F: FnOnce(&AutoCommit) -> T + Send + 'static,
+        T: Send + 'static;
+
+    /// Mutable access to the document without publishing a change (e.g. forks).
+    fn update_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send
+    where
+        F: FnOnce(&mut AutoCommit) -> T + Send + 'static,
+        T: Send + 'static;
+
+    /// One committed change (rolled back on error), published and replicated.
+    fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send
+    where
+        F: FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static,
+        T: Send + 'static,
+        E: Send + 'static;
+
+    /// Replaces the document and its epoch atomically. Under the write lock, `f` gets
+    /// 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,
+        f: F,
+    ) -> impl Future<Output = Result<AutoCommit, E>> + Send
+    where
+        F: FnOnce(&mut AutoCommit) -> Result<AutoCommit, E> + Send + 'static,
+        E: Send + 'static;
+}
+
+#[derive(Clone, Debug)]
+pub(crate) struct PeerSummary {
+    pub(crate) id: String,
+    /// Not dormant.
+    pub(crate) active: bool,
+    /// Handshake completed.
+    pub(crate) established: bool,
+    /// The epoch the peer last advertised.
+    pub(crate) epoch: Option<Epoch>,
+}
+
+/// Admin routes. Merged into the authenticated part of the node's router.
+pub(crate) fn router<H: Host>() -> Router<H> {
+    Router::new()
+        .route(
+            "/admin/remakes",
+            get(api::list::<H>)
+                .post(api::start::<H>)
+                .delete(api::abort_current::<H>),
+        )
+        // The original endpoint, kept as a shortcut for a compacting remake.
+        .route("/admin/compaction", post(api::start::<H>))
+        .route(
+            "/admin/remakes/{id}",
+            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>),
+        )
+}