summaryrefslogtreecommitdiffstats
path: root/src/remake/api.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/remake/api.rs')
-rw-r--r--src/remake/api.rs474
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()
+}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-23 19:52:53 +0000