diff options
| -rw-r--r-- | src/remake/api.rs | 69 | ||||
| -rw-r--r-- | src/remake/coordinator.rs | 82 | ||||
| -rw-r--r-- | src/remake/mod.rs | 12 | ||||
| -rw-r--r-- | src/remake/records.rs | 38 | ||||
| -rw-r--r-- | src/remake/switch.rs | 5 | ||||
| -rw-r--r-- | src/remake/vote.rs | 291 | ||||
| -rw-r--r-- | src/state.rs | 4 |
7 files changed, 406 insertions, 95 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs index 8467429..de8166c 100644 --- a/src/remake/api.rs +++ b/src/remake/api.rs @@ -3,8 +3,9 @@ //! * `POST /admin/remakes` proposes a remake of the current epoch (`POST //! /admin/compaction` is a shortcut, kept from before). The receiving //! node becomes its coordinator. `202` with the report, or with `?wait=true` the final -//! report once decided. A dropped request does not abort anything. `?force=true` -//! accepts a `settleSecs` below the dormancy threshold (tests, single-site swarms). +//! report once decided. A dropped request does not abort anything. The remake +//! commits as soon as a majority of its electorate voted for it (`vote`), and aborts +//! once that is out of reach or `settleSecs` passed without it. //! * `GET /admin/remakes` lists all remakes. //! * `GET /admin/remakes/{id}` reports one; `?wait=true` long-polls until decided //! (bounded by the proposal's own expiry). @@ -21,8 +22,8 @@ //! "archiveRetention": {"epochs": 2, "seconds": 2592000}, //! "settleSecs": 480, "switchDelaySecs": 60 }` //! -//! `?force=true` accepts `settleSecs` below the dormancy threshold and `switchDelaySecs` -//! below one keepalive interval: for tests and debugging, not for real swarms. +//! `?force=true` accepts a `switchDelaySecs` below one keepalive interval: for tests +//! and debugging, not for real swarms. //! //! `kind` is `compact` (no spec), `replace`, whose spec is //! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }` @@ -60,9 +61,9 @@ use uuid::Uuid; use super::coordinator; use super::local::LocalStatus; use super::records::{ - self, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, unix_millis, + self, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, Vote, unix_millis, }; -use super::{Host, kind}; +use super::{Host, kind, vote}; use crate::document::store::StoreError; #[derive(Debug, Default, Deserialize)] @@ -76,7 +77,7 @@ pub(crate) struct WaitQuery { pub(crate) struct StartQuery { #[serde(default)] wait: bool, - /// Allows a settle period shorter than the dormancy threshold. + /// Allows a switch delay shorter than one keepalive interval. #[serde(default)] force: bool, // Only for an import uploaded as `application/octet-stream`, which has no JSON @@ -145,6 +146,7 @@ struct StartRequest { base_heads: Option<Vec<String>>, replay_window: Option<Retention>, archive_retention: Option<Retention>, + /// How long votes are collected; twice the dormancy threshold if absent. settle_secs: Option<u64>, /// How long nodes wait after seeing the commit before switching; one keepalive /// interval if absent. Shorter only with `?force=true`. @@ -354,27 +356,13 @@ async fn start_kind<H: Host>( } }; + // The voting period. No lower bound: the majority protects against competitors, + // a short period only risks an abort for lack of votes. let dormant_after = host.dormant_after(); let settle = request .settle_secs .map(Duration::from_secs) .unwrap_or(dormant_after * 2); - if settle < dormant_after && query.force { - warn!( - settle_secs = settle.as_secs_f64(), - dormant_after_secs = dormant_after.as_secs(), - "Forced settle period shorter than the dormancy threshold; a partition may go unnoticed" - ); - } else if settle < dormant_after { - return error_response( - StatusCode::BAD_REQUEST, - format!( - "settleSecs must be at least {} (the dormancy threshold), so a partition is detected before deciding; add ?force=true to override", - dormant_after.as_secs() - ), - None, - ); - } // The delay lets the commit reach every node through ordinary sync before any of // them stops syncing the old epoch. @@ -401,24 +389,6 @@ async fn start_kind<H: Host>( _ => {} } - // Fail fast instead of proposing something that will abort anyway. - let peers = host.peer_summaries().await; - let dormant: Vec<String> = peers - .iter() - .filter(|peer| !peer.active) - .map(|peer| peer.id.clone()) - .collect(); - if !dormant.is_empty() { - return error_response( - StatusCode::CONFLICT, - "dormant peers block remake; forget them first (DELETE /peers/{pk})", - Some(json!({ "dormantPeers": dormant })), - ); - } - - let mut peers_considered: Vec<String> = peers.into_iter().map(|peer| peer.id).collect(); - peers_considered.sort(); - let now = unix_millis(); let settle_ms = settle.as_millis() as u64; let id = Uuid::new_v4().to_string(); @@ -440,7 +410,9 @@ async fn start_kind<H: Host>( proposed_at: now, settle_until: now + settle_ms, switch_delay_ms: switch_delay.map(|delay| delay.as_millis() as u64), - peers_considered, + // Filled in when the proposal is written, from the document at that moment. + electorate: Vec::new(), + quorum: 0, kind: kind.name().to_owned(), content: prepared.content, translation: prepared.translation, @@ -524,6 +496,8 @@ async fn propose<H: Host>( grace: u64, ) -> Result<Record, StartError> { let record_id = id.clone(); + let electorate_host = host.clone(); + let me = host.node_id(); let record = host .commit_doc(move |doc| -> Result<Record, StartError> { if let Some((open, _)) = records::load_all(doc).into_iter().find(|(_, record)| { @@ -551,11 +525,22 @@ async fn propose<H: Host>( heads.sort(); proposal.base_heads = heads; + // We vote for our own remake, so we are always in the electorate. + let mut electorate = electorate_host.electorate(doc, now); + if let Err(position) = electorate.binary_search(&me) { + electorate.insert(position, me.clone()); + } + proposal.quorum = vote::quorum(electorate.len()); + 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)?; Ok(Record { proposal, decision: None, abort_requested: None, + votes: [(me, own_vote)].into(), }) }) .await?; diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs index 28140cb..5885884 100644 --- a/src/remake/coordinator.rs +++ b/src/remake/coordinator.rs @@ -1,21 +1,20 @@ -//! The coordinator: the node that received the proposal watches the swarm during the -//! settle period and then writes the decision. +//! The coordinator: the node that received the proposal counts the votes (`vote`) and +//! writes the decision. //! -//! It aborts if, at any point before deciding: +//! It commits as soon as a majority of the electorate voted for the remake. It aborts +//! if, before that: //! * an operator requested an abort, -//! * a competing proposal for the same epoch appeared (both coordinators abort), -//! * a peer it considered went dormant (a partition may hide a competitor), -//! * the local epoch changed, or -//! * the decision would come later than the settle period plus the grace margin -//! (other nodes report the proposal as expired by then). +//! * the local epoch changed, +//! * a competing remake of the same epoch committed, +//! * a majority is out of reach (too many members voted for a competing remake), or +//! * the voting period (`settleSecs`) passed without a majority. //! -//! Otherwise it builds the snapshot at the base heads and commits, recording the new -//! epoch so every node can verify its own build against it. +//! Committing builds the snapshot at the base heads and records the new epoch, so +//! 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. +//! proposal expires (its votes become void). -use std::collections::HashMap; use std::time::Duration; use tracing::{info, warn}; @@ -23,6 +22,7 @@ use tracing::{info, warn}; use super::Host; use super::records::{self, Decision, Record}; use super::snapshot; +use super::vote; use crate::document::store::StoreError; const TICK: Duration = Duration::from_secs(1); @@ -61,20 +61,34 @@ async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> { let now = records::unix_millis(); let grace = host.dormant_after().as_millis() as u64; - if let Some(reason) = abort_reason(host, id, &record, &all, now, grace).await { + if let Some(reason) = abort_reason(host, id, &record, &all, now, grace) { decide_abort(host, id, reason).await?; return Ok(true); } - if now < record.proposal.settle_until { - return Ok(false); + let tally = vote::tally(id, &record, &all, now, grace); + if tally.reached() { + info!(remake = %id, votes = %tally, "Majority reached"); + commit(host, id, &record).await?; + return Ok(true); } - - commit(host, id, &record).await?; - Ok(true) + if tally.out_of_reach() { + decide_abort(host, id, format!("a majority is out of reach: {tally}")).await?; + return Ok(true); + } + if now >= record.proposal.settle_until { + decide_abort( + host, + id, + format!("no majority in the voting period: {tally}"), + ) + .await?; + return Ok(true); + } + Ok(false) } -async fn abort_reason<H: Host>( +fn abort_reason<H: Host>( host: &H, id: &str, record: &Record, @@ -89,39 +103,15 @@ async fn abort_reason<H: Host>( } if records::EpochRecord::from(host.current_epoch()) != proposal.from_epoch { - return Some("local epoch changed during settle".to_owned()); + return Some("local epoch changed during the vote".to_owned()); } if let Some((other, _)) = all.iter().find(|(other, candidate)| { other != id && candidate.proposal.from_epoch == proposal.from_epoch - && candidate.phase(now, grace).is_open() + && candidate.phase(now, grace) == records::Phase::Committed }) { - return Some(format!("competing remake {other} for the same epoch")); - } - - // Peers forgotten in the meantime are ignored: an operator decided they are gone. - let peers: HashMap<String, bool> = host - .peer_summaries() - .await - .into_iter() - .map(|peer| (peer.id, peer.active)) - .collect(); - let dormant: Vec<&str> = proposal - .peers_considered - .iter() - .filter(|peer| peers.get(peer.as_str()) == Some(&false)) - .map(String::as_str) - .collect(); - if !dormant.is_empty() { - return Some(format!( - "peers went dormant during settle: {}", - dormant.join(", ") - )); - } - - if now > proposal.settle_until.saturating_add(grace) { - return Some("decision deadline passed".to_owned()); + return Some(format!("competing remake {other} committed")); } None diff --git a/src/remake/mod.rs b/src/remake/mod.rs index 6a9c6f9..0895931 100644 --- a/src/remake/mod.rs +++ b/src/remake/mod.rs @@ -2,10 +2,11 @@ //! Compaction is one: the new document has the old one's content without its history. //! //! * `kind`: what differs between kinds of remake (building the new document, and -//! where writes made after the base go): `compact` and `replace`. +//! where writes made after the base go): `compact`, `replace` and `import`. //! * `snapshot`: deterministic build of the next epoch's initial document. //! * `records`: the replicated remake records under `_continuum/remakes/<id>`. -//! * `coordinator`: proposal, settle period and decision on the node that was asked. +//! * `coordinator`: proposal, vote count and decision on the node that was asked. +//! * `vote`: who votes, for which remake, and when a remake has its majority. //! * `switch`: switching this node to a committed epoch (worker). //! * `catchup`: rejoining a newer epoch from peers (late, new or failed nodes), then //! replaying or quarantining this node's own missing changes. @@ -27,6 +28,7 @@ mod quarantine; mod records; pub(crate) mod snapshot; mod switch; +mod vote; pub(crate) use local::RemakeState; pub(crate) use switch::run_worker; @@ -59,9 +61,13 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { /// The epoch of the local replica. fn current_epoch(&self) -> Epoch; - /// Lower bound and default basis for the settle period. + /// Default basis for the voting period, and the grace margin after it. fn dormant_after(&self) -> Duration; + /// The electorate of a remake proposed now (`vote`): the members that said they are + /// alive recently enough, read from `doc`. + fn electorate(&self, doc: &AutoCommit, now_ms: u64) -> Vec<String>; + fn is_node_ready(&self) -> impl Future<Output = bool> + Send; /// Known (not forgotten) peers. diff --git a/src/remake/records.rs b/src/remake/records.rs index 5c844f1..ed018ca 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -7,6 +7,7 @@ //! * `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`). //! //! 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, @@ -15,6 +16,8 @@ //! Records edited into an unreadable shape through the store API are skipped with a //! warning. +use std::collections::BTreeMap; + use automerge::{AutoCommit, ChangeHash}; use serde::{Deserialize, Serialize}; use serde_json::Value as JsonValue; @@ -127,8 +130,10 @@ pub(crate) struct Proposal { /// interval if absent. #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) switch_delay_ms: Option<u64>, - /// Peers that must stay active during the settle period. - pub(crate) peers_considered: Vec<String>, + /// Members whose votes count (sorted), fixed at proposal time (`vote`). + pub(crate) electorate: Vec<String>, + /// Votes needed to commit: a majority of the electorate. + pub(crate) quorum: usize, /// The remake kind (`compact` for records written before kinds existed). #[serde(default = "default_kind")] pub(crate) kind: String, @@ -173,6 +178,13 @@ pub(crate) struct AbortRequest { pub(crate) at: u64, } +/// A member's vote for a remake. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Vote { + pub(crate) at: u64, +} + #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct Record { @@ -181,6 +193,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)] + pub(crate) votes: BTreeMap<String, Vote>, } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] @@ -316,6 +331,18 @@ 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> { + 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) +} + pub(crate) fn decode_hex32(hex: &str) -> Option<[u8; 32]> { if hex.len() != 64 || !hex.is_ascii() { return None; @@ -360,7 +387,8 @@ mod tests { proposed_at: settle_until.saturating_sub(1_000), settle_until, switch_delay_ms: None, - peers_considered: vec!["node-b".into()], + electorate: vec!["node-a".into(), "node-b".into()], + quorum: 2, kind: DEFAULT_KIND.to_owned(), content: JsonValue::Null, translation: JsonValue::Null, @@ -390,6 +418,7 @@ 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(); doc.commit(); let all = load_all(&doc); @@ -399,6 +428,8 @@ mod tests { assert_eq!(record.proposal.policy, p.policy); assert_eq!(record.proposal.base_change_hashes().unwrap().len(), 1); assert!(record.abort_requested.is_some()); + assert_eq!(record.votes.keys().collect::<Vec<_>>(), ["node-b"]); + assert_eq!(record.proposal.quorum, 2); assert_eq!(record.phase(0, 0), Phase::Committed); } @@ -408,6 +439,7 @@ mod tests { proposal: proposal(10_000), decision: None, abort_requested: None, + votes: BTreeMap::new(), }; assert_eq!(record.phase(10_000, 500), Phase::Proposed); assert_eq!(record.phase(10_500, 500), Phase::Proposed); diff --git a/src/remake/switch.rs b/src/remake/switch.rs index f4a018a..e9f49ee 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -32,6 +32,7 @@ use super::local::LocalStatus; use super::quarantine; use super::records::{self, EpochRecord, Record, unix_millis}; use super::snapshot::{self, META_ROOT, REMAKES}; +use super::vote; use crate::document::replica::Epoch; use crate::document::store; use crate::swarm::sync::paths_related; @@ -78,6 +79,8 @@ async fn check<H: Host>(host: &H) { return catchup::check(host).await; } + vote::cast(host).await; + let all = host.read_doc(records::load_all).await; let local_record = EpochRecord::from(local); @@ -93,7 +96,7 @@ async fn check<H: Host>(host: &H) { [] => return catchup::check(host).await, [(id, record)] => (id.clone(), record.clone()), several => { - // The settle protocol prevents this; never guess which one is right. Peers + // Majority voting prevents this; never guess which one is right. Peers // that did switch can still be followed. let ids: Vec<&str> = several.iter().map(|(id, _)| id.as_str()).collect(); error!(remakes = ?ids, "Several remakes committed for this epoch; not switching on my own"); 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); + } +} diff --git a/src/state.rs b/src/state.rs index 25c734e..a47211f 100644 --- a/src/state.rs +++ b/src/state.rs @@ -1022,6 +1022,10 @@ impl Host for AppState { self.timing.dormant_after } + fn electorate(&self, doc: &AutoCommit, now_ms: u64) -> Vec<String> { + crate::swarm::members::electorate(doc, now_ms, self.electorate_window) + } + fn is_node_ready(&self) -> impl Future<Output = bool> + Send { let state = self.clone(); async move { state.is_ready().await } |