From 6d128ad96ef165eb9781677b2659fcdfcfdee0b2 Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 24 Oct 2026 21:27:01 +0200 Subject: add votes register --- src/remake/records.rs | 131 ++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 110 insertions(+), 21 deletions(-) (limited to 'src/remake/records.rs') 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/`: each written by that member only (`vote`). +//! +//! Votes are kept apart, in the votes register `_continuum/votes/`: 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>; + #[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, #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) abort_requested: Option, - /// 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, } @@ -292,8 +301,37 @@ pub(crate) fn record_path(id: &str, part: &str) -> Vec { 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 { + 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, ®ister))) + }) + .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 { 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 { } } -/// 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 { + 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, ®ister_path(&voter)) { + Ok(values) => values, + Err(error) => { + warn!(voter = %voter, error = %error, "Failed to read vote"); + return None; + } + }; + let votes: Vec = 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, ®ister_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); @@ -456,6 +516,35 @@ mod tests { assert_eq!(record.phase(0, 0), Phase::Committed); } + #[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 { -- cgit v1.3.1