summaryrefslogtreecommitdiffstats
path: root/src/remake/vote.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/remake/vote.rs')
-rw-r--r--src/remake/vote.rs515
1 files changed, 515 insertions, 0 deletions
diff --git a/src/remake/vote.rs b/src/remake/vote.rs
new file mode 100644
index 0000000..16de87a
--- /dev/null
+++ b/src/remake/vote.rs
@@ -0,0 +1,515 @@
+//! Votes on remakes: a remake commits only once a majority of its electorate voted for
+//! it.
+//!
+//! The electorate is fixed in the proposal: the members that renewed their alive stamp
+//! within the coordinator's `--electorate-window-days` (`swarm::members::electorate`),
+//! the coordinator always included.
+//!
+//! Votes live in the votes register (`_continuum/votes/<pk>`, written only by that
+//! member): one entry per member for the whole epoch (the snapshot leaves it behind).
+//! A member votes for at most one remake of its epoch, the first it sees (the oldest
+//! proposal, if it sees several at once), and votes again only once that remake is
+//! aborted. Neither expiry nor a deleted record frees a vote. Two competing remakes,
+//! even in different partitions, can't both reach a majority, so at most one commits.
+//!
+//! A node votes once it is ready, on its current epoch, only if it is in the electorate,
+//! and only once it holds the remake's content (`content`), so a committed remake has
+//! its content on a majority.
+//!
+//! A vote can be lost: a node whose vote reached only the coordinator restarts and
+//! resyncs from nodes that never saw it. Should it vote again for a competitor, its
+//! entry ends up with two concurrent values, and it counts for no remake. To avoid
+//! that, a node doesn't vote while more than one undecided remake it could have voted
+//! for is older than its start, and with one, votes only for that one.
+
+use std::collections::HashSet;
+
+use tracing::{info, warn};
+
+use super::Host;
+use super::content;
+use super::records::{self, Decision, EpochRecord, Phase, Record, Register, Vote, unix_millis};
+use crate::document::store::StoreError;
+
+/// A majority of `electorate` members.
+pub(crate) fn quorum(electorate: usize) -> usize {
+ electorate / 2 + 1
+}
+
+/// Where a remake stands with its electorate.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct Tally {
+ pub(crate) electorate: usize,
+ pub(crate) quorum: usize,
+ /// Voted for it.
+ pub(crate) yes: usize,
+ /// Voted for a competing remake that isn't aborted.
+ pub(crate) elsewhere: usize,
+ /// Voted for two remakes: counts for neither.
+ pub(crate) twice: usize,
+ /// Not voted (yet).
+ pub(crate) pending: usize,
+}
+
+impl Tally {
+ pub(crate) fn reached(&self) -> bool {
+ self.yes >= self.quorum
+ }
+
+ /// Even if every pending member votes for it, it falls short.
+ pub(crate) fn out_of_reach(&self) -> bool {
+ self.yes + self.pending < self.quorum
+ }
+}
+
+impl std::fmt::Display for Tally {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{} of {} voted for it, {} for a competing remake, ",
+ self.yes, self.electorate, self.elsewhere
+ )?;
+ if self.twice > 0 {
+ write!(f, "{} for two, ", self.twice)?;
+ }
+ write!(f, "{} not yet; {} needed", self.pending, self.quorum)
+ }
+}
+
+/// What a member's entry in the register amounts to.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum Held<'a> {
+ /// No vote, or only for aborted remakes.
+ Free,
+ For(&'a str),
+ /// For two remakes, neither aborted.
+ Twice,
+}
+
+/// What `votes` (one member's entry) amount to among `all` records.
+pub(crate) fn held<'a>(votes: &'a [Vote], all: &[(String, Record)]) -> Held<'a> {
+ let aborted = |remake: &str| {
+ all.iter().any(|(id, record)| {
+ id == remake && matches!(record.decision, Some(Decision::Aborted { .. }))
+ })
+ };
+ let mut kept = votes
+ .iter()
+ .map(|vote| vote.remake.as_str())
+ .filter(|remake| !aborted(remake));
+ let Some(first) = kept.next() else {
+ return Held::Free;
+ };
+ if kept.all(|other| other == first) {
+ Held::For(first)
+ } else {
+ Held::Twice
+ }
+}
+
+fn held_by<'a>(member: &str, register: &'a Register, all: &[(String, Record)]) -> Held<'a> {
+ register
+ .get(member)
+ .map_or(Held::Free, |votes| held(votes, all))
+}
+
+/// The tally of remake `id` among `all` records.
+pub(crate) fn tally(
+ id: &str,
+ record: &Record,
+ all: &[(String, Record)],
+ register: &Register,
+) -> Tally {
+ let electorate: HashSet<&str> = record
+ .proposal
+ .electorate
+ .iter()
+ .map(String::as_str)
+ .collect();
+ let (mut yes, mut elsewhere, mut twice) = (0, 0, 0);
+ for member in &electorate {
+ match held_by(member, register, all) {
+ Held::For(remake) if remake == id => yes += 1,
+ Held::For(_) => elsewhere += 1,
+ Held::Twice => twice += 1,
+ Held::Free => {}
+ }
+ }
+ Tally {
+ electorate: electorate.len(),
+ quorum: record.proposal.quorum,
+ yes,
+ elsewhere,
+ twice,
+ pending: electorate.len() - yes - elsewhere - twice,
+ }
+}
+
+/// What `me` may vote for in `epoch`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum Freedom {
+ Any,
+ /// Only this remake: the one it may have voted for before its start.
+ Only(String),
+ /// Nothing, for this reason.
+ Held(String),
+}
+
+/// What `me` may vote for in `epoch`. Remakes proposed before `started` (this node's
+/// start, plus a margin for clocks) may hold a vote of ours that we lost in a restart.
+pub(crate) fn freedom(
+ me: &str,
+ epoch: &EpochRecord,
+ all: &[(String, Record)],
+ register: &Register,
+ started: u64,
+) -> Freedom {
+ match held_by(me, register, all) {
+ Held::Free => {}
+ Held::For(remake) => {
+ return Freedom::Held(format!("this node voted for remake {remake}"));
+ }
+ Held::Twice => {
+ return Freedom::Held("this node voted for two remakes".to_owned());
+ }
+ }
+ let mut older = all
+ .iter()
+ .filter(|(_, record)| {
+ &record.proposal.from_epoch == epoch
+ && record.decision.is_none()
+ && record.proposal.proposed_at <= started
+ && record.proposal.electorate.iter().any(|member| member == me)
+ })
+ .map(|(id, _)| id);
+ match (older.next(), older.next()) {
+ (None, _) => Freedom::Any,
+ (Some(id), None) => Freedom::Only(id.clone()),
+ (Some(_), Some(_)) => Freedom::Held(
+ "several undecided remakes are older than this node's start; it may have voted \
+ for one before"
+ .to_owned(),
+ ),
+ }
+}
+
+/// The remake `me` should vote for now, if any: the oldest proposed one of `epoch`
+/// whose electorate includes it, as far as it is free to vote (`freedom`).
+pub(crate) fn choice(
+ me: &str,
+ epoch: &EpochRecord,
+ all: &[(String, Record)],
+ register: &Register,
+ started: u64,
+ now: u64,
+ grace: u64,
+) -> Option<String> {
+ let only = match freedom(me, epoch, all, register, started) {
+ Freedom::Any => None,
+ Freedom::Only(id) => Some(id),
+ Freedom::Held(_) => return None,
+ };
+ all.iter()
+ .filter(|(id, record)| {
+ &record.proposal.from_epoch == epoch
+ && record.phase(now, grace) == Phase::Proposed
+ && record.proposal.electorate.iter().any(|member| member == me)
+ && only.as_ref().is_none_or(|only| only == id)
+ })
+ .min_by(|(a_id, a), (b_id, b)| {
+ (a.proposal.proposed_at, a_id).cmp(&(b.proposal.proposed_at, b_id))
+ })
+ .map(|(id, _)| id.clone())
+}
+
+/// Votes for a remake of the local epoch if this node should. Called by the remake
+/// worker whenever the records change (and on its timer).
+pub(super) async fn cast<H: Host>(host: &H) {
+ if !host.is_node_ready().await {
+ return;
+ }
+ let me = host.node_id();
+ let epoch = EpochRecord::from(host.current_epoch());
+ let grace = host.dormant_after().as_millis() as u64;
+ let started = host.remake().started_at() + grace;
+
+ // Usually there is nothing to vote on: look first, without writing.
+ let (voter, wanted) = (me.clone(), epoch.clone());
+ let due = host
+ .read_doc(move |doc| {
+ let all = records::load_all(doc);
+ let register = records::load_register(doc);
+ let id = choice(
+ &voter,
+ &wanted,
+ &all,
+ &register,
+ started,
+ unix_millis(),
+ grace,
+ )?;
+ let content = all
+ .into_iter()
+ .find(|(other, _)| *other == id)
+ .and_then(|(_, record)| record.proposal.content);
+ Some((id, content))
+ })
+ .await;
+ let Some((chosen, content)) = due else {
+ return;
+ };
+ // Only with the content in hand: a committed remake must have it on a majority.
+ if let Some(reference) = &content {
+ if content::available_or_fetch(host, reference).is_none() {
+ return;
+ }
+ }
+
+ // Decided again under the write lock, so two passes can't both vote.
+ let voter = me.clone();
+ let result = host
+ .commit_doc(move |doc| -> Result<Option<String>, StoreError> {
+ let now = unix_millis();
+ let all = records::load_all(doc);
+ let register = records::load_register(doc);
+ let Some(id) = choice(&voter, &epoch, &all, &register, started, now, grace)
+ .filter(|id| *id == chosen)
+ else {
+ return Ok(None);
+ };
+ let vote = Vote {
+ remake: id.clone(),
+ at: now,
+ };
+ records::write_vote(doc, &voter, &vote)?;
+ Ok(Some(id))
+ })
+ .await;
+ match result {
+ Ok(Some(id)) => info!(remake = %id, "Voted for remake"),
+ Ok(None) => {}
+ Err(error) => warn!(error = %format!("{error:#}"), "Failed to vote on a remake"),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::document::replica::Epoch;
+ use crate::remake::records::{Policy, Proposal};
+ use serde_json::Value as JsonValue;
+
+ fn record(proposed_at: u64, electorate: &[&str]) -> Record {
+ Record {
+ proposal: Proposal {
+ coordinator: electorate[0].to_owned(),
+ from_epoch: Epoch::GENESIS.into(),
+ base_heads: Vec::new(),
+ policy: Policy {
+ replay_window: Policy::default_replay_window(),
+ archive_retention: Policy::default_archive_retention(),
+ },
+ settle_ms: 1_000,
+ proposed_at,
+ settle_until: proposed_at + 1_000,
+ switch_delay_ms: None,
+ electorate: electorate.iter().map(|m| (*m).to_owned()).collect(),
+ quorum: quorum(electorate.len()),
+ kind: "compact".to_owned(),
+ content: None,
+ translation: JsonValue::Null,
+ },
+ decision: None,
+ abort_requested: None,
+ contested: false,
+ votes: Default::default(),
+ }
+ }
+
+ /// A register from (voter, remakes voted for) pairs.
+ fn register(votes: &[(&str, &[&str])]) -> Register {
+ votes
+ .iter()
+ .map(|(voter, remakes)| {
+ let votes = remakes
+ .iter()
+ .map(|remake| Vote {
+ remake: (*remake).to_owned(),
+ at: 0,
+ })
+ .collect();
+ ((*voter).to_owned(), votes)
+ })
+ .collect()
+ }
+
+ fn aborted(mut record: Record) -> Record {
+ record.decision = Some(Decision::Aborted {
+ reason: "test".into(),
+ decided_at: 150,
+ });
+ record
+ }
+
+ const ALL: [&str; 5] = ["a", "b", "c", "d", "e"];
+ /// Long before any proposal: no restart concerns.
+ const STARTED: u64 = 0;
+
+ fn genesis() -> EpochRecord {
+ Epoch::GENESIS.into()
+ }
+
+ #[test]
+ fn majorities() {
+ assert_eq!(quorum(1), 1);
+ assert_eq!(quorum(2), 2);
+ assert_eq!(quorum(3), 2);
+ assert_eq!(quorum(4), 3);
+ assert_eq!(quorum(250), 126);
+ }
+
+ #[test]
+ fn competing_remakes_split_the_vote() {
+ let all = vec![
+ ("x".to_owned(), record(100, &ALL)),
+ ("y".to_owned(), record(200, &ALL)),
+ ];
+ // z isn't in the electorate.
+ let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["y"]), ("z", &["y"])]);
+ let x = tally("x", &all[0].1, &all, &votes);
+ assert_eq!((x.yes, x.elsewhere, x.pending, x.quorum), (2, 1, 2, 3));
+ assert!(!x.reached() && !x.out_of_reach());
+
+ let y = tally("y", &all[1].1, &all, &votes);
+ assert_eq!((y.yes, y.elsewhere, y.pending), (1, 2, 2));
+ assert!(!y.out_of_reach());
+
+ // d votes for x: x has its majority, y can't get one any more.
+ let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["y"]), ("d", &["x"])]);
+ assert!(tally("x", &all[0].1, &all, &votes).reached());
+ assert!(tally("y", &all[1].1, &all, &votes).out_of_reach());
+ }
+
+ #[test]
+ fn votes_for_aborted_remakes_are_void() {
+ let all = vec![
+ ("x".to_owned(), aborted(record(100, &ALL))),
+ ("y".to_owned(), record(200, &ALL)),
+ ];
+ let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["x"]), ("d", &["y"])]);
+ let y = tally("y", &all[1].1, &all, &votes);
+ assert_eq!((y.yes, y.elsewhere, y.pending), (1, 0, 4));
+ // c may vote again.
+ assert_eq!(
+ choice("c", &genesis(), &all, &votes, STARTED, 250, 0).as_deref(),
+ Some("y")
+ );
+ }
+
+ #[test]
+ fn expiry_frees_no_vote() {
+ // x's coordinator is gone: x expired, but a, b and c still hold their votes.
+ let all = vec![
+ ("x".to_owned(), record(100, &ALL)),
+ ("y".to_owned(), record(5_000, &ALL)),
+ ];
+ let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["x"])]);
+ assert_eq!(all[0].1.phase(5_000, 0), Phase::Expired);
+ assert_eq!(
+ choice("a", &genesis(), &all, &votes, STARTED, 5_000, 0),
+ None
+ );
+ let y = tally("y", &all[1].1, &all, &votes);
+ assert_eq!((y.yes, y.elsewhere, y.pending), (0, 3, 2));
+ assert!(y.out_of_reach());
+ }
+
+ #[test]
+ fn voting_twice_counts_for_neither() {
+ let all = vec![
+ ("x".to_owned(), record(100, &ALL)),
+ ("y".to_owned(), record(200, &ALL)),
+ ];
+ let votes = register(&[("a", &["x", "y"]), ("b", &["x"]), ("c", &["x", "x"])]);
+ let x = tally("x", &all[0].1, &all, &votes);
+ assert_eq!((x.yes, x.elsewhere, x.twice, x.pending), (2, 0, 1, 2));
+ assert_eq!(
+ x.to_string(),
+ "2 of 5 voted for it, 0 for a competing remake, 1 for two, 2 not yet; 3 needed"
+ );
+ assert!(matches!(
+ freedom("a", &genesis(), &all, &votes, STARTED),
+ Freedom::Held(_)
+ ));
+
+ // Once y is aborted, a's vote for x stands.
+ let all = vec![
+ ("x".to_owned(), record(100, &ALL)),
+ ("y".to_owned(), aborted(record(200, &ALL))),
+ ];
+ assert_eq!(tally("x", &all[0].1, &all, &votes).yes, 3);
+ }
+
+ #[test]
+ fn one_vote_per_epoch_for_the_oldest() {
+ let all = vec![
+ ("y".to_owned(), record(200, &ALL)),
+ ("x".to_owned(), record(100, &ALL)),
+ ];
+ let none = register(&[]);
+ assert_eq!(
+ choice("a", &genesis(), &all, &none, STARTED, 150, 0).as_deref(),
+ Some("x")
+ );
+ assert_eq!(
+ choice("outsider", &genesis(), &all, &none, STARTED, 150, 0),
+ None
+ );
+
+ let voted = register(&[("a", &["y"])]);
+ assert_eq!(choice("a", &genesis(), &all, &voted, STARTED, 150, 0), None);
+ }
+
+ #[test]
+ fn after_a_start_only_the_remake_voted_for_before() {
+ let started = 150;
+ // Proposed before the start: a may have voted for x then.
+ let all = vec![
+ ("x".to_owned(), record(100, &ALL)),
+ ("y".to_owned(), record(200, &ALL)),
+ ];
+ let none = register(&[]);
+ assert_eq!(
+ freedom("a", &genesis(), &all, &none, started),
+ Freedom::Only("x".to_owned())
+ );
+ assert_eq!(
+ choice("a", &genesis(), &all, &none, started, 250, 1_000).as_deref(),
+ Some("x")
+ );
+
+ // Two of them: a may have voted for either, so for none.
+ let all = vec![
+ ("w".to_owned(), record(50, &ALL)),
+ ("x".to_owned(), record(100, &ALL)),
+ ("y".to_owned(), record(200, &ALL)),
+ ];
+ assert!(matches!(
+ freedom("a", &genesis(), &all, &none, started),
+ Freedom::Held(_)
+ ));
+ assert_eq!(
+ choice("a", &genesis(), &all, &none, started, 250, 1_000),
+ None
+ );
+
+ // Once they are decided, a is free again; it couldn't vote for an outsider's.
+ let all = vec![
+ ("w".to_owned(), aborted(record(50, &ALL))),
+ ("x".to_owned(), aborted(record(100, &ALL))),
+ ("y".to_owned(), record(200, &ALL)),
+ ("z".to_owned(), record(120, &["b", "c"])),
+ ];
+ assert_eq!(freedom("a", &genesis(), &all, &none, started), Freedom::Any);
+ }
+}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-24 23:30:18 +0000