1 files changed, 110 insertions, 21 deletions

diff --git a/src/remake/records.rs b/src/remake/records.rs
index e3a0a7c..37af491 100644
--- a/src/remake/records.rs
+++ b/src/remake/records.rs
@@ -7,7 +7,10 @@
 //! * `decision`: written once, by the coordinator only (committed or aborted).
 //! * `abortRequested`: written by any node on an operator's request. It is an input to
 //!   the coordinator's decision, not a decision itself.
-//! * `votes/<pk>`: each written by that member only (`vote`).
+//!
+//! Votes are kept apart, in the votes register `_continuum/votes/<pk>`: one entry per
+//! member for the whole epoch, written by that member only (`vote`). Reading a record
+//! fills in the votes that name it.
 //!
 //! A proposal whose coordinator never decides (crash, partition) is reported as
 //! `expired` once its settle period plus a grace margin has passed. Expiry is derived,
@@ -24,7 +27,7 @@ use serde_json::Value as JsonValue;
 use tracing::warn;
 
 use super::kind::DEFAULT_KIND;
-use super::snapshot::{META_ROOT, REMAKES, SnapshotRequest};
+use super::snapshot::{META_ROOT, REMAKES, SnapshotRequest, VOTES};
 use crate::document::replica::Epoch;
 use crate::document::store::{self, StoreError};
 
@@ -201,13 +204,18 @@ pub(crate) struct AbortRequest {
     pub(crate) at: u64,
 }
 
-/// A member's vote for a remake.
-#[derive(Clone, Debug, Serialize, Deserialize)]
+/// A member's vote: its entry in the votes register.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct Vote {
+    pub(crate) remake: String,
     pub(crate) at: u64,
 }
 
+/// The votes register: every concurrent value of each member's entry, by member public
+/// key. Two values mean the member voted again without seeing its first vote (`vote`).
+pub(crate) type Register = BTreeMap<String, Vec<Vote>>;
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct Record {
@@ -216,8 +224,9 @@ pub(crate) struct Record {
     pub(crate) decision: Option<Decision>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub(crate) abort_requested: Option<AbortRequest>,
-    /// By member public key.
-    #[serde(default)]
+    /// The register's votes naming this remake, by member public key. Not stored in
+    /// the record: filled in when it is read (`load`).
+    #[serde(default, skip_deserializing)]
     pub(crate) votes: BTreeMap<String, Vote>,
 }
 
@@ -292,8 +301,37 @@ pub(crate) fn record_path(id: &str, part: &str) -> Vec<String> {
     path
 }
 
-/// The record with this ID, if it exists and is readable.
+/// The record with this ID, if it exists and is readable, with its votes.
 pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> {
+    let record = load_record(doc, id)?;
+    Some(with_votes(id, record, &load_register(doc)))
+}
+
+/// All readable records, with their votes.
+pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> {
+    let register = load_register(doc);
+    store::keys_at(doc, &records_dir())
+        .unwrap_or_default()
+        .into_iter()
+        .filter_map(|id| {
+            let record = load_record(doc, &id)?;
+            Some((id.clone(), with_votes(&id, record, &register)))
+        })
+        .collect()
+}
+
+fn with_votes(id: &str, mut record: Record, register: &Register) -> Record {
+    record.votes = register
+        .iter()
+        .filter_map(|(voter, votes)| {
+            let vote = votes.iter().find(|vote| vote.remake == id)?;
+            Some((voter.clone(), vote.clone()))
+        })
+        .collect();
+    record
+}
+
+fn load_record(doc: &AutoCommit, id: &str) -> Option<Record> {
     let value = match store::read_path(doc, &record_path(id, "")) {
         Ok(Some(value)) => value,
         Ok(None) => return None,
@@ -311,12 +349,35 @@ pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> {
     }
 }
 
-/// All readable records.
-pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> {
-    store::keys_at(doc, &records_dir())
+fn register_path(voter: &str) -> Vec<String> {
+    vec![META_ROOT.to_owned(), VOTES.to_owned(), voter.to_owned()]
+}
+
+/// The votes register; unreadable entries are skipped with a warning.
+pub(crate) fn load_register(doc: &AutoCommit) -> Register {
+    store::keys_at(doc, &[META_ROOT.to_owned(), VOTES.to_owned()])
         .unwrap_or_default()
         .into_iter()
-        .filter_map(|id| load(doc, &id).map(|record| (id, record)))
+        .filter_map(|voter| {
+            let values = match store::read_candidates(doc, &register_path(&voter)) {
+                Ok(values) => values,
+                Err(error) => {
+                    warn!(voter = %voter, error = %error, "Failed to read vote");
+                    return None;
+                }
+            };
+            let votes: Vec<Vote> = values
+                .into_iter()
+                .filter_map(|value| match serde_json::from_value(value) {
+                    Ok(vote) => Some(vote),
+                    Err(error) => {
+                        warn!(voter = %voter, error = %error, "Skipping unreadable vote");
+                        None
+                    }
+                })
+                .collect();
+            (!votes.is_empty()).then_some((voter, votes))
+        })
         .collect()
 }
 
@@ -354,16 +415,10 @@ pub(crate) fn write_abort_request(
     write_part(doc, id, "abortRequested", request)
 }
 
-pub(crate) fn write_vote(
-    doc: &mut AutoCommit,
-    id: &str,
-    voter: &str,
-    vote: &Vote,
-) -> Result<(), StoreError> {
+/// Writes `voter`'s entry in the votes register (replacing its earlier one).
+pub(crate) fn write_vote(doc: &mut AutoCommit, voter: &str, vote: &Vote) -> Result<(), StoreError> {
     let json = serde_json::to_value(vote).map_err(anyhow::Error::from)?;
-    let mut path = record_path(id, "votes");
-    path.push(voter.to_owned());
-    store::write_path(doc, &path, &json)
+    store::write_path(doc, &register_path(voter), &json)
 }
 
 pub(crate) fn decode_hex32(hex: &str) -> Option<[u8; 32]> {
@@ -441,7 +496,12 @@ mod tests {
             decided_at: 11_000,
         };
         write_decision(&mut doc, "c-1", &decision).unwrap();
-        write_vote(&mut doc, "c-1", "node-b", &Vote { at: 7 }).unwrap();
+        let vote = |remake: &str| Vote {
+            remake: remake.into(),
+            at: 7,
+        };
+        write_vote(&mut doc, "node-b", &vote("c-1")).unwrap();
+        write_vote(&mut doc, "node-c", &vote("other")).unwrap();
         doc.commit();
 
         let all = load_all(&doc);
@@ -457,6 +517,35 @@ mod tests {
     }
 
     #[test]
+    fn concurrent_votes_are_kept_apart() {
+        let mut a = AutoCommit::new();
+        let vote = |remake: &str, at| Vote {
+            remake: remake.into(),
+            at,
+        };
+        write_vote(&mut a, "node-a", &vote("x", 1)).unwrap();
+        a.commit();
+        let mut b = a.fork();
+        // node-b votes in both, never seeing its other vote (a lost vote).
+        write_vote(&mut a, "node-b", &vote("x", 2)).unwrap();
+        write_vote(&mut b, "node-b", &vote("y", 3)).unwrap();
+        a.commit();
+        b.commit();
+        a.merge(&mut b).unwrap();
+
+        let register = load_register(&a);
+        assert_eq!(register["node-a"], [vote("x", 1)]);
+        let mut twice = register["node-b"].clone();
+        twice.sort_by_key(|vote| vote.at);
+        assert_eq!(twice, [vote("x", 2), vote("y", 3)]);
+
+        // Seen, a vote is replaced.
+        write_vote(&mut a, "node-b", &vote("z", 4)).unwrap();
+        a.commit();
+        assert_eq!(load_register(&a)["node-b"], [vote("z", 4)]);
+    }
+
+    #[test]
     fn phases() {
         let mut record = Record {
             proposal: proposal(10_000),