//! Remakes: coordinated moves of the whole swarm to a new document, a new epoch. //! Compaction is one: the new document has the old one's content without its history. //! //! * `kind`: what differs between kinds of remake (building the new document, and //! where writes made after the base go): `compact` and `replace`. //! * `snapshot`: deterministic build of the next epoch's initial document. //! * `records`: the replicated remake records under `_continuum/remakes/`. //! * `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 kind; 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 + Send; /// Known (not forgotten) peers. fn peer_summaries(&self) -> impl Future> + 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 + Send; /// Serves requests again after [`Host::suspend`]. fn resume(&self) -> impl Future + 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>; /// Reads the document on the blocking pool. fn read_doc(&self, f: F) -> impl Future + 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(&self, f: F) -> impl Future + 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(&self, f: F) -> impl Future> + Send where F: FnOnce(&mut AutoCommit) -> Result + 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( &self, epoch: Epoch, f: F, ) -> impl Future> + Send where F: FnOnce(&mut AutoCommit) -> Result + 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, } /// Admin routes. Merged into the authenticated part of the node's router. pub(crate) fn router() -> Router { Router::new() .route( "/admin/remakes", get(api::list::) .post(api::start::) .delete(api::abort_current::), ) // The original endpoint, kept as a shortcut for a compacting remake. .route("/admin/compaction", post(api::start_compaction::)) .route( "/admin/remakes/{id}", get(api::get::).delete(api::abort::), ) .route("/admin/epochs", get(api::epochs::)) .route("/admin/quarantine", get(quarantine::list::)) .route( "/admin/quarantine/{id}", get(quarantine::get::).delete(quarantine::discard::), ) .route( "/admin/quarantine/{id}/document", get(quarantine::document::), ) .route( "/admin/quarantine/{id}/replay", post(quarantine::replay::), ) }