diff options
Diffstat (limited to 'src/remake/api.rs')
| -rw-r--r-- | src/remake/api.rs | 260 |
1 files changed, 225 insertions, 35 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs index de8166c..826c81e 100644 --- a/src/remake/api.rs +++ b/src/remake/api.rs @@ -12,9 +12,10 @@ //! * `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>", …], @@ -22,15 +23,19 @@ //! "archiveRetention": {"epochs": 2, "seconds": 2592000}, //! "settleSecs": 480, "switchDelaySecs": 60 }` //! -//! `?force=true` accepts a `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), `replace`, whose spec is //! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }` //! (see `kind::Replace`), or `import`, whose spec is //! `{ "document": "<base64 of an Automerge document>", "rules": […] }` (see -//! `kind::Import`; `GET /admin/document` exports one). Reports leave the document out -//! (`contentBytes`). Requests may be up to 64 MiB. +//! `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 @@ -38,6 +43,7 @@ //! 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; @@ -61,7 +67,8 @@ use uuid::Uuid; use super::coordinator; use super::local::LocalStatus; use super::records::{ - self, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, Vote, unix_millis, + self, AbortRequest, ContentRef, Decision, EpochRecord, Phase, Policy, Proposal, Record, + Retention, Vote, unix_millis, }; use super::{Host, kind, vote}; use crate::document::store::StoreError; @@ -77,9 +84,8 @@ pub(crate) struct WaitQuery { pub(crate) struct StartQuery { #[serde(default)] wait: bool, - /// Allows a switch delay shorter than one keepalive interval. - #[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). @@ -149,10 +155,57 @@ struct StartRequest { /// 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 { @@ -160,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>, @@ -173,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)] @@ -195,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, } @@ -279,6 +332,10 @@ async fn start_kind<H: Host>( 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 is_octet_stream(headers) { match query.import_request(&body) { @@ -369,7 +426,7 @@ async fn start_kind<H: Host>( 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(), @@ -380,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, @@ -394,7 +451,7 @@ async fn start_kind<H: Host>( 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(), @@ -414,9 +471,17 @@ async fn start_kind<H: Host>( 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 @@ -426,6 +491,7 @@ async fn start_kind<H: Host>( id.clone(), proposal, base, + force.supersede, now, grace, )); @@ -450,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); } @@ -492,19 +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 @@ -534,12 +682,16 @@ async fn propose<H: Host>( proposal.electorate = electorate; records::write_proposal(doc, &record_id, &proposal)?; - let own_vote = Vote { at: now }; - records::write_vote(doc, &record_id, &me, &own_vote)?; + 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(), }) }) @@ -585,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) @@ -624,6 +772,7 @@ 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, @@ -631,7 +780,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { Some(json!({ "remake": current })), ); } - Phase::Aborted | Phase::Expired => { + Phase::Aborted => { return error_response( StatusCode::CONFLICT, "already over", @@ -641,7 +790,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { } // 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(), @@ -674,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 { |