//! Proposing a remake: the checks (one open remake per epoch, the proposer free to //! vote, a known base), superseding expired remakes, the electorate and quorum, and the //! proposal and own vote written in one change; then its coordinator. And when a node //! other than the coordinator may abort an expired remake. use std::collections::HashSet; use tracing::info; use super::coordinator; use super::records::{self, Decision, Phase, Proposal, Record, Vote}; use super::{Host, kind, vote}; use crate::clock::UnixMillis; use crate::document::ChangeHash; use crate::document::store::StoreError; use crate::force::Force; /// 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. pub(super) fn abortable_at(record: &Record, grace: u64) -> UnixMillis { record .proposal .settle_until .plus_ms(record.grace(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). pub(super) fn abort_obstacle( record: &Record, me: &str, active_peers: &HashSet, force: &Force, now: UnixMillis, 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.ms_since(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. pub(super) async fn active_peers(host: &H) -> HashSet { 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)] pub(super) struct ExpiredRemake { pub(super) id: String, pub(super) coordinator: String, pub(super) coordinator_active: bool, /// When it expired. pub(super) since: UnixMillis, /// When another node may abort it (`abortable_at`). pub(super) abortable_at: UnixMillis, pub(super) tally: vote::Tally, } /// Superseding expired remakes, as the proposal's `force` asks. pub(super) struct Supersede { pub(super) force: Force, pub(super) active_peers: HashSet, } #[derive(Debug, thiserror::Error)] pub(super) enum StartError { #[error("remake {0} is already open for this epoch")] Open(String), /// 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, /// 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), #[error("base head {0} is not in this node's document")] UnknownBase(String), #[error(transparent)] Store(#[from] StoreError), } /// Writes the proposal (unless another one is open for the epoch) and starts its /// coordinator. Runs as its own task, independent of the HTTP request. pub(super) async fn propose( host: H, id: String, mut proposal: Proposal, base: Option>, supersede: Supersede, now: UnixMillis, grace: u64, ) -> Result { let record_id = id.clone(); let electorate_host = host.clone(); let me = host.node_id(); let started = host.remake().started_at().plus_ms(grace); let base_at_commit = kind::by_name(&proposal.kind).is_some_and(|kind| kind.base_at_commit()); let record = host .commit_doc(move |doc| -> Result { 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 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() { 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.plus_ms(record.grace(grace)), abortable_at: abortable_at(record, grace), tally: vote::tally(id, record, &all, ®ister), }) .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}"), 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 // proposal itself is never in it); or, for a compaction, left to the commit. if !base_at_commit { let base = match base { Some(base) => { if let Some(unknown) = crate::document::changes::missing(doc, &base).first() { return Err(StartError::UnknownBase(unknown.to_string())); } base } None => doc.get_heads(), }; proposal.base_heads = Some(records::encode_heads(&base)); } // 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?; tokio::spawn(coordinator::run(host, id)); Ok(record) }