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.rs291
1 files changed, 291 insertions, 0 deletions
diff --git a/src/remake/vote.rs b/src/remake/vote.rs
new file mode 100644
index 0000000..d73326d
--- /dev/null
+++ b/src/remake/vote.rs
@@ -0,0 +1,291 @@
+//! 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. Each member votes for at most one open remake of
+//! its epoch: the first it sees (the oldest proposal, if it sees several at once). Two
+//! competing remakes, even in different partitions, can't both reach a majority, so at
+//! most one commits. A vote for a remake that aborted or expired is void; the member
+//! may vote again.
+//!
+//! Votes live in the record (`votes/<pk>`, written only by that member). A node votes
+//! once it is ready, on its current epoch, and only if it is in the electorate.
+//!
+//! Gap: votes are not persisted locally. A node whose vote reached only the coordinator
+//! and that then restarts and resyncs from nodes that never saw the vote may vote again
+//! for a competitor.
+
+use std::collections::HashSet;
+
+use tracing::{info, warn};
+
+use super::Host;
+use super::records::{self, EpochRecord, Phase, Record, 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 open remake.
+ pub(crate) elsewhere: 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, {} not yet; {} needed",
+ self.yes, self.electorate, self.elsewhere, self.pending, self.quorum
+ )
+ }
+}
+
+fn open(record: &Record, now: u64, grace: u64) -> bool {
+ record.phase(now, grace).is_open()
+}
+
+/// The tally of remake `id` among `all` records.
+pub(crate) fn tally(
+ id: &str,
+ record: &Record,
+ all: &[(String, Record)],
+ now: u64,
+ grace: u64,
+) -> Tally {
+ let electorate: HashSet<&str> = record
+ .proposal
+ .electorate
+ .iter()
+ .map(String::as_str)
+ .collect();
+ let yes: HashSet<&str> = record
+ .votes
+ .keys()
+ .map(String::as_str)
+ .filter(|voter| electorate.contains(voter))
+ .collect();
+ let elsewhere: HashSet<&str> = all
+ .iter()
+ .filter(|(other, candidate)| {
+ other != id
+ && candidate.proposal.from_epoch == record.proposal.from_epoch
+ && open(candidate, now, grace)
+ })
+ .flat_map(|(_, candidate)| candidate.votes.keys().map(String::as_str))
+ .filter(|voter| electorate.contains(voter) && !yes.contains(voter))
+ .collect();
+ Tally {
+ electorate: electorate.len(),
+ quorum: record.proposal.quorum,
+ yes: yes.len(),
+ elsewhere: elsewhere.len(),
+ pending: electorate.len() - yes.len() - elsewhere.len(),
+ }
+}
+
+/// The remake `me` should vote for now, if any: none if it already voted for an open
+/// remake of `epoch`; otherwise the oldest proposed one whose electorate includes it.
+pub(crate) fn choice(
+ me: &str,
+ epoch: &EpochRecord,
+ all: &[(String, Record)],
+ now: u64,
+ grace: u64,
+) -> Option<String> {
+ let of_epoch = || {
+ all.iter()
+ .filter(move |(_, record)| &record.proposal.from_epoch == epoch)
+ };
+ if of_epoch().any(|(_, record)| open(record, now, grace) && record.votes.contains_key(me)) {
+ return None;
+ }
+ of_epoch()
+ .filter(|(_, record)| {
+ record.phase(now, grace) == Phase::Proposed
+ && record.proposal.electorate.iter().any(|member| member == me)
+ })
+ .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;
+
+ // 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| {
+ choice(
+ &voter,
+ &wanted,
+ &records::load_all(doc),
+ unix_millis(),
+ grace,
+ )
+ .is_some()
+ })
+ .await;
+ if !due {
+ 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 Some(id) = choice(&voter, &epoch, &all, now, grace) else {
+ return Ok(None);
+ };
+ records::write_vote(doc, &id, &voter, &Vote { at: now })?;
+ 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::{Decision, Policy, Proposal};
+ use serde_json::Value as JsonValue;
+ use std::collections::BTreeMap;
+
+ fn record(proposed_at: u64, electorate: &[&str], votes: &[&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: JsonValue::Null,
+ translation: JsonValue::Null,
+ },
+ decision: None,
+ abort_requested: None,
+ votes: votes
+ .iter()
+ .map(|v| ((*v).to_owned(), Vote { at: proposed_at }))
+ .collect::<BTreeMap<_, _>>(),
+ }
+ }
+
+ const ALL: [&str; 5] = ["a", "b", "c", "d", "e"];
+
+ #[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, &["a", "b"])),
+ ("y".to_owned(), record(200, &ALL, &["c", "z"])), // z isn't in the electorate
+ ];
+ let x = tally("x", &all[0].1, &all, 150, 0);
+ 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, 150, 0);
+ 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 all = vec![
+ ("x".to_owned(), record(100, &ALL, &["a", "b", "d"])),
+ ("y".to_owned(), record(200, &ALL, &["c"])),
+ ];
+ assert!(tally("x", &all[0].1, &all, 150, 0).reached());
+ assert!(tally("y", &all[1].1, &all, 150, 0).out_of_reach());
+ }
+
+ #[test]
+ fn votes_for_aborted_remakes_are_void() {
+ let mut aborted = record(100, &ALL, &["a", "b", "c"]);
+ aborted.decision = Some(Decision::Aborted {
+ reason: "test".into(),
+ decided_at: 150,
+ });
+ let all = vec![
+ ("x".to_owned(), aborted),
+ ("y".to_owned(), record(200, &ALL, &["d"])),
+ ];
+ let y = tally("y", &all[1].1, &all, 250, 0);
+ assert_eq!((y.yes, y.elsewhere, y.pending), (1, 0, 4));
+ // c may vote again.
+ assert_eq!(
+ choice("c", &Epoch::GENESIS.into(), &all, 250, 0).as_deref(),
+ Some("y")
+ );
+ }
+
+ #[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 epoch: EpochRecord = Epoch::GENESIS.into();
+ assert_eq!(choice("a", &epoch, &all, 150, 0).as_deref(), Some("x"));
+ assert_eq!(choice("outsider", &epoch, &all, 150, 0), None);
+
+ let voted = vec![
+ ("y".to_owned(), record(200, &ALL, &["a"])),
+ ("x".to_owned(), record(100, &ALL, &[])),
+ ];
+ assert_eq!(choice("a", &epoch, &voted, 150, 0), None);
+ }
+}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-24 21:38:35 +0000