diff options
Diffstat (limited to 'src/remake')
| -rw-r--r-- | src/remake/api.rs | 211 | ||||
| -rw-r--r-- | src/remake/records.rs | 89 | ||||
| -rw-r--r-- | src/remake/switch.rs | 4 | ||||
| -rw-r--r-- | src/remake/vote.rs | 1 |
4 files changed, 279 insertions, 26 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs index ffd1844..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,8 +23,11 @@ //! "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}] }` @@ -39,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; @@ -62,8 +67,8 @@ use uuid::Uuid; use super::coordinator; use super::local::LocalStatus; use super::records::{ - self, AbortRequest, ContentRef, 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; @@ -79,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). @@ -151,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 { @@ -171,6 +222,9 @@ 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), @@ -278,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) { @@ -368,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(), @@ -379,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, @@ -433,6 +491,7 @@ async fn start_kind<H: Host>( id.clone(), proposal, base, + force.supersede, now, grace, )); @@ -457,6 +516,35 @@ 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); } @@ -502,6 +590,7 @@ async fn propose<H: Host>( id: String, mut proposal: Proposal, base: Option<Vec<ChangeHash>>, + supersede: bool, now: u64, grace: u64, ) -> Result<Record, StartError> { @@ -511,14 +600,51 @@ async fn propose<H: Host>( let started = host.remake().started_at() + grace; let record = host .commit_doc(move |doc| -> Result<Record, StartError> { - let all = records::load_all(doc); + 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.clone())); } - let register = records::load_register(doc); + 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) => { @@ -565,6 +691,7 @@ async fn propose<H: Host>( proposal, decision: None, abort_requested: None, + contested: false, votes: [(me, own_vote)].into(), }) }) @@ -610,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) @@ -649,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, @@ -656,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", @@ -666,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(), @@ -699,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 { diff --git a/src/remake/records.rs b/src/remake/records.rs index 37af491..cd3e5be 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -8,6 +8,12 @@ //! * `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 +//! a record is *contested*: it reads as committed, and nodes don't switch to it on +//! their own. +//! //! Votes are kept apart, in the votes register `_continuum/votes/<pk>`: one entry per //! member for the whole epoch, written by that member only (`vote`). Reading a record //! fills in the votes that name it. @@ -224,6 +230,14 @@ pub(crate) struct Record { pub(crate) decision: Option<Decision>, #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) abort_requested: Option<AbortRequest>, + /// Decisions written concurrently disagree; `decision` is then the commit. Not + /// stored: found when the record is read. + #[serde( + default, + skip_deserializing, + skip_serializing_if = "std::ops::Not::not" + )] + pub(crate) contested: bool, /// The register's votes naming this remake, by member public key. Not stored in /// the record: filled in when it is read (`load`). #[serde(default, skip_deserializing)] @@ -340,13 +354,42 @@ fn load_record(doc: &AutoCommit, id: &str) -> Option<Record> { return None; } }; - match serde_json::from_value(value) { - Ok(record) => Some(record), + let mut record: Record = match serde_json::from_value(value) { + Ok(record) => record, Err(error) => { warn!(remake = %id, error = %error, "Skipping unreadable remake record"); - None + return None; } + }; + if let Some(commit) = contested_decision(doc, id) { + record.decision = Some(commit); + record.contested = true; + } + Some(record) +} + +/// A commit among concurrent decisions of remake `id` that disagree, if they do. +fn contested_decision(doc: &AutoCommit, id: &str) -> Option<Decision> { + let decisions: Vec<Decision> = store::read_candidates(doc, &record_path(id, "decision")) + .ok()? + .into_iter() + .filter_map(|value| serde_json::from_value(value).ok()) + .collect(); + let committed: Vec<&EpochRecord> = decisions + .iter() + .filter_map(|decision| match decision { + Decision::Committed { to_epoch, .. } => Some(to_epoch), + Decision::Aborted { .. } => None, + }) + .collect(); + let agree = committed.is_empty() + || (committed.len() == decisions.len() && committed.iter().all(|e| *e == committed[0])); + if agree { + return None; } + decisions + .into_iter() + .find(|decision| matches!(decision, Decision::Committed { .. })) } fn register_path(voter: &str) -> Vec<String> { @@ -551,6 +594,7 @@ mod tests { proposal: proposal(10_000), decision: None, abort_requested: None, + contested: false, votes: BTreeMap::new(), }; assert_eq!(record.phase(10_000, 500), Phase::Proposed); @@ -564,6 +608,45 @@ mod tests { } #[test] + fn disagreeing_decisions_are_contested() { + let mut base = AutoCommit::new(); + write_proposal(&mut base, "c-1", &proposal(10_000)).unwrap(); + base.commit(); + let (mut a, mut b, mut c) = (base.fork(), base.fork(), base.fork()); + let committed = Decision::Committed { + to_epoch: EpochRecord { + generation: 1, + id: "cd".repeat(32), + }, + snapshot_bytes: 1, + decided_at: 2, + }; + write_decision(&mut a, "c-1", &committed).unwrap(); + write_decision( + &mut b, + "c-1", + &Decision::Aborted { + reason: "test".into(), + decided_at: 3, + }, + ) + .unwrap(); + // The same decision, concurrently: they agree. + write_decision(&mut c, "c-1", &committed).unwrap(); + a.commit(); + b.commit(); + c.commit(); + let mut agreeing = a.fork(); + agreeing.merge(&mut c).unwrap(); + assert!(!load(&agreeing, "c-1").unwrap().contested); + + a.merge(&mut b).unwrap(); + let record = load(&a, "c-1").unwrap(); + assert!(record.contested); + assert_eq!(record.phase(0, 0), Phase::Committed); + } + + #[test] fn unreadable_records_are_skipped() { let mut doc = AutoCommit::new(); store::write_path( diff --git a/src/remake/switch.rs b/src/remake/switch.rs index e4644fb..cf5ea75 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -105,6 +105,10 @@ async fn check<H: Host>(host: &H) { let (id, record) = match committed.as_slice() { // Nothing committed here; maybe peers moved on without us. [] => return catchup::check(host).await, + [(id, record)] if record.contested => { + error!(remake = %id, "Decisions on this remake disagree; not switching on my own"); + return catchup::check(host).await; + } [(id, record)] => (id.clone(), record.clone()), several => { // Majority voting prevents this; never guess which one is right. Peers diff --git a/src/remake/vote.rs b/src/remake/vote.rs index f8f7bbd..16de87a 100644 --- a/src/remake/vote.rs +++ b/src/remake/vote.rs @@ -321,6 +321,7 @@ mod tests { }, decision: None, abort_requested: None, + contested: false, votes: Default::default(), } } |