diff options
Diffstat (limited to 'src/remake')
| -rw-r--r-- | src/remake/api.rs | 180 | ||||
| -rw-r--r-- | src/remake/catchup.rs | 57 | ||||
| -rw-r--r-- | src/remake/coordinator.rs | 84 | ||||
| -rw-r--r-- | src/remake/delta.rs | 2 | ||||
| -rw-r--r-- | src/remake/kind.rs | 239 | ||||
| -rw-r--r-- | src/remake/local.rs | 21 | ||||
| -rw-r--r-- | src/remake/mod.rs | 44 | ||||
| -rw-r--r-- | src/remake/quarantine.rs | 8 | ||||
| -rw-r--r-- | src/remake/records.rs | 40 | ||||
| -rw-r--r-- | src/remake/snapshot.rs | 8 | ||||
| -rw-r--r-- | src/remake/switch.rs | 9 | ||||
| -rw-r--r-- | src/remake/vote.rs | 291 |
12 files changed, 793 insertions, 190 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs index fbd5f54..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,25 +22,38 @@ //! "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) or `replace`, whose spec is +//! `kind` is `compact` (no spec), `replace`, whose spec is //! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }` -//! (see `kind::Replace`). Reports leave the replacement document out (`contentBytes`). +//! (see `kind::Replace`), or `import`, whose spec is +//! `{ "document": "<base64 of an Automerge document>", "rules": […] }` (see +//! `kind::Import`; `GET /admin/document` exports one). Reports leave the document out +//! (`contentBytes`). Requests may be up to 64 MiB. +//! +//! An import can also be uploaded as is, with `Content-Type: application/octet-stream`; +//! then `rules` (JSON), `baseHeads` (comma-separated), `settleSecs` and +//! `switchDelaySecs` go in the query. Restoring a backup, keeping writes made since at +//! their paths: +//! `curl --data-binary @backup.automerge -H 'Content-Type: application/octet-stream' --url-query 'rules=[{"from":"","to":""}]' …/admin/remakes` use std::time::Duration; use automerge::ChangeHash; use axum::{ + Json, body::Bytes, extract::{Path, Query, State}, - http::{header::LOCATION, StatusCode}, + http::{ + HeaderMap, StatusCode, + header::{CONTENT_TYPE, LOCATION}, + }, response::{IntoResponse, Response}, - Json, }; +use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; use serde::{Deserialize, Serialize}; -use serde_json::{json, Value as JsonValue}; +use serde_json::{Value as JsonValue, json}; use tokio::sync::broadcast; use tracing::{error, info, warn}; use uuid::Uuid; @@ -47,9 +61,9 @@ use uuid::Uuid; use super::coordinator; use super::local::LocalStatus; use super::records::{ - self, unix_millis, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, + self, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, Vote, unix_millis, }; -use super::{kind, Host}; +use super::{Host, kind, vote}; use crate::document::store::StoreError; #[derive(Debug, Default, Deserialize)] @@ -59,12 +73,64 @@ pub(crate) struct WaitQuery { } #[derive(Debug, Default, Deserialize)] +#[serde(rename_all = "camelCase")] 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 + // body to carry them: + /// The rules, as JSON (as in the spec). + rules: Option<String>, + /// Comma-separated hex. + base_heads: Option<String>, + settle_secs: Option<u64>, + switch_delay_secs: Option<u64>, +} + +impl StartQuery { + fn has_request_fields(&self) -> bool { + self.rules.is_some() + || self.base_heads.is_some() + || self.settle_secs.is_some() + || self.switch_delay_secs.is_some() + } + + /// An import request for a document uploaded as is. + fn import_request(&self, document: &[u8]) -> Result<StartRequest, String> { + let rules = match &self.rules { + None => JsonValue::Array(Vec::new()), + Some(rules) => serde_json::from_str(rules) + .map_err(|error| format!("rules are not valid JSON: {error}"))?, + }; + Ok(StartRequest { + kind: Some("import".to_owned()), + spec: Some(json!({ + "document": BASE64.encode(document), + "rules": rules, + })), + base_heads: self.base_heads.as_ref().map(|heads| { + heads + .split(',') + .filter(|head| !head.trim().is_empty()) + .map(str::to_owned) + .collect() + }), + settle_secs: self.settle_secs, + switch_delay_secs: self.switch_delay_secs, + ..StartRequest::default() + }) + } +} + +fn is_octet_stream(headers: &HeaderMap) -> bool { + headers + .get(CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.split(';').next()) + .is_some_and(|media| media.trim() == "application/octet-stream") } #[derive(Debug, Default, Deserialize)] @@ -80,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`. @@ -186,23 +253,26 @@ async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> { pub(crate) async fn start<H: Host>( State(host): State<H>, Query(query): Query<StartQuery>, + headers: HeaderMap, body: Bytes, ) -> Response { - start_kind(host, query, body, None).await + start_kind(host, query, &headers, body, None).await } /// `POST /admin/compaction`: the original endpoint, for compacting remakes only. pub(crate) async fn start_compaction<H: Host>( State(host): State<H>, Query(query): Query<StartQuery>, + headers: HeaderMap, body: Bytes, ) -> Response { - start_kind(host, query, body, Some(kind::DEFAULT_KIND)).await + start_kind(host, query, &headers, body, Some(kind::DEFAULT_KIND)).await } async fn start_kind<H: Host>( host: H, query: StartQuery, + headers: &HeaderMap, body: Bytes, only: Option<&'static str>, ) -> Response { @@ -210,7 +280,24 @@ async fn start_kind<H: Host>( return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None); } - let request: StartRequest = if body.is_empty() { + let request: StartRequest = if is_octet_stream(headers) { + match query.import_request(&body) { + Ok(request) => request, + Err(reason) => { + return error_response( + StatusCode::BAD_REQUEST, + format!("invalid request: {reason}"), + None, + ); + } + } + } else if query.has_request_fields() { + return error_response( + StatusCode::BAD_REQUEST, + "rules, baseHeads, settleSecs and switchDelaySecs go in the query only for an application/octet-stream import; put them in the JSON body", + None, + ); + } else if body.is_empty() { StartRequest::default() } else { match serde_json::from_slice(&body) { @@ -220,7 +307,7 @@ async fn start_kind<H: Host>( StatusCode::BAD_REQUEST, format!("invalid request: {error}"), None, - ) + ); } } }; @@ -247,7 +334,7 @@ async fn start_kind<H: Host>( StatusCode::BAD_REQUEST, format!("invalid {kind_name} spec: {reason}"), None, - ) + ); } }; @@ -269,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. @@ -316,30 +389,12 @@ 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(); let grace = grace(&host); - let mut proposal = Proposal { + let proposal = Proposal { coordinator: host.node_id(), from_epoch: host.current_epoch().into(), base_heads: Vec::new(), @@ -355,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, @@ -439,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)| { @@ -466,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?; @@ -559,14 +629,14 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { StatusCode::CONFLICT, "already committed; the switch cannot be undone", Some(json!({ "remake": current })), - ) + ); } Phase::Aborted | Phase::Expired => { return error_response( StatusCode::CONFLICT, "already over", Some(json!({ "remake": current })), - ) + ); } } diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index 4aa66e7..42b6526 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -7,8 +7,10 @@ //! 1. stops being ready and keeps its document aside; reads are answered from a copy //! of it, marked stale (`X-Continuum-Stale`), writes get `503`; //! 2. switches to an empty document labelled with the newer epoch: an empty history -//! conflicts with nothing, so ordinary sync fills it from peers on that epoch, -//! exactly like a new node joining; +//! conflicts with nothing. It downloads the epoch's snapshot change in chunks from +//! several peers on that epoch at once (`swarm::blob`; sync waits meanwhile), then +//! ordinary sync adds the changes made since, exactly like a new node joining. If +//! the download fails, sync transfers the snapshot too, in one message; //! 3. once caught up with a peer, checks that the history starts with the epoch's //! snapshot change; //! 4. decides about its own changes the swarm never saw (none, for a new node): @@ -31,13 +33,13 @@ use automerge::{AutoCommit, ChangeHash}; use tokio::time::Instant; use tracing::{error, info, warn}; +use super::Host; use super::delta::{self, ReplayStats}; use super::kind::Translator; use super::local::{CatchUp, LocalStatus, Rejoin}; use super::quarantine; use super::records::unix_millis; use super::snapshot::{self, EpochMetadata}; -use super::Host; use crate::document::replica::Epoch; /// Pause before trying the same target again after a failed rejoin. @@ -50,6 +52,10 @@ const MIN_REJOIN_TIMEOUT: Duration = Duration::from_secs(60); pub(super) async fn check<H: Host>(host: &H) { if let Some((target, started)) = host.remake().rejoin_progress() { let timeout = host.dormant_after().max(MIN_REJOIN_TIMEOUT); + if host.remake().fetching_snapshot() { + // The download has its own stall timeout; the clock restarts after it. + return; + } if host.initial_sync_done() { finish(host).await; } else if started.elapsed() > timeout { @@ -147,6 +153,45 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) { started: Instant::now(), was_ready, }); + + // Sync waits until the snapshot is here (or its download failed). + host.remake().set_fetching_snapshot(Some(target)); + tokio::spawn(fetch_snapshot(host.clone(), target)); +} + +/// Downloads the snapshot change of `target` and applies it, then lets sync go on. +async fn fetch_snapshot<H: Host>(host: H, target: Epoch) { + let outcome = host.fetch_change(ChangeHash(target.id)).await; + let still_rejoining = host + .remake() + .rejoin_progress() + .is_some_and(|(rejoining, _)| rejoining == target) + && host.current_epoch() == target; + if still_rejoining { + match outcome { + Ok(change) => { + let applied = host + .commit_doc(move |doc| doc.apply_changes([change])) + .await; + match applied { + Ok(()) => { + info!(epoch = %target, "Applied the epoch's snapshot; syncing the changes made since") + } + Err(error) => { + warn!(epoch = %target, error = %error, "Couldn't apply the downloaded snapshot; syncing it instead") + } + } + } + Err(error) => warn!( + epoch = %target, + error = %format!("{error:#}"), + "Couldn't download the epoch's snapshot; syncing it instead" + ), + } + host.remake().restart_rejoin_clock(target); + } + host.remake().set_fetching_snapshot(None); + host.request_sync(); } /// What to do with this node's own changes the new epoch lacks. @@ -444,9 +489,9 @@ async fn abandon<H: Host>(host: &H, rejoin: Rejoin, reason: String) { #[cfg(test)] mod tests { use super::*; - use crate::remake::snapshot::{build, SnapshotParams}; - use automerge::{transaction::Transactable, ActorId, ROOT}; - use serde_json::{json, Value as JsonValue}; + use crate::remake::snapshot::{SnapshotParams, build}; + use automerge::{ActorId, ROOT, transaction::Transactable}; + use serde_json::{Value as JsonValue, json}; /// An old-epoch document with one own change after the base, and the swarm's /// document of the next epoch (snapshot plus one foreign change). diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs index d1b3397..5885884 100644 --- a/src/remake/coordinator.rs +++ b/src/remake/coordinator.rs @@ -1,28 +1,28 @@ -//! 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}; +use super::Host; use super::records::{self, Decision, Record}; use super::snapshot; -use super::Host; +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/delta.rs b/src/remake/delta.rs index c161e2c..d318c00 100644 --- a/src/remake/delta.rs +++ b/src/remake/delta.rs @@ -268,7 +268,7 @@ fn patch_path(patch: &automerge::Patch) -> Vec<String> { #[cfg(test)] mod tests { use super::*; - use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, ROOT}; + use automerge::{ObjType, ROOT, ReadDoc, ScalarValue, transaction::Transactable}; use serde_json::json; fn view(doc: &AutoCommit) -> serde_json::Value { diff --git a/src/remake/kind.rs b/src/remake/kind.rs index b709a05..685fa28 100644 --- a/src/remake/kind.rs +++ b/src/remake/kind.rs @@ -10,17 +10,19 @@ //! * where a write made in the old epoch after the base goes in the new one //! ([`RemakeKind::translate`]). //! -//! Kinds: [`Compact`] (same content, no history) and [`Replace`] (an operator's -//! document, with rules for where later writes go). +//! Kinds: [`Compact`] (same content, no history), [`Replace`] (an operator's document +//! in store form, with rules for where later writes go) and [`Import`] (the same with +//! an Automerge document, e.g. one exported with `GET /admin/document`). //! //! Protocol data under `_continuum` is not a kind's business: the generic part carries //! it forward and replays it unchanged. use automerge::AutoCommit; +use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; use serde::{Deserialize, Serialize}; use serde_json::Value as JsonValue; -use super::snapshot::{canonical, META_ROOT}; +use super::snapshot::{META_ROOT, canonical}; use crate::document::store::{self, StoreError}; /// Name of the kind used when a request doesn't name one. @@ -75,6 +77,7 @@ pub(crate) fn by_name(name: &str) -> Option<&'static dyn RemakeKind> { match name { "compact" => Some(&Compact), "replace" => Some(&Replace), + "import" => Some(&Import), _ => None, } } @@ -196,33 +199,9 @@ impl RemakeKind for Replace { )); } - let mut rules = Vec::with_capacity(spec.rules.len()); - for rule in spec.rules { - let from = rule.from.segments()?; - let to = match (rule.to, rule.drop) { - (Some(to), false) => Some(to.segments()?), - (None, true) => None, - _ => { - return Err(format!( - "rule for {from:?} needs exactly one of \"to\" and \"drop\"" - )) - } - }; - if is_reserved(&from) || to.as_deref().is_some_and(is_reserved) { - return Err(format!( - "rules must not touch the reserved key {META_ROOT:?}" - )); - } - if rules.iter().any(|existing: &Rule| existing.from == from) { - return Err(format!("two rules for {from:?}")); - } - rules.push(Rule { from, to }); - } - Ok(Prepared { content: canonical(spec.document), - translation: serde_json::to_value(Rules { rules }) - .map_err(|error| error.to_string())?, + translation: prepare_rules(spec.rules)?, }) } @@ -246,31 +225,132 @@ impl RemakeKind for Replace { } fn translate(&self, path: &[String], translation: &JsonValue) -> Translation { - let Ok(Rules { rules }) = serde_json::from_value::<Rules>(translation.clone()) else { - return Translation::Quarantine("unreadable replacement rules".to_owned()); - }; - let Some(rule) = rules - .iter() - .filter(|rule| path.starts_with(&rule.from)) - .max_by_key(|rule| rule.from.len()) - else { - return Translation::Quarantine("no replacement rule covers this path".to_owned()); + translate_by_rules(path, translation) + } +} + +/// Validates rules from a request; the translation to store. +fn prepare_rules(specs: Vec<RuleSpec>) -> Result<JsonValue, String> { + let mut rules = Vec::with_capacity(specs.len()); + for rule in specs { + let from = rule.from.segments()?; + let to = match (rule.to, rule.drop) { + (Some(to), false) => Some(to.segments()?), + (None, true) => None, + _ => { + return Err(format!( + "rule for {from:?} needs exactly one of \"to\" and \"drop\"" + )); + } }; - match &rule.to { - None => Translation::Drop, - Some(to) => { - let mut target = to.clone(); - target.extend_from_slice(&path[rule.from.len()..]); - if target.as_slice() == path { - Translation::Same - } else { - Translation::To(target) - } + if is_reserved(&from) || to.as_deref().is_some_and(is_reserved) { + return Err(format!( + "rules must not touch the reserved key {META_ROOT:?}" + )); + } + if rules.iter().any(|existing: &Rule| existing.from == from) { + return Err(format!("two rules for {from:?}")); + } + rules.push(Rule { from, to }); + } + serde_json::to_value(Rules { rules }).map_err(|error| error.to_string()) +} + +/// Where `path` goes under the stored rules: the longest matching `from` wins. +fn translate_by_rules(path: &[String], translation: &JsonValue) -> Translation { + let Ok(Rules { rules }) = serde_json::from_value::<Rules>(translation.clone()) else { + return Translation::Quarantine("unreadable replacement rules".to_owned()); + }; + let Some(rule) = rules + .iter() + .filter(|rule| path.starts_with(&rule.from)) + .max_by_key(|rule| rule.from.len()) + else { + return Translation::Quarantine("no replacement rule covers this path".to_owned()); + }; + match &rule.to { + None => Translation::Drop, + Some(to) => { + let mut target = to.clone(); + target.extend_from_slice(&path[rule.from.len()..]); + if target.as_slice() == path { + Translation::Same + } else { + Translation::To(target) } } } } +/// Import: like [`Replace`], but the new document is an Automerge document (bytes, as +/// saved), so Automerge types the store's JSON can't express survive: text, counters, +/// bytes, timestamps, floats. Its `_continuum` (protocol data of whatever swarm it came +/// from) is left out, as is its history: the content is the merged view. +/// +/// Spec: `{ "document": "<base64 of the saved document>", "rules": […] }`, rules as for +/// [`Replace`]. To restore a backup of this swarm with every later write kept where it +/// was: `"rules": [{"from": "", "to": ""}]`. +pub(crate) struct Import; + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct ImportSpec { + document: String, + #[serde(default)] + rules: Vec<RuleSpec>, +} + +/// The document in `content` (base64 of a saved Automerge document). +fn load_content(content: &JsonValue) -> Result<AutoCommit, String> { + let JsonValue::String(encoded) = content else { + return Err("the document must be a base64 string".to_owned()); + }; + let bytes = BASE64 + .decode(encoded.trim()) + .map_err(|error| format!("the document is not valid base64: {error}"))?; + AutoCommit::load(&bytes) + .map_err(|error| format!("the document is not an Automerge document: {error}")) +} + +impl RemakeKind for Import { + fn name(&self) -> &'static str { + "import" + } + + fn propose(&self, spec: &JsonValue) -> Result<Prepared, String> { + let spec: ImportSpec = + serde_json::from_value(spec.clone()).map_err(|error| error.to_string())?; + let imported = load_content(&JsonValue::String(spec.document))?; + + // Keep only what goes into the new epoch: the merged view without protocol data + // and history. Usually much smaller than what was uploaded, and it is what gets + // replicated with the proposal. + let mut content = AutoCommit::new(); + store::materialize_into(&imported, &mut content, &is_reserved) + .map_err(|error| format!("the document can't be read: {error}"))?; + + Ok(Prepared { + content: JsonValue::String(BASE64.encode(content.save())), + translation: prepare_rules(spec.rules)?, + }) + } + + fn build( + &self, + _old_at_base: &AutoCommit, + content: &JsonValue, + target: &mut AutoCommit, + ) -> Result<(), StoreError> { + let imported = load_content(content).map_err(|reason| anyhow::anyhow!(reason))?; + // Deterministic: the same bytes give the same view, copied in a fixed order. + store::materialize_into(&imported, target, &is_reserved) + } + + fn translate(&self, path: &[String], translation: &JsonValue) -> Translation { + translate_by_rules(path, translation) + } +} + /// The translations of several consecutive remakes, applied in order: a node that /// rejoins several epochs later carries its writes through every remake in between. #[derive(Default, Clone)] @@ -399,10 +479,10 @@ mod tests { assert!(replace(json!({ "document": [] })).is_err()); assert!(replace(json!({ "document": { META_ROOT: {} } })).is_err()); assert!(replace(json!({ "document": {}, "rules": [{ "from": "a" }] })).is_err()); - assert!(replace( - json!({ "document": {}, "rules": [{ "from": "a", "to": "b", "drop": true }] }) - ) - .is_err()); + assert!( + replace(json!({ "document": {}, "rules": [{ "from": "a", "to": "b", "drop": true }] })) + .is_err() + ); assert!( replace(json!({ "document": {}, "rules": [{ "from": "a//b", "to": "c" }] })).is_err() ); @@ -448,6 +528,63 @@ mod tests { } #[test] + fn import_keeps_types_and_drops_protocol_data() { + use automerge::{ObjType, ROOT, ReadDoc, transaction::Transactable}; + use serde_json::json; + + let mut source = AutoCommit::new(); + let text = source.put_object(ROOT, "notes", ObjType::Text).unwrap(); + source.splice_text(&text, 0, 0, "hello").unwrap(); + source + .put(ROOT, "hits", automerge::ScalarValue::counter(3)) + .unwrap(); + let meta = source.put_object(ROOT, META_ROOT, ObjType::Map).unwrap(); + source.put(&meta, "members", "elsewhere").unwrap(); + let encoded = BASE64.encode(source.save()); + + assert!( + Import + .propose(&json!({ "document": "not base64!" })) + .is_err() + ); + assert!( + Import + .propose(&json!({ "document": BASE64.encode(b"junk") })) + .is_err() + ); + + let prepared = Import + .propose(&json!({ "document": encoded, "rules": [{ "from": "", "to": "" }] })) + .unwrap(); + let build = || { + let mut target = AutoCommit::new().with_actor(automerge::ActorId::from(&b"fixed"[..])); + Import + .build(&AutoCommit::new(), &prepared.content, &mut target) + .unwrap(); + target.commit_with(automerge::transaction::CommitOptions::default().with_time(0)); + target + }; + let mut a = build(); + let mut b = build(); + assert_eq!(a.get_heads(), b.get_heads()); + + assert!(a.get(ROOT, META_ROOT).unwrap().is_none()); + let (notes, notes_id) = a.get(ROOT, "notes").unwrap().unwrap(); + assert!(matches!(notes, automerge::Value::Object(ObjType::Text))); + assert_eq!(a.text(¬es_id).unwrap(), "hello"); + assert!(matches!( + a.get(ROOT, "hits").unwrap().unwrap().0, + automerge::Value::Scalar(ref s) if matches!(s.as_ref(), automerge::ScalarValue::Counter(_)) + )); + + // Writes after the base keep their paths. + assert_eq!( + Translator::single(&Import, prepared.translation).translate(&path(&["x", "y"])), + Translation::Same + ); + } + + #[test] fn replace_builds_the_document_deterministically() { use serde_json::json; let build = |document: serde_json::Value| { diff --git a/src/remake/local.rs b/src/remake/local.rs index 35a4ff1..787bd5e 100644 --- a/src/remake/local.rs +++ b/src/remake/local.rs @@ -12,7 +12,7 @@ use automerge::AutoCommit; use serde::Serialize; use tokio::time::Instant; -use super::records::{unix_millis, Retention}; +use super::records::{Retention, unix_millis}; use super::snapshot::Snapshot; use crate::document::replica::Epoch; @@ -34,6 +34,8 @@ struct Inner { catch_up: Option<CatchUp>, /// A rejoin in progress: the node syncs an empty document of a newer epoch. rejoin: Option<Rejoin>, + /// While the rejoin downloads the target epoch's snapshot (`catchup`), sync waits. + fetching_snapshot: Option<Epoch>, quarantine: Vec<Quarantined>, } @@ -211,6 +213,23 @@ impl RemakeState { self.lock().rejoin.take() } + /// Whether sync should wait: a rejoin is downloading its epoch's snapshot. + pub(crate) fn fetching_snapshot(&self) -> bool { + self.lock().fetching_snapshot.is_some() + } + + pub(crate) fn set_fetching_snapshot(&self, target: Option<Epoch>) { + self.lock().fetching_snapshot = target; + } + + /// The rejoin of `target`, if still in progress, gets its full time for the sync + /// that follows the download. + pub(crate) fn restart_rejoin_clock(&self, target: Epoch) { + if let Some(rejoin) = self.lock().rejoin.as_mut().filter(|r| r.target == target) { + rejoin.started = Instant::now(); + } + } + pub(crate) fn add_quarantine(&self, quarantined: Quarantined) { self.lock().quarantine.push(quarantined); } diff --git a/src/remake/mod.rs b/src/remake/mod.rs index ff3c9c3..1c2be36 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; @@ -37,8 +39,9 @@ use std::time::Duration; use automerge::AutoCommit; use axum::{ - routing::{get, post}, Router, + extract::DefaultBodyLimit, + routing::{get, post}, }; use tokio::sync::broadcast; @@ -52,15 +55,29 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { /// This node's in-memory remake state. fn remake(&self) -> &RemakeState; - /// This node's peer ID. + /// This node's ID (its peers know it by this). fn node_id(&self) -> String; /// 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>; + + /// Downloads a change from peers on the local epoch, in chunks from several at once, + /// checked against its hash. + fn fetch_change( + &self, + hash: automerge::ChangeHash, + ) -> impl Future<Output = anyhow::Result<automerge::Change>> + Send; + + /// Starts sync with every peer (e.g. after holding it back). + fn request_sync(&self); + fn is_node_ready(&self) -> impl Future<Output = bool> + Send; /// Known (not forgotten) peers. @@ -92,12 +109,6 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { F: FnOnce(&AutoCommit) -> T + Send + 'static, T: Send + 'static; - /// Mutable access to the document without publishing a change (e.g. forks). - fn update_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send - where - F: FnOnce(&mut AutoCommit) -> T + Send + 'static, - T: Send + 'static; - /// One committed change (rolled back on error), published and replicated. fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send where @@ -123,20 +134,25 @@ pub(crate) struct PeerSummary { pub(crate) id: String, /// Not dormant. pub(crate) active: bool, - /// Handshake completed. - pub(crate) established: bool, /// The epoch the peer last advertised. pub(crate) epoch: Option<Epoch>, } /// Admin routes. Merged into the authenticated part of the node's router. +/// Largest remake request (bytes). The proposal carries the new document to every +/// node in one sync message, which Mycelium must deliver within its deadline, so +/// documents near this size are already a stretch. +const REMAKE_REQUEST_LIMIT: usize = 64 * 1024 * 1024; + pub(crate) fn router<H: Host>() -> Router<H> { Router::new() .route( "/admin/remakes", get(api::list::<H>) .post(api::start::<H>) - .delete(api::abort_current::<H>), + .delete(api::abort_current::<H>) + // A replace or import request carries a whole document. + .layer(DefaultBodyLimit::max(REMAKE_REQUEST_LIMIT)), ) // The original endpoint, kept as a shortcut for a compacting remake. .route("/admin/compaction", post(api::start_compaction::<H>)) diff --git a/src/remake/quarantine.rs b/src/remake/quarantine.rs index deed8cd..1090e2c 100644 --- a/src/remake/quarantine.rs +++ b/src/remake/quarantine.rs @@ -16,22 +16,22 @@ use std::convert::Infallible; use automerge::AutoCommit; use axum::{ + Json, extract::{Path, State}, - http::{header, StatusCode}, + http::{StatusCode, header}, response::{IntoResponse, Response}, - Json, }; use serde::Serialize; -use serde_json::{json, Value as JsonValue}; +use serde_json::{Value as JsonValue, json}; use tracing::{info, warn}; use uuid::Uuid; +use super::Host; use super::api::error_response; use super::delta; use super::kind::Translator; use super::local::Quarantined; use super::records::unix_millis; -use super::Host; use crate::document::replica::Epoch; use crate::document::store; diff --git a/src/remake/records.rs b/src/remake/records.rs index 600d4dc..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,13 +16,15 @@ //! 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; use tracing::warn; use super::kind::DEFAULT_KIND; -use super::snapshot::{SnapshotRequest, META_ROOT, REMAKES}; +use super::snapshot::{META_ROOT, REMAKES, SnapshotRequest}; use crate::document::replica::Epoch; use crate::document::store::{self, StoreError}; @@ -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/snapshot.rs b/src/remake/snapshot.rs index bd99854..efe7e37 100644 --- a/src/remake/snapshot.rs +++ b/src/remake/snapshot.rs @@ -24,11 +24,11 @@ use std::collections::BTreeMap; -use automerge::{transaction::CommitOptions, ActorId, AutoCommit, ChangeHash}; -use serde_json::{json, Map as JsonMap, Value as JsonValue}; +use automerge::{ActorId, AutoCommit, ChangeHash, transaction::CommitOptions}; +use serde_json::{Map as JsonMap, Value as JsonValue, json}; use sha2::{Digest, Sha256}; -use super::kind::{by_name, RemakeKind, Translator, DEFAULT_KIND}; +use super::kind::{DEFAULT_KIND, RemakeKind, Translator, by_name}; use super::records::{EpochRecord, Policy}; use crate::document::replica::Epoch; use crate::document::store::{self, StoreError}; @@ -353,7 +353,7 @@ pub(super) fn canonical(value: JsonValue) -> JsonValue { #[cfg(test)] mod tests { use super::*; - use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, Value, ROOT}; + use automerge::{ObjType, ROOT, ReadDoc, ScalarValue, Value, transaction::Transactable}; fn doc(actor: &[u8]) -> AutoCommit { AutoCommit::new().with_actor(ActorId::from(actor)) diff --git a/src/remake/switch.rs b/src/remake/switch.rs index 2670f14..e9f49ee 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -24,14 +24,15 @@ use automerge::{ActorId, AutoCommit, ChangeHash}; use tokio::sync::broadcast; use tracing::{error, info, warn}; +use super::Host; use super::catchup; use super::delta::{self, ReplayStats}; use super::kind::Translator; use super::local::LocalStatus; use super::quarantine; -use super::records::{self, unix_millis, EpochRecord, Record}; +use super::records::{self, EpochRecord, Record, unix_millis}; use super::snapshot::{self, META_ROOT, REMAKES}; -use super::Host; +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); + } +} |