//! Compaction: replacing the document with a snapshot in a new epoch. //! //! * `snapshot`: deterministic snapshot build (step 3). //! * `records`: the replicated compaction records under `_continuum/compactions/`. //! * `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). //! * `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. //! //! 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 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; 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::replica::Epoch; use crate::sync::ChangeSet; use crate::transport::ParsedMessage; /// What compaction 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 compaction state. fn compaction(&self) -> &Compaction; /// 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; /// True if `peer` completed the handshake as a member of our cluster. fn is_cluster_member(&self, peer: &str) -> impl Future + Send; /// Sends a message to a peer over the transport. fn send_message( &self, peer: &str, message: ParsedMessage, ) -> impl Future> + Send; /// 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/compaction", post(api::start::).delete(api::abort_current::), ) .route("/admin/compactions", get(api::list::)) .route( "/admin/compactions/{id}", get(api::get::).delete(api::abort::), ) .route("/admin/epochs", get(api::epochs::)) }