1 files changed, 183 insertions, 0 deletions
diff --git a/src/compaction/coordinator.rs b/src/compaction/coordinator.rs
new file mode 100644
index 0000000..1b41faf
--- /dev/null
+++ b/src/compaction/coordinator.rs
@@ -0,0 +1,183 @@
+//! The coordinator: the node that received the proposal watches the cluster during the
+//! settle period and then writes the decision.
+//!
+//! It aborts if, at any point before deciding:
+//! * an operator requested an abort,
+//! * a competing proposal for the same epoch appeared (both coordinators abort),
+//! * a peer it considered went dormant (a partition may hide a competitor),
+//! * the local epoch changed, or
+//! * the decision would come later than the settle period plus the grace margin
+//!   (other nodes report the proposal as expired by then).
+//!
+//! Otherwise it builds the snapshot at the base heads and commits, recording the new
+//! epoch so every node can verify its own build against it.
+//!
+//! The task lives only in memory. If the coordinator dies, nobody decides and the
+//! proposal expires.
+
+use std::collections::HashMap;
+use std::time::Duration;
+
+use tracing::{info, warn};
+
+use super::records::{self, Decision, Record};
+use super::snapshot;
+use super::Host;
+use crate::store::StoreError;
+
+const TICK: Duration = Duration::from_secs(1);
+
+pub(crate) async fn run<H: Host>(host: H, id: String) {
+    info!(compaction = %id, "Coordinating compaction");
+
+    loop {
+        tokio::time::sleep(TICK).await;
+
+        match step(&host, &id).await {
+            Ok(true) => return,
+            Ok(false) => {}
+            Err(error) => {
+                warn!(compaction = %id, error = %format!("{error:#}"), "Coordinator step failed; retrying");
+            }
+        }
+    }
+}
+
+/// One evaluation. Returns true once the coordinator is done.
+async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
+    let wanted = id.to_owned();
+    let (record, all) = host
+        .read_doc(move |doc| (records::load(doc, &wanted), records::load_all(doc)))
+        .await;
+
+    let Some(record) = record else {
+        warn!(compaction = %id, "Compaction record vanished; stopping coordinator");
+        return Ok(true);
+    };
+    if record.decision.is_some() {
+        return Ok(true);
+    }
+
+    let now = records::unix_millis();
+    let grace = host.dormant_after().as_millis() as u64;
+
+    if let Some(reason) = abort_reason(host, id, &record, &all, now, grace).await {
+        decide_abort(host, id, reason).await?;
+        return Ok(true);
+    }
+
+    if now < record.proposal.settle_until {
+        return Ok(false);
+    }
+
+    commit(host, id, &record).await?;
+    Ok(true)
+}
+
+async fn abort_reason<H: Host>(
+    host: &H,
+    id: &str,
+    record: &Record,
+    all: &[(String, Record)],
+    now: u64,
+    grace: u64,
+) -> Option<String> {
+    let proposal = &record.proposal;
+
+    if let Some(request) = &record.abort_requested {
+        return Some(format!("aborted by operator via {}", request.by));
+    }
+
+    if records::EpochRecord::from(host.current_epoch()) != proposal.from_epoch {
+        return Some("local epoch changed during settle".to_owned());
+    }
+
+    if let Some((other, _)) = all.iter().find(|(other, candidate)| {
+        other != id
+            && candidate.proposal.from_epoch == proposal.from_epoch
+            && candidate.phase(now, grace).is_open()
+    }) {
+        return Some(format!("competing compaction {other} for the same epoch"));
+    }
+
+    // Peers forgotten in the meantime are ignored: an operator decided they are gone.
+    let peers: HashMap<String, bool> = host
+        .peer_summaries()
+        .await
+        .into_iter()
+        .map(|peer| (peer.id, peer.active))
+        .collect();
+    let dormant: Vec<&str> = proposal
+        .peers_considered
+        .iter()
+        .filter(|peer| peers.get(peer.as_str()) == Some(&false))
+        .map(String::as_str)
+        .collect();
+    if !dormant.is_empty() {
+        return Some(format!(
+            "peers went dormant during settle: {}",
+            dormant.join(", ")
+        ));
+    }
+
+    if now > proposal.settle_until.saturating_add(grace) {
+        return Some("decision deadline passed".to_owned());
+    }
+
+    None
+}
+
+async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), StoreError> {
+    let Some(request) = record.snapshot_request(id) else {
+        return decide_abort(
+            host,
+            id,
+            "unreadable epoch or base heads in proposal".into(),
+        )
+        .await;
+    };
+
+    // Clone under the read lock, build without any lock.
+    let copy = host.read_doc(|doc| doc.clone()).await;
+    let mut snapshot = match snapshot::build_detached(copy, request).await {
+        Ok((snapshot, _copy)) => snapshot,
+        Err(error) => {
+            return decide_abort(host, id, format!("snapshot build failed: {error}")).await;
+        }
+    };
+
+    let to_epoch = snapshot.epoch;
+    let snapshot_bytes = snapshot.doc.save().len() as u64;
+
+    let decision = Decision::Committed {
+        to_epoch: to_epoch.into(),
+        snapshot_bytes,
+        decided_at: records::unix_millis(),
+    };
+    write_decision(host, id, decision).await?;
+
+    // The local switch reuses it instead of building again.
+    host.compaction().store_prebuilt(id, snapshot);
+
+    info!(compaction = %id, to_epoch = %to_epoch, snapshot_bytes, "Compaction committed");
+    Ok(())
+}
+
+async fn decide_abort<H: Host>(host: &H, id: &str, reason: String) -> Result<(), StoreError> {
+    warn!(compaction = %id, reason = %reason, "Compaction aborted");
+    write_decision(
+        host,
+        id,
+        Decision::Aborted {
+            reason,
+            decided_at: records::unix_millis(),
+        },
+    )
+    .await
+}
+
+async fn write_decision<H: Host>(host: &H, id: &str, decision: Decision) -> Result<(), StoreError> {
+    let id = id.to_owned();
+    host.commit_doc(move |doc| records::write_decision(doc, &id, &decision))
+        .await
+}