1 files changed, 369 insertions, 0 deletions
diff --git a/src/compaction/api.rs b/src/compaction/api.rs
new file mode 100644
index 0000000..b8c55db
--- /dev/null
+++ b/src/compaction/api.rs
@@ -0,0 +1,369 @@
+//! `/admin/compaction(s)` handlers.
+//!
+//! * `POST /admin/compaction` proposes a compaction of the current epoch. The receiving
+//!   node becomes its coordinator. `202` with the report, or with `?wait=true` the final
+//!   report once decided. A dropped request does not abort anything.
+//! * `GET /admin/compactions` lists all compactions.
+//! * `GET /admin/compactions/{id}` reports one; `?wait=true` long-polls until decided
+//!   (bounded by the proposal's own expiry).
+//! * `GET /admin/epochs` shows the local epoch and the archived previous ones.
+//! * `DELETE /admin/compactions/{id}` asks the coordinator to abort. Only possible while
+//!   proposed; `409` once committed (a switch can't be undone) or already over.
+//!
+//! Request body for POST (all optional):
+//! `{ "replayWindow": {"epochs": 1, "seconds": 604800},
+//!    "archiveRetention": {"epochs": 2, "seconds": 2592000},
+//!    "settleSecs": 480 }`
+
+use std::time::Duration;
+
+use axum::{
+    body::Bytes,
+    extract::{Path, Query, State},
+    http::{header::LOCATION, StatusCode},
+    response::{IntoResponse, Response},
+    Json,
+};
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value as JsonValue};
+use tokio::sync::broadcast;
+use tracing::{error, info};
+use uuid::Uuid;
+
+use super::coordinator;
+use super::local::LocalStatus;
+use super::records::{
+    self, unix_millis, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention,
+};
+use super::Host;
+use crate::store::StoreError;
+
+#[derive(Debug, Default, Deserialize)]
+pub(crate) struct WaitQuery {
+    #[serde(default)]
+    wait: bool,
+}
+
+#[derive(Debug, Default, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+struct StartRequest {
+    replay_window: Option<Retention>,
+    archive_retention: Option<Retention>,
+    settle_secs: Option<u64>,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct Report {
+    id: String,
+    phase: Phase,
+    #[serde(flatten)]
+    record: Record,
+    /// What happened to this compaction on the node answering the request.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    local: Option<LocalStatus>,
+}
+
+#[derive(Debug, thiserror::Error)]
+enum StartError {
+    #[error("compaction {0} is already open for this epoch")]
+    Open(String),
+    #[error(transparent)]
+    Store(#[from] StoreError),
+}
+
+fn error_response(
+    status: StatusCode,
+    message: impl Into<String>,
+    details: Option<JsonValue>,
+) -> Response {
+    let mut body = json!({ "error": message.into() });
+    if let (Some(JsonValue::Object(extra)), Some(object)) = (details, body.as_object_mut()) {
+        object.extend(extra);
+    }
+    (status, Json(body)).into_response()
+}
+
+fn grace<H: Host>(host: &H) -> u64 {
+    host.dormant_after().as_millis() as u64
+}
+
+fn report<H: Host>(host: &H, id: String, record: Record) -> Report {
+    Report {
+        phase: record.phase(unix_millis(), grace(host)),
+        local: host.compaction().status(&id),
+        id,
+        record,
+    }
+}
+
+async fn load_report<H: Host>(host: &H, id: &str) -> Option<Report> {
+    let wanted = id.to_owned();
+    let record = host
+        .read_doc(move |doc| records::load(doc, &wanted))
+        .await?;
+    Some(report(host, id.to_owned(), record))
+}
+
+/// Waits until the compaction leaves `proposed`. Bounded by the proposal's own expiry,
+/// at which point it is reported as `expired`.
+async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> {
+    let mut changes = host.subscribe_changes();
+    loop {
+        let report = load_report(host, id).await?;
+        if report.phase.is_final() {
+            return Some(report);
+        }
+        // Re-read on every published change, and at least once a second: expiry is a
+        // matter of time, not of document changes.
+        if let Ok(Err(broadcast::error::RecvError::Closed)) =
+            tokio::time::timeout(Duration::from_secs(1), changes.recv()).await
+        {
+            tokio::time::sleep(Duration::from_secs(1)).await;
+        }
+    }
+}
+
+pub(crate) async fn start<H: Host>(
+    State(host): State<H>,
+    Query(query): Query<WaitQuery>,
+    body: Bytes,
+) -> Response {
+    if !host.is_node_ready().await {
+        return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None);
+    }
+
+    let request: StartRequest = if body.is_empty() {
+        StartRequest::default()
+    } else {
+        match serde_json::from_slice(&body) {
+            Ok(request) => request,
+            Err(error) => {
+                return error_response(
+                    StatusCode::BAD_REQUEST,
+                    format!("invalid request: {error}"),
+                    None,
+                )
+            }
+        }
+    };
+
+    let dormant_after = host.dormant_after();
+    let settle = request
+        .settle_secs
+        .map(Duration::from_secs)
+        .unwrap_or(dormant_after * 2);
+    if settle < dormant_after {
+        return error_response(
+            StatusCode::BAD_REQUEST,
+            format!(
+                "settleSecs must be at least {} (the dormancy threshold), so a partition is detected before deciding",
+                dormant_after.as_secs()
+            ),
+            None,
+        );
+    }
+
+    // Fail fast instead of proposing something that will abort anyway.
+    let peers = host.peer_summaries().await;
+    let dormant: Vec<String> = peers
+        .iter()
+        .filter(|peer| !peer.active)
+        .map(|peer| peer.id.clone())
+        .collect();
+    if !dormant.is_empty() {
+        return error_response(
+            StatusCode::CONFLICT,
+            "dormant peers block compaction; forget them first (DELETE /peers/{pk})",
+            Some(json!({ "dormantPeers": dormant })),
+        );
+    }
+
+    let mut peers_considered: Vec<String> = peers.into_iter().map(|peer| peer.id).collect();
+    peers_considered.sort();
+
+    let now = unix_millis();
+    let settle_ms = settle.as_millis() as u64;
+    let id = Uuid::new_v4().to_string();
+    let grace = grace(&host);
+
+    let mut proposal = Proposal {
+        coordinator: host.node_id(),
+        from_epoch: host.current_epoch().into(),
+        base_heads: Vec::new(),
+        policy: Policy {
+            replay_window: request
+                .replay_window
+                .unwrap_or_else(Policy::default_replay_window),
+            archive_retention: request
+                .archive_retention
+                .unwrap_or_else(Policy::default_archive_retention),
+        },
+        settle_ms,
+        proposed_at: now,
+        settle_until: now + settle_ms,
+        peers_considered,
+    };
+
+    let record_id = id.clone();
+    let written = host
+        .commit_doc(move |doc| -> Result<Record, StartError> {
+            if let Some((open, _)) = records::load_all(doc).into_iter().find(|(_, record)| {
+                record.proposal.from_epoch == proposal.from_epoch
+                    && record.phase(now, grace).is_open()
+            }) {
+                return Err(StartError::Open(open));
+            }
+
+            // The base is everything before the proposal; the proposal itself is not in it.
+            let mut heads: Vec<String> = doc.get_heads().iter().map(|h| h.to_string()).collect();
+            heads.sort();
+            proposal.base_heads = heads;
+
+            records::write_proposal(doc, &record_id, &proposal)?;
+            Ok(Record {
+                proposal,
+                decision: None,
+                abort_requested: None,
+            })
+        })
+        .await;
+
+    let record = match written {
+        Ok(record) => record,
+        Err(StartError::Open(open)) => {
+            return error_response(
+                StatusCode::CONFLICT,
+                "another compaction of this epoch is still open",
+                Some(json!({ "compaction": open })),
+            );
+        }
+        Err(StartError::Store(error)) => {
+            error!(error = %format!("{error:#}"), "Failed to write compaction proposal");
+            return error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "failed to write the proposal",
+                None,
+            );
+        }
+    };
+
+    info!(
+        compaction = %id,
+        settle_secs = settle.as_secs(),
+        "Compaction proposed"
+    );
+    tokio::spawn(coordinator::run(host.clone(), id.clone()));
+
+    if query.wait {
+        return match wait_until_final(&host, &id).await {
+            Some(report) => Json(report).into_response(),
+            None => error_response(StatusCode::NOT_FOUND, "compaction record vanished", None),
+        };
+    }
+
+    let location = format!("/admin/compactions/{id}");
+    (
+        StatusCode::ACCEPTED,
+        [(LOCATION, location)],
+        Json(report(&host, id, record)),
+    )
+        .into_response()
+}
+
+pub(crate) async fn list<H: Host>(State(host): State<H>) -> Response {
+    let all = host.read_doc(records::load_all).await;
+    let mut reports: Vec<Report> = all
+        .into_iter()
+        .map(|(id, record)| report(&host, id, record))
+        .collect();
+    reports.sort_by_key(|report| report.record.proposal.proposed_at);
+    Json(reports).into_response()
+}
+
+pub(crate) async fn get<H: Host>(
+    State(host): State<H>,
+    Path(id): Path<String>,
+    Query(query): Query<WaitQuery>,
+) -> Response {
+    let found = if query.wait {
+        wait_until_final(&host, &id).await
+    } else {
+        load_report(&host, &id).await
+    };
+    match found {
+        Some(report) => Json(report).into_response(),
+        None => error_response(StatusCode::NOT_FOUND, "no such compaction", None),
+    }
+}
+
+pub(crate) async fn abort<H: Host>(
+    State(host): State<H>,
+    Path(id): Path<String>,
+    Query(query): Query<WaitQuery>,
+) -> Response {
+    let Some(current) = load_report(&host, &id).await else {
+        return error_response(StatusCode::NOT_FOUND, "no such compaction", None);
+    };
+
+    match current.phase {
+        Phase::Proposed => {}
+        Phase::Committed => {
+            return error_response(
+                StatusCode::CONFLICT,
+                "already committed; the switch cannot be undone",
+                Some(json!({ "compaction": current })),
+            )
+        }
+        Phase::Aborted | Phase::Expired => {
+            return error_response(
+                StatusCode::CONFLICT,
+                "already over",
+                Some(json!({ "compaction": current })),
+            )
+        }
+    }
+
+    // The coordinator applies the request on its next check (within about a second).
+    // If it is gone, the proposal expires anyway.
+    if current.record.abort_requested.is_none() {
+        let request = AbortRequest {
+            by: host.node_id(),
+            at: unix_millis(),
+        };
+        let record_id = id.clone();
+        if let Err(error) = host
+            .commit_doc(move |doc| records::write_abort_request(doc, &record_id, &request))
+            .await
+        {
+            error!(compaction = %id, error = %format!("{error:#}"), "Failed to write abort request");
+            return error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "failed to record the abort request",
+                None,
+            );
+        }
+        info!(compaction = %id, "Abort requested");
+    }
+
+    let report = if query.wait {
+        wait_until_final(&host, &id).await
+    } else {
+        load_report(&host, &id).await
+    };
+    match report {
+        Some(report) if report.phase.is_final() => Json(report).into_response(),
+        Some(report) => (StatusCode::ACCEPTED, Json(report)).into_response(),
+        None => error_response(StatusCode::NOT_FOUND, "no such compaction", None),
+    }
+}
+
+/// `GET /admin/epochs`: the local epoch and the previous epochs this node still keeps.
+pub(crate) async fn epochs<H: Host>(State(host): State<H>) -> Response {
+    let current = EpochRecord::from(host.current_epoch());
+    Json(json!({
+        "current": current,
+        "archived": host.compaction().archived_epochs(),
+    }))
+    .into_response()
+}