introduce majority voting on remakes

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);
+    }
+}