From 0cdd5075c88d938117c5dbddc22a2331a8816c34 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 21 Oct 2026 03:56:31 +0200 Subject: rename compaction to remake --- src/remake/mod.rs | 150 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 150 insertions(+) create mode 100644 src/remake/mod.rs (limited to 'src/remake/mod.rs') diff --git a/src/remake/mod.rs b/src/remake/mod.rs new file mode 100644 index 0000000..974acde --- /dev/null +++ b/src/remake/mod.rs @@ -0,0 +1,150 @@ +//! 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/`. +//! * `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::{ + Router, + routing::{get, post}, +}; +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 cluster 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; + + /// 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, + /// 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::)) + .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::), + ) +} -- cgit v1.3.1