summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/document/store.rs23
-rw-r--r--src/remake/api.rs29
-rw-r--r--src/remake/coordinator.rs14
-rw-r--r--src/remake/delta.rs7
-rw-r--r--src/remake/local.rs17
-rw-r--r--src/remake/records.rs131
-rw-r--r--src/remake/snapshot.rs18
-rw-r--r--src/remake/vote.rs392
8 files changed, 507 insertions, 124 deletions
diff --git a/src/document/store.rs b/src/document/store.rs
index 9e18693..573b811 100644
--- a/src/document/store.rs
+++ b/src/document/store.rs
@@ -243,6 +243,29 @@ pub fn keys_at(doc: &AutoCommit, path: &[String]) -> Result<Vec<String>, StoreEr
Ok(merged_keys(doc, &maps).into_iter().collect())
}
+/// Every concurrent value at `path`, newest first; empty if there is none. Where
+/// `read_path` picks one (or merges concurrent maps into one), this keeps them apart.
+pub(crate) fn read_candidates(
+ doc: &AutoCommit,
+ path: &[String],
+) -> Result<Vec<JsonValue>, StoreError> {
+ let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?;
+ let mut maps = vec![automerge::ROOT];
+ for segment in parents {
+ maps = match resolve(doc, &maps, segment, None)? {
+ Some(Resolved::Maps(next)) => next,
+ _ => return Ok(Vec::new()),
+ };
+ }
+ candidates(doc, &maps, leaf)?
+ .into_iter()
+ .map(|(value, id)| match value {
+ Value::Scalar(scalar) => scalar_to_json(scalar.into_owned()),
+ Value::Object(_) => object_to_json(doc, &id, None),
+ })
+ .collect()
+}
+
/// Writes (replaces) the value at `path`, creating intermediate maps.
pub fn write_path(
doc: &mut AutoCommit,
diff --git a/src/remake/api.rs b/src/remake/api.rs
index 3345086..ffd1844 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -171,6 +171,9 @@ struct Report {
enum StartError {
#[error("remake {0} is already open for this epoch")]
Open(String),
+ /// The proposer votes for its own remake, so it must be free to.
+ #[error("this node can't vote for a new remake: {0}")]
+ VoteHeld(String),
#[error("base head {0} is not in this node's document")]
UnknownBase(String),
#[error(transparent)]
@@ -454,6 +457,9 @@ async fn start_kind<H: Host>(
Some(json!({ "remake": open })),
);
}
+ Err(error @ StartError::VoteHeld(_)) => {
+ return error_response(StatusCode::CONFLICT, error.to_string(), None);
+ }
Err(error @ StartError::UnknownBase(_)) => {
return error_response(StatusCode::BAD_REQUEST, error.to_string(), None);
}
@@ -502,13 +508,25 @@ async fn propose<H: Host>(
let record_id = id.clone();
let electorate_host = host.clone();
let me = host.node_id();
+ let started = host.remake().started_at() + grace;
let record = host
.commit_doc(move |doc| -> Result<Record, StartError> {
- if let Some((open, _)) = records::load_all(doc).into_iter().find(|(_, record)| {
+ let all = records::load_all(doc);
+ if let Some((open, _)) = all.iter().find(|(_, record)| {
record.proposal.from_epoch == proposal.from_epoch
&& record.phase(now, grace).is_open()
}) {
- return Err(StartError::Open(open));
+ return Err(StartError::Open(open.clone()));
+ }
+ let register = records::load_register(doc);
+ match vote::freedom(&me, &proposal.from_epoch, &all, &register, started) {
+ vote::Freedom::Any => {}
+ vote::Freedom::Only(id) => {
+ return Err(StartError::VoteHeld(format!(
+ "it may have voted for remake {id} before its start"
+ )));
+ }
+ vote::Freedom::Held(reason) => return Err(StartError::VoteHeld(reason)),
}
// The base is the requested heads, or everything before the proposal (the
@@ -538,8 +556,11 @@ async fn propose<H: Host>(
proposal.electorate = electorate;
records::write_proposal(doc, &record_id, &proposal)?;
- let own_vote = Vote { at: now };
- records::write_vote(doc, &record_id, &me, &own_vote)?;
+ let own_vote = Vote {
+ remake: record_id.clone(),
+ at: now,
+ };
+ records::write_vote(doc, &me, &own_vote)?;
Ok(Record {
proposal,
decision: None,
diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs
index 370d5fc..96edb91 100644
--- a/src/remake/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -13,7 +13,7 @@
//! every node can verify its own build against it.
//!
//! The task lives only in memory. If the coordinator dies, nobody decides and the
-//! proposal expires (its votes become void).
+//! proposal expires; its votes stay cast (`vote`).
use std::time::Duration;
@@ -47,8 +47,14 @@ pub(crate) async fn run<H: Host>(host: H, id: String) {
/// One evaluation. Returns true once the coordinator is done.
async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
let wanted = id.to_owned();
- let (record, all) = host
- .read_doc(move |doc| (records::load(doc, &wanted), records::load_all(doc)))
+ let (record, all, register) = host
+ .read_doc(move |doc| {
+ (
+ records::load(doc, &wanted),
+ records::load_all(doc),
+ records::load_register(doc),
+ )
+ })
.await;
let Some(record) = record else {
@@ -67,7 +73,7 @@ async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
return Ok(true);
}
- let tally = vote::tally(id, &record, &all, now, grace);
+ let tally = vote::tally(id, &record, &all, &register);
if tally.reached() {
info!(remake = %id, votes = %tally, "Majority reached");
commit(host, id, &record).await?;
diff --git a/src/remake/delta.rs b/src/remake/delta.rs
index d318c00..c8f6b7c 100644
--- a/src/remake/delta.rs
+++ b/src/remake/delta.rs
@@ -27,7 +27,7 @@ use automerge::{ActorId, AutoCommit, ChangeHash, PatchAction, Prop};
use tracing::warn;
use super::kind::{Translation, Translator};
-use super::snapshot::{EPOCHS, META_ROOT, REMAKES};
+use super::snapshot::{EPOCHS, META_ROOT, REMAKES, VOTES};
use crate::document::store::{self, StoreError};
#[derive(Debug, Default, Clone)]
@@ -135,7 +135,10 @@ fn log_failure(path: &[String], error: &StoreError) {
}
fn is_protocol_path(path: &[String]) -> bool {
- matches!(path, [root, part, ..] if root == META_ROOT && (part == REMAKES || part == EPOCHS))
+ let [root, part, ..] = path else {
+ return false;
+ };
+ root == META_ROOT && [REMAKES, VOTES, EPOCHS].contains(&part.as_str())
}
/// True for the root and for `_continuum` itself.
diff --git a/src/remake/local.rs b/src/remake/local.rs
index ec2f70a..0d6c6c7 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -17,12 +17,23 @@ use super::records::{Retention, unix_millis};
use super::snapshot::Snapshot;
use crate::document::replica::Epoch;
-#[derive(Default)]
pub(crate) struct RemakeState {
+ /// When this node started (Unix milliseconds): votes it cast before may be lost
+ /// (`vote`).
+ started_at: u64,
// A std mutex: every access is short and never spans an await.
inner: Mutex<Inner>,
}
+impl Default for RemakeState {
+ fn default() -> Self {
+ Self {
+ started_at: unix_millis(),
+ inner: Mutex::default(),
+ }
+ }
+}
+
#[derive(Default)]
struct Inner {
/// Built by the coordinator at commit time; consumed by the local switch.
@@ -130,6 +141,10 @@ pub(crate) struct ArchivedEpoch {
}
impl RemakeState {
+ pub(crate) fn started_at(&self) -> u64 {
+ self.started_at
+ }
+
fn lock(&self) -> MutexGuard<'_, Inner> {
self.inner
.lock()
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),
diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs
index d308dcd..e50a2a2 100644
--- a/src/remake/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -35,6 +35,8 @@ use crate::document::store::{self, StoreError};
pub(crate) const META_ROOT: &str = "_continuum";
pub(crate) const REMAKES: &str = "remakes";
+/// The votes register (`records`).
+pub(crate) const VOTES: &str = "votes";
pub(crate) const EPOCHS: &str = "epochs";
/// Everything that determines a snapshot besides the base content.
@@ -265,15 +267,18 @@ pub(crate) fn build(
kind.build(&base, params.content, &mut snapshot)?;
// Protocol data carries forward, whatever the kind: forget records, earlier epochs'
- // metadata (rejoining nodes need it). Remake records describe the old epoch's
- // protocol and stay behind.
+ // metadata (rejoining nodes need it). Remake records and the votes register describe
+ // the old epoch's protocol and stay behind.
match store::delete_path(&mut snapshot, &[META_ROOT.to_owned()], false) {
Ok(()) | Err(StoreError::NotFound) => {}
Err(error) => return Err(error.into()),
}
let mut protocol_keys = store::keys_at(&base, &[META_ROOT.to_owned()])?;
protocol_keys.sort();
- for key in protocol_keys.into_iter().filter(|key| key != REMAKES) {
+ for key in protocol_keys
+ .into_iter()
+ .filter(|key| key != REMAKES && key != VOTES)
+ {
store::copy_path(&base, &mut snapshot, &[META_ROOT.to_owned(), key])?;
}
@@ -526,6 +531,12 @@ mod tests {
.unwrap();
store::write_path(
&mut a,
+ &[META_ROOT.into(), VOTES.into(), "pk".into()],
+ &json!({ "remake": "c-1", "at": 1 }),
+ )
+ .unwrap();
+ store::write_path(
+ &mut a,
&[META_ROOT.into(), "forgotten".into(), "pk".into()],
&json!({ "at": 1 }),
)
@@ -540,6 +551,7 @@ mod tests {
.unwrap()
.unwrap();
assert!(meta.get(REMAKES).is_none());
+ assert!(meta.get(VOTES).is_none());
assert_eq!(meta["forgotten"]["pk"]["at"], json!(1));
let epoch = &meta[EPOCHS]["1"];
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);
}
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-25 08:08:42 +0000