diff options
Diffstat (limited to 'src/compaction')
| -rw-r--r-- | src/compaction/mod.rs | 88 | ||||
| -rw-r--r-- | src/compaction/snapshot.rs | 2 |
2 files changed, 87 insertions, 3 deletions
diff --git a/src/compaction/mod.rs b/src/compaction/mod.rs index 703764b..5608eae 100644 --- a/src/compaction/mod.rs +++ b/src/compaction/mod.rs @@ -1,6 +1,90 @@ //! Compaction: replacing the document with a snapshot in a new epoch. //! -//! Step 3 provides the deterministic snapshot build. The protocol (proposal, settle, -//! commit, switch) and the admin API follow in step 4. +//! * `snapshot`: deterministic snapshot build (step 3). +//! * `records`: the replicated compaction records under `_continuum/compactions/<id>`. +//! * `coordinator`: proposal, settle period and decision on the node that was asked. +//! * `api`: `/admin/compaction(s)` handlers. +//! +//! Step 4 decides compactions but does not switch epochs yet; a committed record only +//! names the new epoch. +//! +//! The rest of the node is reached exclusively through [`Host`], so this module never +//! depends on `AppState`, peers or sync internals. +mod api; +mod coordinator; +mod records; pub(crate) mod snapshot; + +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; + +/// 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 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; + + /// 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; +} + +#[derive(Clone, Debug)] +pub(crate) struct PeerSummary { + pub(crate) id: String, + /// Not dormant. + pub(crate) active: bool, +} + +/// Admin routes. Merged into the authenticated part of the node's router. +pub(crate) fn router<H: Host>() -> Router<H> { + Router::new() + .route("/admin/compaction", post(api::start::<H>)) + .route("/admin/compactions", get(api::list::<H>)) + .route( + "/admin/compactions/{id}", + get(api::get::<H>).delete(api::abort::<H>), + ) +} diff --git a/src/compaction/snapshot.rs b/src/compaction/snapshot.rs index 9dd54c0..83ff28a 100644 --- a/src/compaction/snapshot.rs +++ b/src/compaction/snapshot.rs @@ -18,7 +18,7 @@ //! at the snapshot change itself, so edits made afterwards through the store API can't //! change what the protocol sees. -// Wired into the compaction protocol in step 4. +// `Snapshot::doc` and `Snapshot::clock` are consumed by the epoch switch (step 5). #![cfg_attr(not(test), allow(dead_code))] use std::collections::BTreeMap; |
