diff options
Diffstat (limited to 'src/remake')
| -rw-r--r-- | src/remake/api.rs | 212 | ||||
| -rw-r--r-- | src/remake/coordinator.rs | 2 | ||||
| -rw-r--r-- | src/remake/records.rs | 7 |
3 files changed, 179 insertions, 42 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs index 6b1a30c..0428a43 100644 --- a/src/remake/api.rs +++ b/src/remake/api.rs @@ -81,6 +81,14 @@ 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 { @@ -162,18 +170,83 @@ struct StartRequest { } /// What `?force=` may override on a proposal (`force`). -const FORCE: &[&str] = &["switchDelaySecs", "supersede"]; +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)) +} + +/// 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. #[derive(Debug)] 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 { @@ -190,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), @@ -469,7 +548,10 @@ async fn start_kind<H: Host>( id.clone(), proposal, base, - force.allows("supersede"), + Supersede { + force: force.clone(), + active_peers: active_peers(&host).await, + }, now, grace, )); @@ -494,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(_)) => { @@ -568,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> { @@ -599,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}"), @@ -710,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 @@ -735,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", @@ -744,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, @@ -802,9 +916,30 @@ 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, + unix_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!( @@ -813,11 +948,12 @@ async fn abort_expired<H: Host>(host: &H, id: &str) -> Response { ), 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); + let now = unix_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)?; } diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs index 3a7b071..427de08 100644 --- a/src/remake/coordinator.rs +++ b/src/remake/coordinator.rs @@ -12,7 +12,7 @@ //! The decision is written in the same step that checks the records once more, so a //! commit never overrides an abort request or a competitor's commit that arrived while //! the snapshot was being built. Nor is a commit written once the remake has expired: -//! from then on, any node may abort it (`api`), and the two mustn't race. +//! a grace period later, any node may abort it (`api`), and the two mustn't race. //! //! Committing builds the snapshot at the base heads and records the new epoch, so //! every node can verify its own build against it. diff --git a/src/remake/records.rs b/src/remake/records.rs index 0e40960..283f612 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -8,9 +8,10 @@ //! * `abortRequested`: written by any node on an operator's request. It is an input to //! the coordinator's decision, not a decision itself. //! -//! Once a remake has expired, its coordinator no longer commits it, and any node may -//! abort it on an operator's request (`api`). Should clocks disagree by more than the -//! grace margin, a commit and such an abort could still be written concurrently. Such +//! Once a remake has expired, its coordinator no longer commits it, and a grace period +//! later any node may abort it on an operator's request (`api`). Should clocks disagree +//! by more than that grace period, a commit and such an abort could still be written +//! concurrently. Such //! a record is *contested*: it reads as committed, and nodes don't switch to it on //! their own. //! |