add votes register

1 files changed, 303 insertions, 89 deletions

diff --git a/src/remake/vote.rs b/src/remake/vote.rs
index 4416293..f8f7bbd 100644
--- a/src/remake/vote.rs
+++ b/src/remake/vote.rs
@@ -3,20 +3,24 @@
 //!
 //! 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.
+//! the coordinator always included.
 //!
-//! Votes live in the record (`votes/<pk>`, written only by that member). 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.
+//! 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.
 //!
-//! 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.
+//! 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;
 
@@ -24,7 +28,7 @@ use tracing::{info, warn};
 
 use super::Host;
 use super::content;
-use super::records::{self, EpochRecord, Phase, Record, Vote, unix_millis};
+use super::records::{self, Decision, EpochRecord, Phase, Record, Register, Vote, unix_millis};
 use crate::document::store::StoreError;
 
 /// A majority of `electorate` members.
@@ -39,8 +43,10 @@ pub(crate) struct Tally {
     pub(crate) quorum: usize,
     /// Voted for it.
     pub(crate) yes: usize,
-    /// Voted for a competing open remake.
+    /// 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,
 }
@@ -60,14 +66,51 @@ 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
-        )
+            "{} 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 open(record: &Record, now: u64, grace: u64) -> bool {
-    record.phase(now, grace).is_open()
+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.
@@ -75,8 +118,7 @@ pub(crate) fn tally(
     id: &str,
     record: &Record,
     all: &[(String, Record)],
-    now: u64,
-    grace: u64,
+    register: &Register,
 ) -> Tally {
     let electorate: HashSet<&str> = record
         .proposal
@@ -84,51 +126,95 @@ pub(crate) fn tally(
         .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();
+    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: yes.len(),
-        elsewhere: elsewhere.len(),
-        pending: electorate.len() - yes.len() - elsewhere.len(),
+        yes,
+        elsewhere,
+        twice,
+        pending: electorate.len() - yes - elsewhere - twice,
     }
 }
 
-/// 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.
+/// 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 of_epoch = || {
-        all.iter()
-            .filter(move |(_, record)| &record.proposal.from_epoch == epoch)
+    let only = match freedom(me, epoch, all, register, started) {
+        Freedom::Any => None,
+        Freedom::Only(id) => Some(id),
+        Freedom::Held(_) => return None,
     };
-    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
+    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))
@@ -145,13 +231,23 @@ pub(super) async fn cast<H: Host>(host: &H) {
     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 id = choice(&voter, &wanted, &all, unix_millis(), grace)?;
+            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)
@@ -175,11 +271,17 @@ pub(super) async fn cast<H: Host>(host: &H) {
         .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).filter(|id| *id == chosen)
+            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);
             };
-            records::write_vote(doc, &id, &voter, &Vote { at: now })?;
+            let vote = Vote {
+                remake: id.clone(),
+                at: now,
+            };
+            records::write_vote(doc, &voter, &vote)?;
             Ok(Some(id))
         })
         .await;
@@ -194,11 +296,10 @@ pub(super) async fn cast<H: Host>(host: &H) {
 mod tests {
     use super::*;
     use crate::document::replica::Epoch;
-    use crate::remake::records::{Decision, Policy, Proposal};
+    use crate::remake::records::{Policy, Proposal};
     use serde_json::Value as JsonValue;
-    use std::collections::BTreeMap;
 
-    fn record(proposed_at: u64, electorate: &[&str], votes: &[&str]) -> Record {
+    fn record(proposed_at: u64, electorate: &[&str]) -> Record {
         Record {
             proposal: Proposal {
                 coordinator: electorate[0].to_owned(),
@@ -220,14 +321,42 @@ mod tests {
             },
             decision: None,
             abort_requested: None,
-            votes: votes
-                .iter()
-                .map(|v| ((*v).to_owned(), Vote { at: proposed_at }))
-                .collect::<BTreeMap<_, _>>(),
+            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() {
@@ -241,60 +370,145 @@ mod tests {
     #[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
+            ("x".to_owned(), record(100, &ALL)),
+            ("y".to_owned(), record(200, &ALL)),
         ];
-        let x = tally("x", &all[0].1, &all, 150, 0);
+        // 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, 150, 0);
+        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 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());
+        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 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"])),
+            ("x".to_owned(), aborted(record(100, &ALL))),
+            ("y".to_owned(), record(200, &ALL)),
         ];
-        let y = tally("y", &all[1].1, &all, 250, 0);
+        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", &Epoch::GENESIS.into(), &all, 250, 0).as_deref(),
+            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, &[])),
+            ("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 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 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")
+        );
 
-        let voted = vec![
-            ("y".to_owned(), record(200, &ALL, &["a"])),
-            ("x".to_owned(), record(100, &ALL, &[])),
+        // 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!(choice("a", &epoch, &voted, 150, 0), None);
+        assert_eq!(freedom("a", &genesis(), &all, &none, started), Freedom::Any);
     }
 }