add compaction coordinator

1 files changed, 74 insertions, 0 deletions

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 }
+    }
+}