summaryrefslogtreecommitdiffstats
path: root/src/compaction/api.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/compaction/api.rs')
-rw-r--r--src/compaction/api.rs473
1 files changed, 0 insertions, 473 deletions
diff --git a/src/compaction/api.rs b/src/compaction/api.rs
deleted file mode 100644
index 7f423da..0000000
--- a/src/compaction/api.rs
+++ /dev/null
@@ -1,473 +0,0 @@
-//! `/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. `?force=true`
-//! accepts a `settleSecs` below the dormancy threshold (tests, single-site clusters).
-//! * `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, the archived previous ones and whether
-//! the node is catching up with a newer epoch from a peer.
-//! * `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.
-//! `DELETE /admin/compaction` does the same for the compaction 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::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 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),
-}
-
-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.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> {
- // 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!(compaction = %id, seen, misses, "Compaction 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 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,
- };
-
- // 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 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"
- );
-
- 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()
-}
-
-/// 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 compaction", None),
- }
-}
-
-/// `DELETE /admin/compactions/{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/compaction`: aborts whichever compaction 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 compaction 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 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 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, 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.compaction().archived_epochs(),
- "catchUp": host.compaction().catch_up_report(),
- "quarantine": host.compaction().list_quarantine(|entry| entry.id.clone()),
- }))
- .into_response()
-}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-22 23:19:41 +0000