1 files changed, 0 insertions, 183 deletions
diff --git a/src/compaction/coordinator.rs b/src/compaction/coordinator.rs
deleted file mode 100644
index 1b41faf..0000000
--- a/src/compaction/coordinator.rs
+++ /dev/null
@@ -1,183 +0,0 @@
-//! 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
-}