diff options
Diffstat (limited to 'src/remake/api.rs')
| -rw-r--r-- | src/remake/api.rs | 258 |
1 files changed, 181 insertions, 77 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs index 3530423..24fb7e8 100644 --- a/src/remake/api.rs +++ b/src/remake/api.rs @@ -69,10 +69,11 @@ use super::coordinator; use super::local::LocalStatus; use super::records::{ self, AbortRequest, ContentRef, Decision, EpochRecord, Phase, Policy, Proposal, Record, - Retention, Vote, unix_millis, + Retention, Vote, }; use super::{Host, kind, vote}; use crate::document::store::StoreError; +use crate::force::Force; #[derive(Debug, Default, Deserialize)] pub(crate) struct WaitQuery { @@ -80,12 +81,20 @@ pub(crate) struct WaitQuery { wait: bool, } +#[derive(Debug, Deserialize)] +pub(crate) struct AbortQuery { + #[serde(default)] + wait: bool, + /// The checks to override (`force`): `coordinatorActive`. + force: Option<String>, +} + #[derive(Debug, Default, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct StartQuery { #[serde(default)] wait: bool, - /// The checks to override (`Force`). + /// 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: @@ -160,41 +169,63 @@ struct StartRequest { 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, +/// What `?force=` may override on a proposal (`force`). +const FORCE: &[&str] = &["switchDelaySecs", "supersede", "coordinatorActive"]; +/// What `?force=` may override on an abort. +const ABORT_FORCE: &[&str] = &["coordinatorActive"]; + +/// When a node other than the coordinator may abort an expired remake: a whole grace +/// period after it expired. Its coordinator stops committing at expiry +/// (`coordinator::decide`), so an abort from elsewhere and a commit can only cross if +/// their clocks are more than a grace period apart. +fn abortable_at(record: &Record, grace: u64) -> u64 { + record + .proposal + .settle_until + .saturating_add(grace.saturating_mul(2)) } -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) +/// Why this node may not abort expired `record` now, and the `force` name that would +/// override it (none for one that only time cures). +fn abort_obstacle( + record: &Record, + me: &str, + active_peers: &HashSet<String>, + force: &Force, + now: u64, + grace: u64, +) -> Option<(String, Option<&'static str>)> { + let at = abortable_at(record, grace); + if now < at { + return Some(( + format!( + "only its coordinator may abort it until a grace period after it expired ({}s from now)", + (at - now).div_ceil(1000) + ), + None, + )); } + let coordinator = &record.proposal.coordinator; + if coordinator != me && active_peers.contains(coordinator) && !force.allows("coordinatorActive") + { + return Some(( + format!( + "its coordinator {coordinator} is an active peer, so it is stuck rather than gone (look at its log); add force=coordinatorActive to abort it anyway" + ), + Some("coordinatorActive"), + )); + } + None +} + +/// The active peers this node sees. +async fn active_peers<H: Host>(host: &H) -> HashSet<String> { + host.peer_summaries() + .await + .into_iter() + .filter(|peer| peer.active) + .map(|peer| peer.id) + .collect() } /// An expired remake of the epoch a proposal would supersede. @@ -202,11 +233,20 @@ impl Force { struct ExpiredRemake { id: String, coordinator: String, + coordinator_active: bool, /// When it expired (Unix milliseconds). since: u64, + /// When another node may abort it (`abortable_at`). + abortable_at: u64, tally: vote::Tally, } +/// Superseding expired remakes, as the proposal's `force` asks. +struct Supersede { + force: Force, + active_peers: HashSet<String>, +} + #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct Report { @@ -223,9 +263,15 @@ 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>), + /// Expired remakes still hold votes, and aren't superseded: not asked to + /// (`force=supersede`), or not possible yet (`abort_obstacle`). + #[error("{message}")] + Expired { + message: String, + remakes: Vec<ExpiredRemake>, + /// The `force` name that would override it, if any. + force: Option<&'static str>, + }, /// 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), @@ -253,7 +299,7 @@ fn grace<H: Host>(host: &H) -> u64 { fn report<H: Host>(host: &H, id: String, record: Record) -> Report { Report { - phase: record.phase(unix_millis(), grace(host)), + phase: record.phase(host.now_millis(), grace(host)), local: host.remake().status(&id), id, record, @@ -333,7 +379,7 @@ 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()) { + let force = match Force::parse(query.force.as_deref(), FORCE) { Ok(force) => force, Err(reason) => return error_response(StatusCode::BAD_REQUEST, reason, None), }; @@ -436,7 +482,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 && force.switch_delay => { + Some(delay) if delay < keepalive && force.allows("switchDelaySecs") => { warn!( switch_delay_secs = delay.as_secs(), keepalive_secs = keepalive.as_secs(), @@ -456,7 +502,7 @@ async fn start_kind<H: Host>( _ => {} } - let now = unix_millis(); + let now = host.now_millis(); let settle_ms = settle.as_millis() as u64; let id = Uuid::new_v4().to_string(); let grace = grace(&host); @@ -502,7 +548,10 @@ async fn start_kind<H: Host>( id.clone(), proposal, base, - force.supersede, + Supersede { + force: force.clone(), + active_peers: active_peers(&host).await, + }, now, grace, )); @@ -527,33 +576,28 @@ 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 + Err(StartError::Expired { + message, + remakes, + force, + }) => { + let expired: Vec<JsonValue> = remakes .iter() .map(|remake| { json!({ "id": remake.id, "coordinator": remake.coordinator, - "coordinatorActive": remake.coordinator == me - || active.contains(&remake.coordinator), + "coordinatorActive": remake.coordinator_active, "expiredAt": remake.since, + "abortableAt": remake.abortable_at, "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" })), + message, + Some(json!({ "expired": expired, "force": force })), ); } Err(error @ StartError::VoteHeld(_)) => { @@ -601,7 +645,7 @@ async fn propose<H: Host>( id: String, mut proposal: Proposal, base: Option<Vec<ChangeHash>>, - supersede: bool, + supersede: Supersede, now: u64, grace: u64, ) -> Result<Record, StartError> { @@ -632,18 +676,50 @@ async fn propose<H: Host>( .map(|(id, _)| id) .collect(); if !expired.is_empty() { - if !supersede { - let details = all - .iter() + let remakes = || { + all.iter() .filter(|(id, _)| expired.contains(&id)) .map(|(id, record)| ExpiredRemake { id: id.clone(), coordinator: record.proposal.coordinator.clone(), + coordinator_active: supersede + .active_peers + .contains(&record.proposal.coordinator), since: record.proposal.settle_until.saturating_add(grace), + abortable_at: abortable_at(record, grace), tally: vote::tally(id, record, &all, ®ister), }) - .collect(); - return Err(StartError::Expired(details)); + .collect() + }; + if !supersede.force.allows("supersede") { + return Err(StartError::Expired { + message: "an expired remake of this epoch still holds votes: abort it, \ + or add ?force=supersede to abort it and propose in its place" + .to_owned(), + remakes: remakes(), + force: Some("supersede"), + }); + } + let obstacle = + all.iter() + .filter(|(id, _)| expired.contains(&id)) + .find_map(|(id, record)| { + let (why, force) = abort_obstacle( + record, + &me, + &supersede.active_peers, + &supersede.force, + now, + grace, + )?; + Some((format!("can't supersede remake {id}: {why}"), force)) + }); + if let Some((message, force)) = obstacle { + return Err(StartError::Expired { + message, + remakes: remakes(), + force, + }); } let decision = Decision::Aborted { reason: format!("superseded by remake {record_id}"), @@ -743,16 +819,16 @@ pub(crate) async fn get<H: Host>( pub(crate) async fn abort<H: Host>( State(host): State<H>, Path(id): Path<String>, - Query(query): Query<WaitQuery>, + Query(query): Query<AbortQuery>, ) -> Response { - abort_by_id(host, id, query.wait).await + abort_by_id(host, id, query).await } /// `DELETE /admin/remakes`: aborts whichever remake of the local epoch is still /// 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>, + Query(query): Query<AbortQuery>, ) -> Response { let local = EpochRecord::from(host.current_epoch()); let open = host @@ -768,7 +844,7 @@ pub(crate) async fn abort_current<H: Host>( .await; match open { - Some(id) => abort_by_id(host, id, query.wait).await, + Some(id) => abort_by_id(host, id, query).await, None => error_response( StatusCode::NOT_FOUND, "no remake of the local epoch is in progress", @@ -777,14 +853,19 @@ pub(crate) async fn abort_current<H: Host>( } } -async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { +async fn abort_by_id<H: Host>(host: H, id: String, query: AbortQuery) -> Response { + let force = match Force::parse(query.force.as_deref(), ABORT_FORCE) { + Ok(force) => force, + Err(reason) => return error_response(StatusCode::BAD_REQUEST, reason, None), + }; + let wait = query.wait; 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::Expired => return abort_expired(&host, &id).await, + Phase::Expired => return abort_expired(&host, current, &force).await, Phase::Committed => { return error_response( StatusCode::CONFLICT, @@ -806,7 +887,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { if current.record.abort_requested.is_none() { let request = AbortRequest { by: host.node_id(), - at: unix_millis(), + at: host.now_millis(), }; let record_id = id.clone(); if let Err(error) = host @@ -835,22 +916,45 @@ 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 { +/// Aborts expired remake `current` from this node: its coordinator is gone, and can't +/// commit it any more (`coordinator`), unless `abort_obstacle` says otherwise. Answers +/// with the final report. +async fn abort_expired<H: Host>(host: &H, current: Report, force: &Force) -> Response { + let grace = grace(host); + if let Some((why, force)) = abort_obstacle( + ¤t.record, + &host.node_id(), + &active_peers(host).await, + force, + host.now_millis(), + grace, + ) { + return error_response( + StatusCode::CONFLICT, + format!("can't abort it yet: {why}"), + Some(json!({ + "remake": current, + "abortableAt": abortable_at(¤t.record, grace), + "force": force, + })), + ); + } + let id = current.id.as_str(); 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(), + decided_at: host.now_millis(), }; - let grace = grace(host); + let clock = host.clock(); 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); + let now = clock.now_millis(); + let expired = records::load(doc, &record_id).is_some_and(|record| { + record.phase(now, grace) == Phase::Expired && now >= abortable_at(&record, grace) + }); if expired { records::write_decision(doc, &record_id, &decision)?; } |