diff options
Diffstat (limited to 'src/remake/api.rs')
| -rw-r--r-- | src/remake/api.rs | 422 |
1 files changed, 341 insertions, 81 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs index 5267e48..826c81e 100644 --- a/src/remake/api.rs +++ b/src/remake/api.rs @@ -3,17 +3,19 @@ //! * `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 swarms). +//! report once decided. A dropped request does not abort anything. The remake +//! commits as soon as a majority of its electorate voted for it (`vote`), and aborts +//! once that is out of reach or `settleSecs` passed without it. //! * `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. +//! undecided; `409` once committed (a switch can't be undone) or aborted. An expired +//! remake is aborted right away by the node asked: its coordinator is gone, and +//! can't commit it any more anyway (`coordinator`). `DELETE /admin/remakes` does the +//! same for the undecided remake of the local epoch, without having to know its ID. //! //! Request body for POST (all optional): //! `{ "kind": "compact", "spec": null, "baseHeads": ["<hex>", …], @@ -21,25 +23,43 @@ //! "archiveRetention": {"epochs": 2, "seconds": 2592000}, //! "settleSecs": 480, "switchDelaySecs": 60 }` //! -//! `?force=true` accepts `settleSecs` below the dormancy threshold and `switchDelaySecs` -//! below one keepalive interval: for tests and debugging, not for real swarms. +//! `?force=` overrides checks: `true` all of them, or a comma-separated list of +//! * `switchDelaySecs`: accept one below a keepalive interval (for tests and debugging, +//! not for real swarms); +//! * `supersede`: abort expired remakes of this epoch and propose in their place. +//! Without it, an expired remake still holding votes answers `409`. //! -//! `kind` is `compact` (no spec) or `replace`, whose spec is +//! `kind` is `compact` (no spec), `replace`, whose spec is //! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }` -//! (see `kind::Replace`). Reports leave the replacement document out (`contentBytes`). +//! (see `kind::Replace`), or `import`, whose spec is +//! `{ "document": "<base64 of an Automerge document>", "rules": […] }` (see +//! `kind::Import`; `GET /admin/document` exports one). The document stays on the +//! coordinator; the proposal refers to it by hash (`proposal.content`), and every voter +//! downloads it (`content`). Requests may be up to 64 MiB. +//! +//! An import can also be uploaded as is, with `Content-Type: application/octet-stream`; +//! then `rules` (JSON), `baseHeads` (comma-separated), `settleSecs` and +//! `switchDelaySecs` go in the query. Restoring a backup, keeping writes made since at +//! their paths: +//! `curl --data-binary @backup.automerge -H 'Content-Type: application/octet-stream' --url-query 'rules=[{"from":"","to":""}]' …/admin/remakes` +use std::collections::HashSet; use std::time::Duration; use automerge::ChangeHash; use axum::{ + Json, body::Bytes, extract::{Path, Query, State}, - http::{header::LOCATION, StatusCode}, + http::{ + HeaderMap, StatusCode, + header::{CONTENT_TYPE, LOCATION}, + }, response::{IntoResponse, Response}, - Json, }; +use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; use serde::{Deserialize, Serialize}; -use serde_json::{json, Value as JsonValue}; +use serde_json::{Value as JsonValue, json}; use tokio::sync::broadcast; use tracing::{error, info, warn}; use uuid::Uuid; @@ -47,9 +67,10 @@ use uuid::Uuid; use super::coordinator; use super::local::LocalStatus; use super::records::{ - self, unix_millis, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, + self, AbortRequest, ContentRef, Decision, EpochRecord, Phase, Policy, Proposal, Record, + Retention, Vote, unix_millis, }; -use super::{kind, Host}; +use super::{Host, kind, vote}; use crate::document::store::StoreError; #[derive(Debug, Default, Deserialize)] @@ -59,12 +80,63 @@ pub(crate) struct WaitQuery { } #[derive(Debug, Default, Deserialize)] +#[serde(rename_all = "camelCase")] pub(crate) struct StartQuery { #[serde(default)] wait: bool, - /// Allows a settle period shorter than the dormancy threshold. - #[serde(default)] - force: bool, + /// The checks to override (`Force`). + force: Option<String>, + // Only for an import uploaded as `application/octet-stream`, which has no JSON + // body to carry them: + /// The rules, as JSON (as in the spec). + rules: Option<String>, + /// Comma-separated hex. + base_heads: Option<String>, + settle_secs: Option<u64>, + switch_delay_secs: Option<u64>, +} + +impl StartQuery { + fn has_request_fields(&self) -> bool { + self.rules.is_some() + || self.base_heads.is_some() + || self.settle_secs.is_some() + || self.switch_delay_secs.is_some() + } + + /// An import request for a document uploaded as is. + fn import_request(&self, document: &[u8]) -> Result<StartRequest, String> { + let rules = match &self.rules { + None => JsonValue::Array(Vec::new()), + Some(rules) => serde_json::from_str(rules) + .map_err(|error| format!("rules are not valid JSON: {error}"))?, + }; + Ok(StartRequest { + kind: Some("import".to_owned()), + spec: Some(json!({ + "document": BASE64.encode(document), + "rules": rules, + })), + base_heads: self.base_heads.as_ref().map(|heads| { + heads + .split(',') + .filter(|head| !head.trim().is_empty()) + .map(str::to_owned) + .collect() + }), + settle_secs: self.settle_secs, + switch_delay_secs: self.switch_delay_secs, + ..StartRequest::default() + }) + } +} + +fn is_octet_stream(headers: &HeaderMap) -> bool { + headers + .get(CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.split(';').next()) + .is_some_and(|media| media.trim() == "application/octet-stream") } #[derive(Debug, Default, Deserialize)] @@ -80,12 +152,60 @@ struct StartRequest { base_heads: Option<Vec<String>>, replay_window: Option<Retention>, archive_retention: Option<Retention>, + /// How long votes are collected; twice the dormancy threshold if absent. settle_secs: Option<u64>, /// How long nodes wait after seeing the commit before switching; one keepalive - /// interval if absent. Shorter only with `?force=true`. + /// interval if absent. Shorter only with `?force=switchDelaySecs`. switch_delay_secs: Option<u64>, } +/// The checks `?force=` overrides: `true` all of them, or a comma-separated list. +#[derive(Debug, Default, Clone, Copy)] +struct Force { + switch_delay: bool, + supersede: bool, +} + +impl Force { + fn parse(value: Option<&str>) -> Result<Self, String> { + let mut force = Force::default(); + match value.map(str::trim) { + None | Some("" | "false") => {} + Some("true") => { + force = Force { + switch_delay: true, + supersede: true, + }; + } + Some(list) => { + for name in list.split(',').map(str::trim) { + match name { + "switchDelaySecs" => force.switch_delay = true, + "supersede" => force.supersede = true, + other => { + return Err(format!( + "unknown force {other:?}: use true, or a list of \ + switchDelaySecs and supersede" + )); + } + } + } + } + } + Ok(force) + } +} + +/// An expired remake of the epoch a proposal would supersede. +#[derive(Debug)] +struct ExpiredRemake { + id: String, + coordinator: String, + /// When it expired (Unix milliseconds). + since: u64, + tally: vote::Tally, +} + #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct Report { @@ -93,10 +213,6 @@ struct Report { phase: Phase, #[serde(flatten)] record: Record, - /// Size of the kind's build input (e.g. a replacement document), which reports - /// leave out. - #[serde(skip_serializing_if = "Option::is_none")] - content_bytes: Option<usize>, /// What happened to this remake on the node answering the request. #[serde(skip_serializing_if = "Option::is_none")] local: Option<LocalStatus>, @@ -106,6 +222,12 @@ struct Report { enum StartError { #[error("remake {0} is already open for this epoch")] Open(String), + /// Expired remakes still hold votes; superseding them takes `force=supersede`. + #[error("expired remakes of this epoch still hold votes")] + Expired(Vec<ExpiredRemake>), + /// The proposer votes for its own remake, so it must be free to. + #[error("this node can't vote for a new remake: {0}")] + VoteHeld(String), #[error("base head {0} is not in this node's document")] UnknownBase(String), #[error(transparent)] @@ -128,12 +250,10 @@ fn grace<H: Host>(host: &H) -> u64 { host.dormant_after().as_millis() as u64 } -fn report<H: Host>(host: &H, id: String, mut record: Record) -> Report { - let content = std::mem::take(&mut record.proposal.content); +fn report<H: Host>(host: &H, id: String, record: Record) -> Report { Report { phase: record.phase(unix_millis(), grace(host)), local: host.remake().status(&id), - content_bytes: (!content.is_null()).then(|| content.to_string().len()), id, record, } @@ -186,31 +306,55 @@ async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> { pub(crate) async fn start<H: Host>( State(host): State<H>, Query(query): Query<StartQuery>, + headers: HeaderMap, body: Bytes, ) -> Response { - start_kind(host, query, body, None).await + start_kind(host, query, &headers, body, None).await } /// `POST /admin/compaction`: the original endpoint, for compacting remakes only. pub(crate) async fn start_compaction<H: Host>( State(host): State<H>, Query(query): Query<StartQuery>, + headers: HeaderMap, body: Bytes, ) -> Response { - start_kind(host, query, body, Some(kind::DEFAULT_KIND)).await + start_kind(host, query, &headers, body, Some(kind::DEFAULT_KIND)).await } async fn start_kind<H: Host>( host: H, query: StartQuery, + headers: &HeaderMap, body: Bytes, only: Option<&'static str>, ) -> Response { if !host.is_node_ready().await { return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None); } + let force = match Force::parse(query.force.as_deref()) { + Ok(force) => force, + Err(reason) => return error_response(StatusCode::BAD_REQUEST, reason, None), + }; - let request: StartRequest = if body.is_empty() { + let request: StartRequest = if is_octet_stream(headers) { + match query.import_request(&body) { + Ok(request) => request, + Err(reason) => { + return error_response( + StatusCode::BAD_REQUEST, + format!("invalid request: {reason}"), + None, + ); + } + } + } else if query.has_request_fields() { + return error_response( + StatusCode::BAD_REQUEST, + "rules, baseHeads, settleSecs and switchDelaySecs go in the query only for an application/octet-stream import; put them in the JSON body", + None, + ); + } else if body.is_empty() { StartRequest::default() } else { match serde_json::from_slice(&body) { @@ -220,7 +364,7 @@ async fn start_kind<H: Host>( StatusCode::BAD_REQUEST, format!("invalid request: {error}"), None, - ) + ); } } }; @@ -247,7 +391,7 @@ async fn start_kind<H: Host>( StatusCode::BAD_REQUEST, format!("invalid {kind_name} spec: {reason}"), None, - ) + ); } }; @@ -269,34 +413,20 @@ async fn start_kind<H: Host>( } }; + // The voting period. No lower bound: the majority protects against competitors, + // a short period only risks an abort for lack of votes. 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, - ); - } // The delay lets the commit reach every node through ordinary sync before any of // them stops syncing the old epoch. let keepalive = dormant_after / 4; let switch_delay = request.switch_delay_secs.map(Duration::from_secs); match switch_delay { - Some(delay) if delay < keepalive && query.force => { + Some(delay) if delay < keepalive && force.switch_delay => { warn!( switch_delay_secs = delay.as_secs(), keepalive_secs = keepalive.as_secs(), @@ -307,7 +437,7 @@ async fn start_kind<H: Host>( return error_response( StatusCode::BAD_REQUEST, format!( - "switchDelaySecs must be at least {} (one keepalive interval), so the commit reaches every node before they switch; add ?force=true to override", + "switchDelaySecs must be at least {} (one keepalive interval), so the commit reaches every node before they switch; add ?force=switchDelaySecs to override", keepalive.as_secs() ), None, @@ -316,30 +446,12 @@ async fn start_kind<H: Host>( _ => {} } - // 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 { + let mut proposal = Proposal { coordinator: host.node_id(), from_epoch: host.current_epoch().into(), base_heads: Vec::new(), @@ -355,11 +467,21 @@ async fn start_kind<H: Host>( proposed_at: now, settle_until: now + settle_ms, switch_delay_ms: switch_delay.map(|delay| delay.as_millis() as u64), - peers_considered, + // Filled in when the proposal is written, from the document at that moment. + electorate: Vec::new(), + quorum: 0, kind: kind.name().to_owned(), - content: prepared.content, + content: None, translation: prepared.translation, }; + // The content stays here, referred to by hash; voters fetch it from us. + if let Some(bytes) = prepared.content { + let reference = ContentRef::of(&bytes); + if let Some(hash) = reference.hash() { + host.remake().put_content(hash, std::sync::Arc::new(bytes)); + } + proposal.content = Some(reference); + } // 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 @@ -369,6 +491,7 @@ async fn start_kind<H: Host>( id.clone(), proposal, base, + force.supersede, now, grace, )); @@ -393,6 +516,38 @@ async fn start_kind<H: Host>( Some(json!({ "remake": open })), ); } + Err(StartError::Expired(expired)) => { + let active: HashSet<String> = host + .peer_summaries() + .await + .into_iter() + .filter(|peer| peer.active) + .map(|peer| peer.id) + .collect(); + let me = host.node_id(); + let expired: Vec<JsonValue> = expired + .iter() + .map(|remake| { + json!({ + "id": remake.id, + "coordinator": remake.coordinator, + "coordinatorActive": remake.coordinator == me + || active.contains(&remake.coordinator), + "expiredAt": remake.since, + "tally": remake.tally.to_string(), + }) + }) + .collect(); + return error_response( + StatusCode::CONFLICT, + "an expired remake of this epoch still holds votes: abort it, or add \ + ?force=supersede to abort it and propose in its place", + Some(json!({ "expired": expired, "force": "supersede" })), + ); + } + Err(error @ StartError::VoteHeld(_)) => { + return error_response(StatusCode::CONFLICT, error.to_string(), None); + } Err(error @ StartError::UnknownBase(_)) => { return error_response(StatusCode::BAD_REQUEST, error.to_string(), None); } @@ -435,17 +590,69 @@ async fn propose<H: Host>( id: String, mut proposal: Proposal, base: Option<Vec<ChangeHash>>, + supersede: bool, now: u64, grace: u64, ) -> Result<Record, StartError> { let record_id = id.clone(); + let electorate_host = host.clone(); + let me = host.node_id(); + let started = host.remake().started_at() + grace; let record = host .commit_doc(move |doc| -> Result<Record, StartError> { - if let Some((open, _)) = records::load_all(doc).into_iter().find(|(_, record)| { + let mut all = records::load_all(doc); + if let Some((open, _)) = all.iter().find(|(_, record)| { record.proposal.from_epoch == proposal.from_epoch && record.phase(now, grace).is_open() }) { - return Err(StartError::Open(open)); + return Err(StartError::Open(open.clone())); + } + let mut register = records::load_register(doc); + + // Expired remakes hold their votes; their coordinators can't commit them + // any more, so they may be aborted here. + let expired: Vec<&String> = all + .iter() + .filter(|(_, record)| { + record.proposal.from_epoch == proposal.from_epoch + && record.phase(now, grace) == Phase::Expired + }) + .map(|(id, _)| id) + .collect(); + if !expired.is_empty() { + if !supersede { + let details = all + .iter() + .filter(|(id, _)| expired.contains(&id)) + .map(|(id, record)| ExpiredRemake { + id: id.clone(), + coordinator: record.proposal.coordinator.clone(), + since: record.proposal.settle_until.saturating_add(grace), + tally: vote::tally(id, record, &all, ®ister), + }) + .collect(); + return Err(StartError::Expired(details)); + } + let decision = Decision::Aborted { + reason: format!("superseded by remake {record_id}"), + decided_at: now, + }; + for id in expired { + records::write_decision(doc, id, &decision)?; + info!(remake = %id, by = %record_id, "Superseded expired remake"); + } + all = records::load_all(doc); + register = records::load_register(doc); + } + + match vote::freedom(&me, &proposal.from_epoch, &all, ®ister, started) { + vote::Freedom::Any => {} + vote::Freedom::Only(id) => { + return Err(StartError::VoteHeld(format!( + "it may have voted for remake {id} before its start" + ))); + } + vote::Freedom::Held(reason) => return Err(StartError::VoteHeld(reason)), } // The base is the requested heads, or everything before the proposal (the @@ -466,11 +673,26 @@ async fn propose<H: Host>( heads.sort(); proposal.base_heads = heads; + // We vote for our own remake, so we are always in the electorate. + let mut electorate = electorate_host.electorate(doc, now); + if let Err(position) = electorate.binary_search(&me) { + electorate.insert(position, me.clone()); + } + proposal.quorum = vote::quorum(electorate.len()); + proposal.electorate = electorate; + records::write_proposal(doc, &record_id, &proposal)?; + let own_vote = Vote { + remake: record_id.clone(), + at: now, + }; + records::write_vote(doc, &me, &own_vote)?; Ok(Record { proposal, decision: None, abort_requested: None, + contested: false, + votes: [(me, own_vote)].into(), }) }) .await?; @@ -515,22 +737,18 @@ pub(crate) async fn abort<H: Host>( } /// `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). +/// undecided (proposed or expired), the oldest if there are several. 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 + record.proposal.from_epoch == local && record.decision.is_none() }) .min_by_key(|(_, record)| record.proposal.proposed_at) .map(|(id, _)| id) @@ -554,24 +772,25 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { match current.phase { Phase::Proposed => {} + Phase::Expired => return abort_expired(&host, &id).await, Phase::Committed => { return error_response( StatusCode::CONFLICT, "already committed; the switch cannot be undone", Some(json!({ "remake": current })), - ) + ); } - Phase::Aborted | Phase::Expired => { + Phase::Aborted => { 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 it is gone, the proposal expires, and can then be aborted right away. if current.record.abort_requested.is_none() { let request = AbortRequest { by: host.node_id(), @@ -604,6 +823,47 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { } } +/// Aborts expired remake `id` from this node: its coordinator is gone, and can't commit +/// it any more (`coordinator`). Answers with the final report. +async fn abort_expired<H: Host>(host: &H, id: &str) -> Response { + let record_id = id.to_owned(); + let decision = Decision::Aborted { + reason: format!( + "aborted by operator via {} after it expired", + host.node_id() + ), + decided_at: unix_millis(), + }; + let grace = grace(host); + let written = host + .commit_doc(move |doc| -> Result<bool, StoreError> { + let expired = records::load(doc, &record_id) + .is_some_and(|record| record.phase(unix_millis(), grace) == Phase::Expired); + if expired { + records::write_decision(doc, &record_id, &decision)?; + } + Ok(expired) + }) + .await; + match written { + Ok(true) => info!(remake = %id, "Aborted expired remake"), + // Decided meanwhile: the report says how. + Ok(false) => {} + Err(error) => { + error!(remake = %id, error = %format!("{error:#}"), "Failed to abort expired remake"); + return error_response( + StatusCode::INTERNAL_SERVER_ERROR, + "failed to record the abort", + None, + ); + } + } + match load_report(host, id).await { + Some(report) => 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 { |