rename compaction to remake

1 files changed, 474 insertions, 0 deletions

diff --git a/src/remake/api.rs b/src/remake/api.rs
new file mode 100644
index 0000000..9b490f4
--- /dev/null
+++ b/src/remake/api.rs
@@ -0,0 +1,474 @@
+//! `/admin/remakes` handlers.
+//!
+//! * `POST /admin/remakes` proposes a remake of the current epoch (`POST
+//!   /admin/compaction` is a shortcut, kept from before). 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. `?force=true`
+//!   accepts a `settleSecs` below the dormancy threshold (tests, single-site clusters).
+//! * `GET /admin/remakes` lists all remakes.
+//! * `GET /admin/remakes/{id}` reports one; `?wait=true` long-polls until decided
+//!   (bounded by the proposal's own expiry).
+//! * `GET /admin/epochs` shows the local epoch, the archived previous ones and whether
+//!   the node is catching up with a newer epoch from a peer.
+//! * `DELETE /admin/remakes/{id}` asks the coordinator to abort. Only possible while
+//!   proposed; `409` once committed (a switch can't be undone) or already over.
+//!   `DELETE /admin/remakes` does the same for the remake of the local epoch that
+//!   is in progress, without having to know its ID.
+//!
+//! 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, warn};
+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::document::store::StoreError;
+
+#[derive(Debug, Default, Deserialize)]
+pub(crate) struct WaitQuery {
+    #[serde(default)]
+    wait: bool,
+}
+
+#[derive(Debug, Default, Deserialize)]
+pub(crate) struct StartQuery {
+    #[serde(default)]
+    wait: bool,
+    /// Allows a settle period shorter than the dormancy threshold.
+    #[serde(default)]
+    force: 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 remake on the node answering the request.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    local: Option<LocalStatus>,
+}
+
+#[derive(Debug, thiserror::Error)]
+enum StartError {
+    #[error("remake {0} is already open for this epoch")]
+    Open(String),
+    #[error(transparent)]
+    Store(#[from] StoreError),
+}
+
+pub(super) 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.remake().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 remake 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> {
+    // A record that was readable may be briefly unreadable or absent while concurrent
+    // changes merge; give up only if that persists.
+    const MAX_MISSES: u32 = 5;
+
+    let mut changes = host.subscribe_changes();
+    let mut seen = false;
+    let mut misses = 0;
+    loop {
+        let Some(report) = load_report(host, id).await else {
+            misses += 1;
+            warn!(remake = %id, seen, misses, "Remake record not readable while waiting");
+            if !seen || misses >= MAX_MISSES {
+                return None;
+            }
+            tokio::time::sleep(Duration::from_secs(1)).await;
+            continue;
+        };
+        seen = true;
+        misses = 0;
+        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<StartQuery>,
+    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 && query.force {
+        warn!(
+            settle_secs = settle.as_secs_f64(),
+            dormant_after_secs = dormant_after.as_secs(),
+            "Forced settle period shorter than the dormancy threshold; a partition may go unnoticed"
+        );
+    } else 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; add ?force=true to override",
+                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 remake; 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 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,
+    };
+
+    // From here on, nothing depends on the client: writing the proposal and starting the
+    // coordinator run in their own task, so a client that disconnects mid-request can't
+    // leave a proposal without a coordinator. Dropping the handle doesn't cancel it.
+    let proposing = tokio::spawn(propose(host.clone(), id.clone(), proposal, now, grace));
+    let written = match proposing.await {
+        Ok(written) => written,
+        Err(error) => {
+            error!(error = %error, "Proposal task failed");
+            return error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "failed to write the proposal",
+                None,
+            );
+        }
+    };
+
+    let record = match written {
+        Ok(record) => record,
+        Err(StartError::Open(open)) => {
+            return error_response(
+                StatusCode::CONFLICT,
+                "another remake of this epoch is still open",
+                Some(json!({ "remake": open })),
+            );
+        }
+        Err(StartError::Store(error)) => {
+            error!(error = %format!("{error:#}"), "Failed to write remake proposal");
+            return error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "failed to write the proposal",
+                None,
+            );
+        }
+    };
+
+    info!(
+        remake = %id,
+        settle_secs = settle.as_secs(),
+        "Remake proposed"
+    );
+
+    if query.wait {
+        return match wait_until_final(&host, &id).await {
+            Some(report) => Json(report).into_response(),
+            None => error_response(StatusCode::NOT_FOUND, "remake record vanished", None),
+        };
+    }
+
+    let location = format!("/admin/remakes/{id}");
+    (
+        StatusCode::ACCEPTED,
+        [(LOCATION, location)],
+        Json(report(&host, id, record)),
+    )
+        .into_response()
+}
+
+/// Writes the proposal (unless another one is open for the epoch) and starts its
+/// coordinator. Runs as its own task, independent of the HTTP request.
+async fn propose<H: Host>(
+    host: H,
+    id: String,
+    mut proposal: Proposal,
+    now: u64,
+    grace: u64,
+) -> Result<Record, StartError> {
+    let record_id = id.clone();
+    let record = 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?;
+
+    tokio::spawn(coordinator::run(host, id));
+    Ok(record)
+}
+
+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 remake", None),
+    }
+}
+
+/// `DELETE /admin/remakes/{id}`.
+pub(crate) async fn abort<H: Host>(
+    State(host): State<H>,
+    Path(id): Path<String>,
+    Query(query): Query<WaitQuery>,
+) -> Response {
+    abort_by_id(host, id, query.wait).await
+}
+
+/// `DELETE /admin/remakes`: aborts whichever remake of the local epoch is still
+/// proposed. There is at most one, unless two coordinators raced (then both abort anyway
+/// and this picks the oldest).
+pub(crate) async fn abort_current<H: Host>(
+    State(host): State<H>,
+    Query(query): Query<WaitQuery>,
+) -> Response {
+    let local = EpochRecord::from(host.current_epoch());
+    let now = unix_millis();
+    let grace = grace(&host);
+    let open = host
+        .read_doc(move |doc| {
+            records::load_all(doc)
+                .into_iter()
+                .filter(|(_, record)| {
+                    record.proposal.from_epoch == local
+                        && record.phase(now, grace) == Phase::Proposed
+                })
+                .min_by_key(|(_, record)| record.proposal.proposed_at)
+                .map(|(id, _)| id)
+        })
+        .await;
+
+    match open {
+        Some(id) => abort_by_id(host, id, query.wait).await,
+        None => error_response(
+            StatusCode::NOT_FOUND,
+            "no remake of the local epoch is in progress",
+            None,
+        ),
+    }
+}
+
+async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
+    let Some(current) = load_report(&host, &id).await else {
+        return error_response(StatusCode::NOT_FOUND, "no such remake", None);
+    };
+
+    match current.phase {
+        Phase::Proposed => {}
+        Phase::Committed => {
+            return error_response(
+                StatusCode::CONFLICT,
+                "already committed; the switch cannot be undone",
+                Some(json!({ "remake": current })),
+            )
+        }
+        Phase::Aborted | Phase::Expired => {
+            return error_response(
+                StatusCode::CONFLICT,
+                "already over",
+                Some(json!({ "remake": 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!(remake = %id, error = %format!("{error:#}"), "Failed to write abort request");
+            return error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "failed to record the abort request",
+                None,
+            );
+        }
+        info!(remake = %id, "Abort requested");
+    }
+
+    let report = if 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 remake", None),
+    }
+}
+
+/// `GET /admin/epochs`: the local epoch, the previous epochs this node still keeps, and
+/// an ongoing catch-up with a peer's newer epoch (`null` if none).
+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.remake().archived_epochs(),
+        "catchUp": host.remake().catch_up_report(),
+        "quarantine": host.remake().list_quarantine(|entry| entry.id.clone()),
+    }))
+    .into_response()
+}