//! The coordinator: the node that received the proposal counts the votes (`vote`) and //! writes the decision. //! //! It commits as soon as a majority of the electorate voted for the remake. It aborts //! if, before that: //! * an operator requested an abort, //! * the local epoch changed, //! * a competing remake of the same epoch committed, //! * the voting period (`settleSecs`) passed without a majority, or //! * a majority is out of reach (too many members voted for a competing remake). //! //! 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. //! //! Committing builds the snapshot at the base heads and records the new epoch, so //! every node can verify its own build against it. //! //! The task lives only in memory. If the coordinator dies, nobody decides and the //! proposal expires; its votes stay cast (`vote`). use std::time::Duration; use tracing::{info, warn}; use super::Host; use super::content; use super::records::{self, Decision, Phase, Record}; use super::snapshot; use super::vote; use crate::document::store::StoreError; const TICK: Duration = Duration::from_secs(1); pub(crate) async fn run(host: H, id: String) { info!(remake = %id, "Coordinating remake"); loop { tokio::time::sleep(TICK).await; match step(&host, &id).await { Ok(true) => return, Ok(false) => {} Err(error) => { warn!(remake = %id, error = %format!("{error:#}"), "Coordinator step failed; retrying"); } } } } /// One evaluation. Returns true once the coordinator is done. async fn step(host: &H, id: &str) -> Result { let wanted = id.to_owned(); let (record, all, register) = host .read_doc(move |doc| { ( records::load(doc, &wanted), records::load_all(doc), records::load_register(doc), ) }) .await; let Some(record) = record else { warn!(remake = %id, "Remake record vanished; stopping coordinator"); return Ok(true); }; if record.decision.is_some() { return Ok(true); } if records::EpochRecord::from(host.current_epoch()) != record.proposal.from_epoch { decide_abort(host, id, "local epoch changed during the vote".to_owned()).await?; return Ok(true); } if let Some(reason) = obstacle(id, &record, &all) { decide_abort(host, id, reason).await?; return Ok(true); } let tally = vote::tally(id, &record, &all, ®ister); // The deadline first: a majority counted after it is too late. if records::unix_millis() >= record.proposal.settle_until { decide_abort( host, id, format!("no majority in the voting period: {tally}"), ) .await?; return Ok(true); } if tally.reached() { info!(remake = %id, votes = %tally, "Majority reached"); commit(host, id, &record).await?; return Ok(true); } if tally.out_of_reach() { decide_abort(host, id, format!("a majority is out of reach: {tally}")).await?; return Ok(true); } Ok(false) } /// Why remake `id` must be aborted whatever its votes, if it must. fn obstacle(id: &str, record: &Record, all: &[(String, Record)]) -> Option { if let Some(request) = &record.abort_requested { return Some(format!("aborted by operator via {}", request.by)); } let (other, _) = all.iter().find(|(other, candidate)| { other != id && candidate.proposal.from_epoch == record.proposal.from_epoch && candidate.committed_epoch().is_some() })?; Some(format!("competing remake {other} committed")) } async fn commit(host: &H, id: &str, record: &Record) -> Result<(), StoreError> { let Some(mut request) = record.snapshot_request(id) else { return decide_abort( host, id, "unreadable epoch or base heads in proposal".into(), ) .await; }; if let Some(reference) = &record.proposal.content { match content::ensure(host, reference).await { Ok(bytes) => request.content = Some(bytes), Err(reason) => return decide_abort(host, id, reason).await, } } // Clone under the read lock, build without any lock. let copy = host.read_doc(|doc| doc.clone()).await; let mut snapshot = match snapshot::build_detached(copy, request).await { Ok((snapshot, _copy)) => snapshot, Err(error) => { return decide_abort(host, id, format!("snapshot build failed: {error}")).await; } }; let to_epoch = snapshot.epoch; let snapshot_bytes = snapshot.doc.save().len() as u64; let decision = Decision::Committed { to_epoch: to_epoch.into(), snapshot_bytes, decided_at: records::unix_millis(), }; match decide(host, id, decision).await? { Some(Decision::Committed { .. }) => { // The local switch reuses it instead of building again. host.remake().store_prebuilt(id, snapshot); info!(remake = %id, to_epoch = %to_epoch, snapshot_bytes, "Remake committed"); } Some(Decision::Aborted { reason, .. }) => { warn!(remake = %id, reason = %reason, "Remake aborted while building its snapshot"); } None => {} } Ok(()) } async fn decide_abort(host: &H, id: &str, reason: String) -> Result<(), StoreError> { let decision = Decision::Aborted { reason: reason.clone(), decided_at: records::unix_millis(), }; if decide(host, id, decision).await?.is_some() { warn!(remake = %id, reason = %reason, "Remake aborted"); } Ok(()) } /// Writes `decision`, checked again under the write lock against the records as they /// are now (building the snapshot takes a while): a commit becomes an abort if the /// operator asked for one, a competing remake committed, the remake expired, or the /// majority is gone. /// Returns what was written; nothing if the remake is decided already or gone. async fn decide( host: &H, id: &str, decision: Decision, ) -> Result, StoreError> { let id = id.to_owned(); let grace = host.dormant_after().as_millis() as u64; host.commit_doc(move |doc| -> Result, StoreError> { let all = records::load_all(doc); let Some((_, record)) = all.iter().find(|(other, _)| *other == id) else { return Ok(None); }; if record.decision.is_some() { return Ok(None); } let decision = match decision { Decision::Committed { decided_at, .. } => { let register = records::load_register(doc); let tally = vote::tally(&id, record, &all, ®ister); let expired = record.phase(records::unix_millis(), grace) == Phase::Expired; match obstacle(&id, record, &all) { Some(reason) => Decision::Aborted { reason, decided_at }, // Others may abort it by now (`api`): too late to commit. None if expired => Decision::Aborted { reason: format!("expired while building the snapshot: {tally}"), decided_at, }, None if !tally.reached() => Decision::Aborted { reason: format!("the majority was lost: {tally}"), decided_at, }, None => decision, } } aborted @ Decision::Aborted { .. } => aborted, }; records::write_decision(doc, &id, &decision)?; Ok(Some(decision)) }) .await }