add compaction coordinator

5 files changed, 163 insertions, 5 deletions

diff --git a/src/api/mod.rs b/src/api/mod.rs
index cb0d701..98400c7 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -35,7 +35,8 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) ->
         )
         .route("/stream/{*key}", get(store::stream_store_entry))
         .route("/peers", get(peers::list_peers))
-        .route("/peers/{peer}", delete(peers::delete_peer));
+        .route("/peers/{peer}", delete(peers::delete_peer))
+        .merge(crate::compaction::router::<AppState>());
 
     if debug_api {
         warn!("Debug API enabled");
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;
diff --git a/src/state.rs b/src/state.rs
index c6d3d86..8e595af 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -1,6 +1,7 @@
 //! Shared node state: the document, lifecycle, per-peer state and timing.
 
 use std::collections::HashMap;
+use std::future::Future;
 use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
 use std::time::Duration;
@@ -11,6 +12,7 @@ use tokio::sync::{broadcast, mpsc, watch, Mutex, RwLock};
 use tokio::time::Instant;
 use tracing::{debug, error, warn};
 
+use crate::compaction::{Host, PeerSummary};
 use crate::firewall::Firewall;
 use crate::replica::{Epoch, Replica};
 use crate::sync::{ChangeSet, SyncEvent};
@@ -499,3 +501,75 @@ pub(crate) fn unix_millis() -> u64 {
         .map(|d| d.as_millis() as u64)
         .unwrap_or(0)
 }
+
+/// The node as seen by the compaction module.
+impl Host for AppState {
+    fn node_id(&self) -> String {
+        self.local_id().to_owned()
+    }
+
+    fn current_epoch(&self) -> Epoch {
+        self.local_epoch()
+    }
+
+    fn dormant_after(&self) -> Duration {
+        self.timing.dormant_after
+    }
+
+    fn is_node_ready(&self) -> impl Future<Output = bool> + Send {
+        let state = self.clone();
+        async move { state.is_ready().await }
+    }
+
+    fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send {
+        let state = self.clone();
+        async move {
+            let peers: Vec<(String, Arc<Mutex<PeerState>>)> = state
+                .peer_states
+                .read()
+                .await
+                .iter()
+                .map(|(id, peer)| (id.clone(), Arc::clone(peer)))
+                .collect();
+
+            let mut summaries = Vec::with_capacity(peers.len());
+            for (id, peer) in peers {
+                let active = peer.lock().await.liveness == Liveness::Active;
+                summaries.push(PeerSummary { id, active });
+            }
+            summaries
+        }
+    }
+
+    fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>> {
+        self.change_events.subscribe()
+    }
+
+    fn read_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send
+    where
+        F: FnOnce(&AutoCommit) -> T + Send + 'static,
+        T: Send + 'static,
+    {
+        let state = self.clone();
+        async move { state.render(f, |value| value).await }
+    }
+
+    fn update_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send
+    where
+        F: FnOnce(&mut AutoCommit) -> T + Send + 'static,
+        T: Send + 'static,
+    {
+        let state = self.clone();
+        async move { state.write_doc(f).await }
+    }
+
+    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,
+    {
+        let state = self.clone();
+        async move { state.mutate_doc(f).await }
+    }
+}
diff --git a/src/store.rs b/src/store.rs
index 513728e..1f1398e 100644
--- a/src/store.rs
+++ b/src/store.rs
@@ -314,7 +314,6 @@ pub fn delete_path(doc: &mut AutoCommit, path: &[String], prune: bool) -> Result
 ///
 /// `skip` sees the path of every map entry and list element (list indices as decimal
 /// strings); returning true leaves that subtree out.
-#[cfg_attr(not(test), allow(dead_code))] // used by compaction from step 4 on
 pub(crate) fn materialize_into(
     src: &AutoCommit,
     dst: &mut AutoCommit,