diff options
Diffstat (limited to 'src/remake')
| -rw-r--r-- | src/remake/api.rs | 422 | ||||
| -rw-r--r-- | src/remake/catchup.rs | 127 | ||||
| -rw-r--r-- | src/remake/content.rs | 95 | ||||
| -rw-r--r-- | src/remake/coordinator.rs | 251 | ||||
| -rw-r--r-- | src/remake/delta.rs | 9 | ||||
| -rw-r--r-- | src/remake/kind.rs | 267 | ||||
| -rw-r--r-- | src/remake/local.rs | 101 | ||||
| -rw-r--r-- | src/remake/mod.rs | 61 | ||||
| -rw-r--r-- | src/remake/quarantine.rs | 8 | ||||
| -rw-r--r-- | src/remake/records.rs | 265 | ||||
| -rw-r--r-- | src/remake/snapshot.rs | 36 | ||||
| -rw-r--r-- | src/remake/switch.rs | 32 | ||||
| -rw-r--r-- | src/remake/vote.rs | 515 |
13 files changed, 1879 insertions, 310 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs index 5267e48..826c81e 100644 --- a/src/remake/api.rs +++ b/src/remake/api.rs @@ -3,17 +3,19 @@ //! * `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). //! * `GET /admin/epochs` shows the local epoch, the archived previous ones and whether //! the node is catching up with a newer epoch from a peer. //! * `DELETE /admin/remakes/{id}` asks the coordinator to abort. Only possible while -//! proposed; `409` once committed (a switch can't be undone) or already over. -//! `DELETE /admin/remakes` does the same for the remake of the local epoch that -//! is in progress, without having to know its ID. +//! undecided; `409` once committed (a switch can't be undone) or aborted. An expired +//! remake is aborted right away by the node asked: its coordinator is gone, and +//! can't commit it any more anyway (`coordinator`). `DELETE /admin/remakes` does the +//! same for the undecided remake of the local epoch, without having to know its ID. //! //! Request body for POST (all optional): //! `{ "kind": "compact", "spec": null, "baseHeads": ["<hex>", …], @@ -21,25 +23,43 @@ //! "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=` overrides checks: `true` all of them, or a comma-separated list of +//! * `switchDelaySecs`: accept one below a keepalive interval (for tests and debugging, +//! not for real swarms); +//! * `supersede`: abort expired remakes of this epoch and propose in their place. +//! Without it, an expired remake still holding votes answers `409`. //! -//! `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). The document stays on the +//! coordinator; the proposal refers to it by hash (`proposal.content`), and every voter +//! downloads it (`content`). 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::collections::HashSet; 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 +67,10 @@ 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, ContentRef, Decision, 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 +80,63 @@ 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. - #[serde(default)] - force: bool, + /// The checks to override (`Force`). + force: Option<String>, + // 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,12 +152,60 @@ 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`. + /// interval if absent. Shorter only with `?force=switchDelaySecs`. switch_delay_secs: Option<u64>, } +/// The checks `?force=` overrides: `true` all of them, or a comma-separated list. +#[derive(Debug, Default, Clone, Copy)] +struct Force { + switch_delay: bool, + supersede: bool, +} + +impl Force { + fn parse(value: Option<&str>) -> Result<Self, String> { + let mut force = Force::default(); + match value.map(str::trim) { + None | Some("" | "false") => {} + Some("true") => { + force = Force { + switch_delay: true, + supersede: true, + }; + } + Some(list) => { + for name in list.split(',').map(str::trim) { + match name { + "switchDelaySecs" => force.switch_delay = true, + "supersede" => force.supersede = true, + other => { + return Err(format!( + "unknown force {other:?}: use true, or a list of \ + switchDelaySecs and supersede" + )); + } + } + } + } + } + Ok(force) + } +} + +/// An expired remake of the epoch a proposal would supersede. +#[derive(Debug)] +struct ExpiredRemake { + id: String, + coordinator: String, + /// When it expired (Unix milliseconds). + since: u64, + tally: vote::Tally, +} + #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct Report { @@ -93,10 +213,6 @@ struct Report { phase: Phase, #[serde(flatten)] record: Record, - /// Size of the kind's build input (e.g. a replacement document), which reports - /// leave out. - #[serde(skip_serializing_if = "Option::is_none")] - content_bytes: Option<usize>, /// What happened to this remake on the node answering the request. #[serde(skip_serializing_if = "Option::is_none")] local: Option<LocalStatus>, @@ -106,6 +222,12 @@ struct Report { enum StartError { #[error("remake {0} is already open for this epoch")] Open(String), + /// Expired remakes still hold votes; superseding them takes `force=supersede`. + #[error("expired remakes of this epoch still hold votes")] + Expired(Vec<ExpiredRemake>), + /// 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)] @@ -128,12 +250,10 @@ fn grace<H: Host>(host: &H) -> u64 { host.dormant_after().as_millis() as u64 } -fn report<H: Host>(host: &H, id: String, mut record: Record) -> Report { - let content = std::mem::take(&mut record.proposal.content); +fn report<H: Host>(host: &H, id: String, record: Record) -> Report { Report { phase: record.phase(unix_millis(), grace(host)), local: host.remake().status(&id), - content_bytes: (!content.is_null()).then(|| content.to_string().len()), id, record, } @@ -186,31 +306,55 @@ 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 { if !host.is_node_ready().await { return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None); } + let force = match Force::parse(query.force.as_deref()) { + Ok(force) => force, + Err(reason) => return error_response(StatusCode::BAD_REQUEST, reason, 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 +364,7 @@ async fn start_kind<H: Host>( StatusCode::BAD_REQUEST, format!("invalid request: {error}"), None, - ) + ); } } }; @@ -247,7 +391,7 @@ async fn start_kind<H: Host>( StatusCode::BAD_REQUEST, format!("invalid {kind_name} spec: {reason}"), None, - ) + ); } }; @@ -269,34 +413,20 @@ 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. let keepalive = dormant_after / 4; let switch_delay = request.switch_delay_secs.map(Duration::from_secs); match switch_delay { - Some(delay) if delay < keepalive && query.force => { + Some(delay) if delay < keepalive && force.switch_delay => { warn!( switch_delay_secs = delay.as_secs(), keepalive_secs = keepalive.as_secs(), @@ -307,7 +437,7 @@ async fn start_kind<H: Host>( return error_response( StatusCode::BAD_REQUEST, format!( - "switchDelaySecs must be at least {} (one keepalive interval), so the commit reaches every node before they switch; add ?force=true to override", + "switchDelaySecs must be at least {} (one keepalive interval), so the commit reaches every node before they switch; add ?force=switchDelaySecs to override", keepalive.as_secs() ), None, @@ -316,30 +446,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 proposal = Proposal { + let mut proposal = Proposal { coordinator: host.node_id(), from_epoch: host.current_epoch().into(), base_heads: Vec::new(), @@ -355,11 +467,21 @@ 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, + content: None, translation: prepared.translation, }; + // The content stays here, referred to by hash; voters fetch it from us. + if let Some(bytes) = prepared.content { + let reference = ContentRef::of(&bytes); + if let Some(hash) = reference.hash() { + host.remake().put_content(hash, std::sync::Arc::new(bytes)); + } + proposal.content = Some(reference); + } // From here on, nothing depends on the client: writing the proposal and starting the // coordinator run in their own task, so a client that disconnects mid-request can't @@ -369,6 +491,7 @@ async fn start_kind<H: Host>( id.clone(), proposal, base, + force.supersede, now, grace, )); @@ -393,6 +516,38 @@ async fn start_kind<H: Host>( Some(json!({ "remake": open })), ); } + Err(StartError::Expired(expired)) => { + let active: HashSet<String> = host + .peer_summaries() + .await + .into_iter() + .filter(|peer| peer.active) + .map(|peer| peer.id) + .collect(); + let me = host.node_id(); + let expired: Vec<JsonValue> = expired + .iter() + .map(|remake| { + json!({ + "id": remake.id, + "coordinator": remake.coordinator, + "coordinatorActive": remake.coordinator == me + || active.contains(&remake.coordinator), + "expiredAt": remake.since, + "tally": remake.tally.to_string(), + }) + }) + .collect(); + return error_response( + StatusCode::CONFLICT, + "an expired remake of this epoch still holds votes: abort it, or add \ + ?force=supersede to abort it and propose in its place", + Some(json!({ "expired": expired, "force": "supersede" })), + ); + } + 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); } @@ -435,17 +590,69 @@ async fn propose<H: Host>( id: String, mut proposal: Proposal, base: Option<Vec<ChangeHash>>, + supersede: bool, now: u64, grace: u64, ) -> Result<Record, StartError> { 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 mut 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 mut register = records::load_register(doc); + + // Expired remakes hold their votes; their coordinators can't commit them + // any more, so they may be aborted here. + let expired: Vec<&String> = all + .iter() + .filter(|(_, record)| { + record.proposal.from_epoch == proposal.from_epoch + && record.phase(now, grace) == Phase::Expired + }) + .map(|(id, _)| id) + .collect(); + if !expired.is_empty() { + if !supersede { + let details = all + .iter() + .filter(|(id, _)| expired.contains(&id)) + .map(|(id, record)| ExpiredRemake { + id: id.clone(), + coordinator: record.proposal.coordinator.clone(), + since: record.proposal.settle_until.saturating_add(grace), + tally: vote::tally(id, record, &all, ®ister), + }) + .collect(); + return Err(StartError::Expired(details)); + } + let decision = Decision::Aborted { + reason: format!("superseded by remake {record_id}"), + decided_at: now, + }; + for id in expired { + records::write_decision(doc, id, &decision)?; + info!(remake = %id, by = %record_id, "Superseded expired remake"); + } + all = records::load_all(doc); + register = records::load_register(doc); + } + + match vote::freedom(&me, &proposal.from_epoch, &all, ®ister, 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 @@ -466,11 +673,26 @@ 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 { + remake: record_id.clone(), + at: now, + }; + records::write_vote(doc, &me, &own_vote)?; Ok(Record { proposal, decision: None, abort_requested: None, + contested: false, + votes: [(me, own_vote)].into(), }) }) .await?; @@ -515,22 +737,18 @@ pub(crate) async fn abort<H: Host>( } /// `DELETE /admin/remakes`: aborts whichever remake of the local epoch is still -/// proposed. There is at most one, unless two coordinators raced (then both abort anyway -/// and this picks the oldest). +/// undecided (proposed or expired), the oldest if there are several. pub(crate) async fn abort_current<H: Host>( State(host): State<H>, Query(query): Query<WaitQuery>, ) -> Response { let local = EpochRecord::from(host.current_epoch()); - let now = unix_millis(); - let grace = grace(&host); let open = host .read_doc(move |doc| { records::load_all(doc) .into_iter() .filter(|(_, record)| { - record.proposal.from_epoch == local - && record.phase(now, grace) == Phase::Proposed + record.proposal.from_epoch == local && record.decision.is_none() }) .min_by_key(|(_, record)| record.proposal.proposed_at) .map(|(id, _)| id) @@ -554,24 +772,25 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { match current.phase { Phase::Proposed => {} + Phase::Expired => return abort_expired(&host, &id).await, Phase::Committed => { return error_response( StatusCode::CONFLICT, "already committed; the switch cannot be undone", Some(json!({ "remake": current })), - ) + ); } - Phase::Aborted | Phase::Expired => { + Phase::Aborted => { return error_response( StatusCode::CONFLICT, "already over", Some(json!({ "remake": current })), - ) + ); } } // The coordinator applies the request on its next check (within about a second). - // If it is gone, the proposal expires anyway. + // If it is gone, the proposal expires, and can then be aborted right away. if current.record.abort_requested.is_none() { let request = AbortRequest { by: host.node_id(), @@ -604,6 +823,47 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { } } +/// Aborts expired remake `id` from this node: its coordinator is gone, and can't commit +/// it any more (`coordinator`). Answers with the final report. +async fn abort_expired<H: Host>(host: &H, id: &str) -> Response { + let record_id = id.to_owned(); + let decision = Decision::Aborted { + reason: format!( + "aborted by operator via {} after it expired", + host.node_id() + ), + decided_at: unix_millis(), + }; + let grace = grace(host); + let written = host + .commit_doc(move |doc| -> Result<bool, StoreError> { + let expired = records::load(doc, &record_id) + .is_some_and(|record| record.phase(unix_millis(), grace) == Phase::Expired); + if expired { + records::write_decision(doc, &record_id, &decision)?; + } + Ok(expired) + }) + .await; + match written { + Ok(true) => info!(remake = %id, "Aborted expired remake"), + // Decided meanwhile: the report says how. + Ok(false) => {} + Err(error) => { + error!(remake = %id, error = %format!("{error:#}"), "Failed to abort expired remake"); + return error_response( + StatusCode::INTERNAL_SERVER_ERROR, + "failed to record the abort", + None, + ); + } + } + match load_report(host, id).await { + Some(report) => Json(report).into_response(), + None => error_response(StatusCode::NOT_FOUND, "no such remake", None), + } +} + /// `GET /admin/epochs`: the local epoch, the previous epochs this node still keeps, and /// an ongoing catch-up with a peer's newer epoch (`null` if none). pub(crate) async fn epochs<H: Host>(State(host): State<H>) -> Response { diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index 4aa66e7..cad5a22 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -2,13 +2,17 @@ //! //! Peers on another epoch don't sync, so a node that missed a switch (new, late, //! returning, or one whose own build failed) can't follow through ordinary sync. Once -//! an active peer advertises a newer generation, the node +//! active, ready members advertise a newer generation (two of them for a node that is +//! ready itself, one for a node that never was; at most `MAX_GENERATION_GAP` ahead), +//! the node //! //! 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): @@ -23,33 +27,44 @@ //! If no peer completes a sync in time, or the check fails, the node restores its old //! document and epoch and tries again later. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; use std::convert::Infallible; use std::time::Duration; use automerge::{AutoCommit, ChangeHash}; use tokio::time::Instant; -use tracing::{error, info, warn}; +use tracing::{debug, 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. const RETRY: Duration = Duration::from_secs(30); /// Lower bound for how long a rejoin may take to complete a sync. const MIN_REJOIN_TIMEOUT: Duration = Duration::from_secs(60); +/// A peer more than this many generations ahead is lying or broken (every generation +/// is a committed remake); its epoch is ignored instead of followed. +const MAX_GENERATION_GAP: u64 = 1000; +/// Peers that must advertise a newer epoch before a ready node leaves its document for +/// it (fewer if it has fewer active members). A node that was never ready has nothing +/// to lose and follows the first peer it sees. +const CORROBORATION: usize = 2; /// Called by the switch worker when there is nothing to switch to locally, and on every /// tick while a rejoin is in progress. 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 { @@ -65,27 +80,62 @@ pub(super) async fn check<H: Host>(host: &H) { } let local = host.current_epoch(); - let mut ahead: Vec<(String, Epoch)> = host - .peer_summaries() - .await + let peers = host.peer_summaries().await; + let active_members = peers + .iter() + .filter(|peer| peer.active && peer.member) + .count(); + // Only ready members count: a rejoining peer advertises an epoch it doesn't hold + // yet, and a node of another swarm (or a stranger) advertises one that isn't ours. + let mut ahead: Vec<(String, Epoch)> = peers .into_iter() - .filter(|peer| peer.active) + .filter(|peer| peer.serves()) .filter_map(|peer| peer.epoch.map(|epoch| (peer.id, epoch))) .filter(|(_, epoch)| epoch.generation > local.generation) .collect(); ahead.sort_by(|a, b| a.0.cmp(&b.0)); + ahead.retain(|(peer, epoch)| { + let plausible = epoch.generation - local.generation <= MAX_GENERATION_GAP; + if !plausible { + warn!(peer = %peer, epoch = %epoch, local = %local, "Peer advertises an implausible epoch; ignoring it"); + } + plausible + }); + + // One peer's word is not enough for a ready node to give up its document: a buggy + // or hostile member could otherwise keep the swarm rejoining an epoch nobody + // serves. A newer epoch is followed once enough distinct peers advertise it. + let required = if host.is_node_ready().await { + active_members.clamp(1, CORROBORATION) + } else { + 1 + }; + let mut advertised: HashMap<Epoch, usize> = HashMap::new(); + for (_, epoch) in &ahead { + *advertised.entry(*epoch).or_default() += 1; + } + let corroborated: Vec<Epoch> = advertised + .into_iter() + .filter(|(_, count)| *count >= required) + .map(|(epoch, _)| epoch) + .collect(); - let Some(generation) = ahead.iter().map(|(_, epoch)| epoch.generation).max() else { + let Some(generation) = corroborated.iter().map(|epoch| epoch.generation).max() else { if host.remake().catch_up().is_some() { host.remake().set_catch_up(None); } + if !ahead.is_empty() { + debug!( + peers = ahead.len(), + required, "A newer epoch is advertised, but not by enough peers yet" + ); + } return; }; - let mut targets: Vec<Epoch> = ahead - .iter() - .filter(|(_, epoch)| epoch.generation == generation) - .map(|(_, epoch)| *epoch) + let mut targets: Vec<Epoch> = corroborated + .into_iter() + .filter(|epoch| epoch.generation == generation) .collect(); targets.sort_by_key(|epoch| epoch.id); targets.dedup(); @@ -147,6 +197,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 +533,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). @@ -465,7 +554,7 @@ mod tests { policy, proposed_at: 0, kind: "compact", - content: &JsonValue::Null, + content: None, translation: &JsonValue::Null, }, ) diff --git a/src/remake/content.rs b/src/remake/content.rs new file mode 100644 index 0000000..e5d78c5 --- /dev/null +++ b/src/remake/content.rs @@ -0,0 +1,95 @@ +//! Content of remakes: the new document of a `replace` or `import` remake. +//! +//! The proposal records only a reference (BLAKE3 hash and size). The coordinator keeps +//! the bytes in memory; every member fetches them before voting (in chunks, from +//! several peers at once: `swarm::blob`) and serves them in turn. A committed remake +//! therefore has its content on a majority of the electorate. Nodes that switch need +//! it to build the snapshot; nodes that rejoin later don't (they download the snapshot +//! itself). +//! +//! Content no open remake refers to is dropped an hour after it was stored. + +use std::collections::HashSet; +use std::sync::Arc; +use std::time::Duration; + +use tracing::{info, warn}; + +use super::Host; +use super::records::{ContentRef, Record}; + +/// The content `reference` names, if this node has it; otherwise starts fetching it in +/// the background (once) and returns None. +pub(super) fn available_or_fetch<H: Host>( + host: &H, + reference: &ContentRef, +) -> Option<Arc<Vec<u8>>> { + let hash = reference.hash()?; + if let Some(bytes) = host.remake().content(&hash) { + return Some(bytes); + } + if host.remake().start_content_fetch(hash) { + let host = host.clone(); + let reference = reference.clone(); + tokio::spawn(async move { + fetch(&host, hash, &reference).await; + host.remake().end_content_fetch(&hash); + }); + } + None +} + +/// The content `reference` names: from this node, or fetched now. Waits for a download +/// already running. +pub(super) async fn ensure<H: Host>( + host: &H, + reference: &ContentRef, +) -> Result<Arc<Vec<u8>>, String> { + let hash = reference + .hash() + .ok_or_else(|| format!("unreadable content hash {:?}", reference.blake3))?; + loop { + if let Some(bytes) = host.remake().content(&hash) { + return Ok(bytes); + } + if host.remake().start_content_fetch(hash) { + let fetched = fetch(host, hash, reference).await; + host.remake().end_content_fetch(&hash); + return fetched + .ok_or_else(|| format!("couldn't download content {}", reference.blake3)); + } + // Someone else is downloading it. + while host.remake().content_fetch_running(&hash) { + tokio::time::sleep(Duration::from_millis(500)).await; + } + } +} + +async fn fetch<H: Host>(host: &H, hash: [u8; 32], reference: &ContentRef) -> Option<Arc<Vec<u8>>> { + match host.fetch_content(hash).await { + Ok(bytes) if bytes.len() as u64 == reference.size => { + info!(content = %reference.blake3, bytes = bytes.len(), "Downloaded remake content"); + let bytes = Arc::new(bytes); + host.remake().put_content(hash, Arc::clone(&bytes)); + Some(bytes) + } + Ok(bytes) => { + warn!(content = %reference.blake3, bytes = bytes.len(), expected = reference.size, "Remake content has the wrong size"); + None + } + Err(error) => { + warn!(content = %reference.blake3, error = %format!("{error:#}"), "Couldn't download remake content"); + None + } + } +} + +/// Drops content no open remake in `records` refers to (after its grace period). +pub(super) fn prune<H: Host>(host: &H, records: &[(String, Record)], now: u64, grace: u64) { + let referenced: HashSet<[u8; 32]> = records + .iter() + .filter(|(_, record)| record.phase(now, grace).is_open()) + .filter_map(|(_, record)| record.proposal.content.as_ref()?.hash()) + .collect(); + host.remake().prune_contents(&referenced); +} diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs index d1b3397..9bcf8ba 100644 --- a/src/remake/coordinator.rs +++ b/src/remake/coordinator.rs @@ -1,53 +1,90 @@ -//! 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, +//! * the voting period (`settleSecs`) passed without a majority, or +//! * a majority is out of reach (too many members voted for a competing remake). //! -//! 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. +//! The decision is written in the same step that checks the records once more, so a +//! commit never overrides an abort request or a competitor's commit that arrived while +//! the snapshot was being built. Nor is a commit written once the remake has expired: +//! from then on, any node may abort it (`api`), and the two mustn't race. //! -//! The task lives only in memory. If the coordinator dies, nobody decides and the -//! proposal expires. +//! 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. A coordinator that restarts takes its undecided +//! remakes up again once ready (`resume`); past the voting period, that means aborting +//! them. If it is gone for good, nobody decides and the proposal expires; its votes +//! stay cast (`vote`). -use std::collections::HashMap; use std::time::Duration; use tracing::{info, warn}; -use super::records::{self, Decision, Record}; -use super::snapshot; use super::Host; +use super::content; +use super::records::{self, Decision, EpochRecord, Phase, Record}; +use super::snapshot; +use super::vote; use crate::document::store::StoreError; const TICK: Duration = Duration::from_secs(1); +/// Coordinates remake `id` until it is decided; returns at once if this node already +/// does. pub(crate) async fn run<H: Host>(host: H, id: String) { + if !host.remake().start_coordinating(&id) { + return; + } info!(remake = %id, "Coordinating remake"); loop { tokio::time::sleep(TICK).await; match step(&host, &id).await { - Ok(true) => return, + Ok(true) => break, Ok(false) => {} Err(error) => { warn!(remake = %id, error = %format!("{error:#}"), "Coordinator step failed; retrying"); } } } + host.remake().end_coordinating(&id); +} + +/// Takes up coordinating this node's undecided remakes of `epoch` among `all` that it +/// doesn't coordinate (it restarted since proposing them). Called by the remake worker +/// once the node is ready, so its records are complete. +pub(super) fn resume<H: Host>(host: &H, epoch: &EpochRecord, all: &[(String, Record)]) { + let me = host.node_id(); + for (id, record) in all { + if record.decision.is_none() + && record.proposal.coordinator == me + && &record.proposal.from_epoch == epoch + && !host.remake().is_coordinating(id) + { + info!(remake = %id, "Resuming coordination of own remake"); + tokio::spawn(run(host.clone(), id.clone())); + } + } } /// 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 { @@ -58,77 +95,53 @@ async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> { return Ok(true); } - 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 EpochRecord::from(host.current_epoch()) != record.proposal.from_epoch { + decide_abort(host, id, "local epoch changed during the vote".to_owned()).await?; + return Ok(true); + } + if let Some(reason) = obstacle(id, &record, &all) { 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, ®ister); + // The deadline first: a majority counted after it is too late. + if records::unix_millis() >= record.proposal.settle_until { + decide_abort( + host, + id, + format!("no majority in the voting period: {tally}"), + ) + .await?; + return Ok(true); } - - commit(host, id, &record).await?; - Ok(true) + if tally.reached() { + info!(remake = %id, votes = %tally, "Majority reached"); + commit(host, id, &record).await?; + return Ok(true); + } + if tally.out_of_reach() { + decide_abort(host, id, format!("a majority is out of reach: {tally}")).await?; + return Ok(true); + } + Ok(false) } -async fn abort_reason<H: Host>( - host: &H, - id: &str, - record: &Record, - all: &[(String, Record)], - now: u64, - grace: u64, -) -> Option<String> { - let proposal = &record.proposal; - +/// Why remake `id` must be aborted whatever its votes, if it must. +fn obstacle(id: &str, record: &Record, all: &[(String, Record)]) -> Option<String> { if let Some(request) = &record.abort_requested { return Some(format!("aborted by operator via {}", request.by)); } - - if records::EpochRecord::from(host.current_epoch()) != proposal.from_epoch { - return Some("local epoch changed during settle".to_owned()); - } - - if let Some((other, _)) = all.iter().find(|(other, candidate)| { + let (other, _) = all.iter().find(|(other, candidate)| { other != id - && candidate.proposal.from_epoch == proposal.from_epoch - && candidate.phase(now, grace).is_open() - }) { - 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()); - } - - None + && candidate.proposal.from_epoch == record.proposal.from_epoch + && candidate.committed_epoch().is_some() + })?; + Some(format!("competing remake {other} committed")) } async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), StoreError> { - let Some(request) = record.snapshot_request(id) else { + let Some(mut request) = record.snapshot_request(id) else { return decide_abort( host, id, @@ -136,6 +149,12 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor ) .await; }; + if let Some(reference) = &record.proposal.content { + match content::ensure(host, reference).await { + Ok(bytes) => request.content = Some(bytes), + Err(reason) => return decide_abort(host, id, reason).await, + } + } // Clone under the read lock, build without any lock. let copy = host.read_doc(|doc| doc.clone()).await; @@ -154,30 +173,74 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor snapshot_bytes, decided_at: records::unix_millis(), }; - write_decision(host, id, decision).await?; - - // The local switch reuses it instead of building again. - host.remake().store_prebuilt(id, snapshot); - - info!(remake = %id, to_epoch = %to_epoch, snapshot_bytes, "Remake committed"); + match decide(host, id, decision).await? { + Some(Decision::Committed { .. }) => { + // The local switch reuses it instead of building again. + host.remake().store_prebuilt(id, snapshot); + info!(remake = %id, to_epoch = %to_epoch, snapshot_bytes, "Remake committed"); + } + Some(Decision::Aborted { reason, .. }) => { + warn!(remake = %id, reason = %reason, "Remake aborted while building its snapshot"); + } + None => {} + } Ok(()) } async fn decide_abort<H: Host>(host: &H, id: &str, reason: String) -> Result<(), StoreError> { - warn!(remake = %id, reason = %reason, "Remake aborted"); - write_decision( - host, - id, - Decision::Aborted { - reason, - decided_at: records::unix_millis(), - }, - ) - .await + let decision = Decision::Aborted { + reason: reason.clone(), + decided_at: records::unix_millis(), + }; + if decide(host, id, decision).await?.is_some() { + warn!(remake = %id, reason = %reason, "Remake aborted"); + } + Ok(()) } -async fn write_decision<H: Host>(host: &H, id: &str, decision: Decision) -> Result<(), StoreError> { +/// Writes `decision`, checked again under the write lock against the records as they +/// are now (building the snapshot takes a while): a commit becomes an abort if the +/// operator asked for one, a competing remake committed, the remake expired, or the +/// majority is gone. +/// Returns what was written; nothing if the remake is decided already or gone. +async fn decide<H: Host>( + host: &H, + id: &str, + decision: Decision, +) -> Result<Option<Decision>, StoreError> { let id = id.to_owned(); - host.commit_doc(move |doc| records::write_decision(doc, &id, &decision)) - .await + let grace = host.dormant_after().as_millis() as u64; + host.commit_doc(move |doc| -> Result<Option<Decision>, StoreError> { + let all = records::load_all(doc); + let Some((_, record)) = all.iter().find(|(other, _)| *other == id) else { + return Ok(None); + }; + if record.decision.is_some() { + return Ok(None); + } + let decision = match decision { + Decision::Committed { decided_at, .. } => { + let register = records::load_register(doc); + let tally = vote::tally(&id, record, &all, ®ister); + let expired = record.phase(records::unix_millis(), grace) == Phase::Expired; + match obstacle(&id, record, &all) { + Some(reason) => Decision::Aborted { reason, decided_at }, + // Others may abort it by now (`api`): too late to commit. + None if expired => Decision::Aborted { + reason: format!("expired while building the snapshot: {tally}"), + decided_at, + }, + None if !tally.reached() => Decision::Aborted { + reason: format!("the majority was lost: {tally}"), + decided_at, + }, + None => decision, + } + } + aborted @ Decision::Aborted { .. } => aborted, + }; + records::write_decision(doc, &id, &decision)?; + Ok(Some(decision)) + }) + .await } diff --git a/src/remake/delta.rs b/src/remake/delta.rs index c161e2c..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. @@ -268,7 +271,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..8131f7f 100644 --- a/src/remake/kind.rs +++ b/src/remake/kind.rs @@ -10,27 +10,30 @@ //! * 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. pub(crate) const DEFAULT_KIND: &str = "compact"; -/// What the coordinator stores in the proposal for every node. +/// What the coordinator prepares for every node. #[derive(Debug, Default)] pub(crate) struct Prepared { - /// Input to [`RemakeKind::build`]; replicated with the proposal (old epoch only). - pub(crate) content: JsonValue, + /// Input to [`RemakeKind::build`], if the kind takes any. The proposal refers to it + /// by hash; voters fetch it from the coordinator and each other (`content`). + pub(crate) content: Option<Vec<u8>>, /// Input to [`RemakeKind::translate`]; also kept in the new epoch's metadata, so /// nodes that rejoin later can translate their writes. pub(crate) translation: JsonValue, @@ -62,7 +65,7 @@ pub(crate) trait RemakeKind: Send + Sync + 'static { fn build( &self, old_at_base: &AutoCommit, - content: &JsonValue, + content: Option<&[u8]>, target: &mut AutoCommit, ) -> Result<(), StoreError>; @@ -75,6 +78,7 @@ pub(crate) fn by_name(name: &str) -> Option<&'static dyn RemakeKind> { match name { "compact" => Some(&Compact), "replace" => Some(&Replace), + "import" => Some(&Import), _ => None, } } @@ -104,7 +108,7 @@ impl RemakeKind for Compact { fn build( &self, old_at_base: &AutoCommit, - _content: &JsonValue, + _content: Option<&[u8]>, target: &mut AutoCommit, ) -> Result<(), StoreError> { store::materialize_into(old_at_base, target, &is_reserved) @@ -196,45 +200,26 @@ 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 }); - } - + let content = + serde_json::to_vec(&canonical(spec.document)).map_err(|error| error.to_string())?; Ok(Prepared { - content: canonical(spec.document), - translation: serde_json::to_value(Rules { rules }) - .map_err(|error| error.to_string())?, + content: Some(content), + translation: prepare_rules(spec.rules)?, }) } fn build( &self, _old_at_base: &AutoCommit, - content: &JsonValue, + content: Option<&[u8]>, target: &mut AutoCommit, ) -> Result<(), StoreError> { - // Canonical again: the content went through the old epoch's document, and the - // operations (hence the epoch ID) must not depend on any key order. - let JsonValue::Object(document) = canonical(content.clone()) else { + let content = content.ok_or_else(|| anyhow::anyhow!("replace without content"))?; + let document: JsonValue = serde_json::from_slice(content) + .map_err(|error| anyhow::anyhow!("replacement document: {error}"))?; + // Canonical again: the operations (hence the epoch ID) must not depend on any + // key order. + let JsonValue::Object(document) = canonical(document) else { return Ok(()); // propose only accepts objects }; for (key, value) in document { @@ -246,31 +231,130 @@ 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 a request (base64 of a saved Automerge document). +fn load_encoded(encoded: &str) -> Result<AutoCommit, String> { + 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_encoded(&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 every + // voter downloads. + 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: Some(content.save()), + translation: prepare_rules(spec.rules)?, + }) + } + + fn build( + &self, + _old_at_base: &AutoCommit, + content: Option<&[u8]>, + target: &mut AutoCommit, + ) -> Result<(), StoreError> { + let content = content.ok_or_else(|| anyhow::anyhow!("import without content"))?; + let imported = AutoCommit::load(content)?; + // 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)] @@ -332,7 +416,7 @@ mod tests { fn build( &self, _old: &AutoCommit, - _content: &JsonValue, + _content: Option<&[u8]>, _target: &mut AutoCommit, ) -> Result<(), StoreError> { Ok(()) @@ -399,10 +483,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,13 +532,70 @@ 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.as_deref(), &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| { let prepared = replace(json!({ "document": document })).unwrap(); let mut target = AutoCommit::new().with_actor(automerge::ActorId::from(&b"fixed"[..])); Replace - .build(&AutoCommit::new(), &prepared.content, &mut target) + .build(&AutoCommit::new(), prepared.content.as_deref(), &mut target) .unwrap(); target.commit_with(automerge::transaction::CommitOptions::default().with_time(0)); ( diff --git a/src/remake/local.rs b/src/remake/local.rs index 35a4ff1..04ac0b0 100644 --- a/src/remake/local.rs +++ b/src/remake/local.rs @@ -1,27 +1,39 @@ //! Per-node, in-memory remake state: snapshots built ahead of the switch, switch -//! progress for reports, a rejoin in progress, quarantined changes, and archived -//! documents of previous epochs. +//! progress for reports, a rejoin in progress, quarantined changes, archived documents +//! of previous epochs, and the content of remakes (`content`). //! //! Nothing here is replicated or durable. After a restart the node simply has no //! archive or quarantine, and rebuilds snapshots when needed. -use std::collections::HashMap; -use std::sync::{Mutex, MutexGuard}; +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; 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; -#[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. @@ -34,9 +46,21 @@ 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>, + /// Content of remakes, by BLAKE3 hash, and when it was stored (served to peers). + contents: HashMap<[u8; 32], (Arc<Vec<u8>>, Instant)>, + /// Content being downloaded. + content_fetches: HashSet<[u8; 32]>, + /// Remakes this node coordinates now (`coordinator`). + coordinating: HashSet<String>, } +/// Content no remake refers to any more is kept this long after it was stored, for +/// nodes that switch late. +const CONTENT_GRACE: Duration = Duration::from_secs(3600); + /// Catching up with a newer epoch: attempts and the last failure. #[derive(Clone, Debug)] pub(crate) struct CatchUp { @@ -119,6 +143,23 @@ pub(crate) struct ArchivedEpoch { } impl RemakeState { + pub(crate) fn started_at(&self) -> u64 { + self.started_at + } + + /// Whether this node starts coordinating `id` now (false if it already does). + pub(crate) fn start_coordinating(&self, id: &str) -> bool { + self.lock().coordinating.insert(id.to_owned()) + } + + pub(crate) fn end_coordinating(&self, id: &str) { + self.lock().coordinating.remove(id); + } + + pub(crate) fn is_coordinating(&self, id: &str) -> bool { + self.lock().coordinating.contains(id) + } + fn lock(&self) -> MutexGuard<'_, Inner> { self.inner .lock() @@ -211,6 +252,54 @@ impl RemakeState { self.lock().rejoin.take() } + pub(crate) fn content(&self, hash: &[u8; 32]) -> Option<Arc<Vec<u8>>> { + self.lock() + .contents + .get(hash) + .map(|(bytes, _)| Arc::clone(bytes)) + } + + pub(crate) fn put_content(&self, hash: [u8; 32], bytes: Arc<Vec<u8>>) { + self.lock().contents.insert(hash, (bytes, Instant::now())); + } + + /// Marks a download of `hash` as started; false if one already is. + pub(crate) fn start_content_fetch(&self, hash: [u8; 32]) -> bool { + self.lock().content_fetches.insert(hash) + } + + pub(crate) fn end_content_fetch(&self, hash: &[u8; 32]) { + self.lock().content_fetches.remove(hash); + } + + pub(crate) fn content_fetch_running(&self, hash: &[u8; 32]) -> bool { + self.lock().content_fetches.contains(hash) + } + + /// Drops content that no remake in `referenced` needs, once its grace has passed. + pub(crate) fn prune_contents(&self, referenced: &HashSet<[u8; 32]>) { + self.lock().contents.retain(|hash, (_, stored)| { + referenced.contains(hash) || stored.elapsed() < CONTENT_GRACE + }); + } + + /// 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 ef036a6..ca17956 100644 --- a/src/remake/mod.rs +++ b/src/remake/mod.rs @@ -2,10 +2,12 @@ //! 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. +//! * `content`: the new document of a `replace`/`import` remake, fetched by every voter. //! * `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. @@ -19,6 +21,7 @@ mod api; mod catchup; +mod content; mod coordinator; mod delta; mod kind; @@ -27,6 +30,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 +41,9 @@ use std::time::Duration; use automerge::AutoCommit; use axum::{ - routing::{get, post}, Router, + extract::DefaultBodyLimit, + routing::{get, post}, }; use tokio::sync::broadcast; @@ -58,9 +63,28 @@ 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>; + + /// 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; + + /// Downloads content by its BLAKE3 hash from peers on the local epoch, checked + /// against it. + fn fetch_content(&self, hash: [u8; 32]) + -> impl Future<Output = anyhow::Result<Vec<u8>>> + 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. @@ -117,18 +141,45 @@ pub(crate) struct PeerSummary { pub(crate) id: String, /// Not dormant. pub(crate) active: bool, + /// Completed the handshake and belongs to our swarm. A peer that merely sent us + /// something (a node of another swarm on the same network, a stranger before + /// admission) is not one, and nothing it advertises is followed. + pub(crate) member: bool, + /// What it last said about being ready; None before it said anything. + pub(crate) ready: Option<bool>, /// The epoch the peer last advertised. pub(crate) epoch: Option<Epoch>, } +impl PeerSummary { + /// A peer whose epoch can be followed and downloaded from: an active member that + /// holds the document it advertises. A rejoining peer advertises its target epoch + /// with an empty document and `ready: false`. + pub(crate) fn serves_epoch(&self, epoch: Epoch) -> bool { + self.serves() && self.epoch == Some(epoch) + } + + /// An active, ready member. + pub(crate) fn serves(&self) -> bool { + self.active && self.member && self.ready == Some(true) + } +} + /// 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..cd3e5be 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -8,6 +8,16 @@ //! * `abortRequested`: written by any node on an operator's request. It is an input to //! the coordinator's decision, not a decision itself. //! +//! Once a remake has expired, its coordinator no longer commits it, and any node may +//! abort it on an operator's request (`api`). Should clocks disagree by more than the +//! grace margin, a commit and such an abort could still be written concurrently. Such +//! a record is *contested*: it reads as committed, and nodes don't switch to it on +//! their own. +//! +//! 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, //! never written, so a late but valid commit can't collide with someone's abort. @@ -15,13 +25,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, VOTES}; use crate::document::replica::Epoch; use crate::document::store::{self, StoreError}; @@ -127,19 +139,44 @@ 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, - /// The kind's build input. - #[serde(default, skip_serializing_if = "JsonValue::is_null")] - pub(crate) content: JsonValue, + /// The kind's build input, if it has one: held by the coordinator and fetched by + /// every voter (`content`), not stored in the record. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) content: Option<ContentRef>, /// The kind's translation input. #[serde(default, skip_serializing_if = "JsonValue::is_null")] pub(crate) translation: JsonValue, } +/// Content of a remake, by reference. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ContentRef { + /// BLAKE3 of the bytes, hex. + pub(crate) blake3: String, + pub(crate) size: u64, +} + +impl ContentRef { + pub(crate) fn of(bytes: &[u8]) -> Self { + Self { + blake3: blake3::hash(bytes).to_hex().to_string(), + size: bytes.len() as u64, + } + } + + pub(crate) fn hash(&self) -> Option<[u8; 32]> { + decode_hex32(&self.blake3) + } +} + fn default_kind() -> String { DEFAULT_KIND.to_owned() } @@ -173,6 +210,18 @@ pub(crate) struct AbortRequest { pub(crate) at: u64, } +/// 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 { @@ -181,6 +230,18 @@ pub(crate) struct Record { pub(crate) decision: Option<Decision>, #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) abort_requested: Option<AbortRequest>, + /// Decisions written concurrently disagree; `decision` is then the commit. Not + /// stored: found when the record is read. + #[serde( + default, + skip_deserializing, + skip_serializing_if = "std::ops::Not::not" + )] + pub(crate) contested: bool, + /// 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>, } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] @@ -214,8 +275,8 @@ impl Record { } } - /// Everything needed to (re)build this remake's snapshot; None if the proposal - /// is unreadable. + /// Everything needed to (re)build this remake's snapshot but the content (which the + /// caller adds, `content::ensure`); None if the proposal is unreadable. pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> { Some(SnapshotRequest { remake_id: id.to_owned(), @@ -224,7 +285,7 @@ impl Record { policy: serde_json::to_value(&self.proposal.policy).ok()?, proposed_at: self.proposal.proposed_at, kind: self.proposal.kind.clone(), - content: self.proposal.content.clone(), + content: None, translation: self.proposal.translation.clone(), }) } @@ -254,8 +315,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, ®ister))) + }) + .collect() +} + +fn with_votes(id: &str, mut record: Record, register: &Register) -> Record { + record.votes = register + .iter() + .filter_map(|(voter, votes)| { + let vote = votes.iter().find(|vote| vote.remake == id)?; + Some((voter.clone(), vote.clone())) + }) + .collect(); + record +} + +fn load_record(doc: &AutoCommit, id: &str) -> Option<Record> { let value = match store::read_path(doc, &record_path(id, "")) { Ok(Some(value)) => value, Ok(None) => return None, @@ -264,21 +354,73 @@ pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> { return None; } }; - match serde_json::from_value(value) { - Ok(record) => Some(record), + let mut record: Record = match serde_json::from_value(value) { + Ok(record) => record, Err(error) => { warn!(remake = %id, error = %error, "Skipping unreadable remake record"); - None + return None; } + }; + if let Some(commit) = contested_decision(doc, id) { + record.decision = Some(commit); + record.contested = true; } + Some(record) } -/// All readable records. -pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> { - store::keys_at(doc, &records_dir()) +/// A commit among concurrent decisions of remake `id` that disagree, if they do. +fn contested_decision(doc: &AutoCommit, id: &str) -> Option<Decision> { + let decisions: Vec<Decision> = store::read_candidates(doc, &record_path(id, "decision")) + .ok()? + .into_iter() + .filter_map(|value| serde_json::from_value(value).ok()) + .collect(); + let committed: Vec<&EpochRecord> = decisions + .iter() + .filter_map(|decision| match decision { + Decision::Committed { to_epoch, .. } => Some(to_epoch), + Decision::Aborted { .. } => None, + }) + .collect(); + let agree = committed.is_empty() + || (committed.len() == decisions.len() && committed.iter().all(|e| *e == committed[0])); + if agree { + return None; + } + decisions + .into_iter() + .find(|decision| matches!(decision, Decision::Committed { .. })) +} + +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, ®ister_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() } @@ -316,6 +458,12 @@ pub(crate) fn write_abort_request( write_part(doc, id, "abortRequested", request) } +/// 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)?; + store::write_path(doc, ®ister_path(voter), &json) +} + pub(crate) fn decode_hex32(hex: &str) -> Option<[u8; 32]> { if hex.len() != 64 || !hex.is_ascii() { return None; @@ -360,9 +508,10 @@ 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, + content: None, translation: JsonValue::Null, } } @@ -390,6 +539,12 @@ mod tests { decided_at: 11_000, }; write_decision(&mut doc, "c-1", &decision).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); @@ -399,15 +554,48 @@ 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); } #[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), decision: None, abort_requested: None, + contested: false, + votes: BTreeMap::new(), }; assert_eq!(record.phase(10_000, 500), Phase::Proposed); assert_eq!(record.phase(10_500, 500), Phase::Proposed); @@ -420,6 +608,45 @@ mod tests { } #[test] + fn disagreeing_decisions_are_contested() { + let mut base = AutoCommit::new(); + write_proposal(&mut base, "c-1", &proposal(10_000)).unwrap(); + base.commit(); + let (mut a, mut b, mut c) = (base.fork(), base.fork(), base.fork()); + let committed = Decision::Committed { + to_epoch: EpochRecord { + generation: 1, + id: "cd".repeat(32), + }, + snapshot_bytes: 1, + decided_at: 2, + }; + write_decision(&mut a, "c-1", &committed).unwrap(); + write_decision( + &mut b, + "c-1", + &Decision::Aborted { + reason: "test".into(), + decided_at: 3, + }, + ) + .unwrap(); + // The same decision, concurrently: they agree. + write_decision(&mut c, "c-1", &committed).unwrap(); + a.commit(); + b.commit(); + c.commit(); + let mut agreeing = a.fork(); + agreeing.merge(&mut c).unwrap(); + assert!(!load(&agreeing, "c-1").unwrap().contested); + + a.merge(&mut b).unwrap(); + let record = load(&a, "c-1").unwrap(); + assert!(record.contested); + assert_eq!(record.phase(0, 0), Phase::Committed); + } + + #[test] fn unreadable_records_are_skipped() { let mut doc = AutoCommit::new(); store::write_path( diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs index bd99854..e50a2a2 100644 --- a/src/remake/snapshot.rs +++ b/src/remake/snapshot.rs @@ -24,17 +24,19 @@ 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}; 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. @@ -51,8 +53,8 @@ pub(crate) struct SnapshotParams<'a> { pub(crate) proposed_at: u64, /// Name of the remake kind, which builds the user content. pub(crate) kind: &'a str, - /// The kind's build input, from the proposal. - pub(crate) content: &'a JsonValue, + /// The kind's build input (the proposal's content). + pub(crate) content: Option<&'a [u8]>, /// The kind's translation input; recorded in the epoch metadata. pub(crate) translation: &'a JsonValue, } @@ -92,7 +94,7 @@ pub(crate) struct SnapshotRequest { pub(crate) policy: JsonValue, pub(crate) proposed_at: u64, pub(crate) kind: String, - pub(crate) content: JsonValue, + pub(crate) content: Option<std::sync::Arc<Vec<u8>>>, pub(crate) translation: JsonValue, } @@ -105,7 +107,7 @@ impl SnapshotRequest { policy: &self.policy, proposed_at: self.proposed_at, kind: &self.kind, - content: &self.content, + content: self.content.as_deref().map(Vec::as_slice), translation: &self.translation, } } @@ -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])?; } @@ -353,7 +358,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)) @@ -367,7 +372,7 @@ mod tests { policy, proposed_at: 0, kind: "compact", - content: &NULL, + content: None, translation: &NULL, } } @@ -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/switch.rs b/src/remake/switch.rs index 2670f14..cf5ea75 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -24,14 +24,17 @@ use automerge::{ActorId, AutoCommit, ChangeHash}; use tokio::sync::broadcast; use tracing::{error, info, warn}; +use super::Host; use super::catchup; +use super::content; +use super::coordinator; 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,8 +81,19 @@ 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; + content::prune( + host, + &all, + unix_millis(), + host.dormant_after().as_millis() as u64, + ); let local_record = EpochRecord::from(local); + if host.is_node_ready().await { + coordinator::resume(host, &local_record, &all); + } let committed: Vec<(String, Record)> = all .into_iter() @@ -91,9 +105,13 @@ async fn check<H: Host>(host: &H) { let (id, record) = match committed.as_slice() { // Nothing committed here; maybe peers moved on without us. [] => return catchup::check(host).await, + [(id, record)] if record.contested => { + error!(remake = %id, "Decisions on this remake disagree; not switching on my own"); + 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"); @@ -154,10 +172,16 @@ async fn switch<H: Host>( id: &str, record: &Record, ) -> Result<(ReplayStats, Option<String>), String> { - let (Some(expected), Some(request)) = (record.committed_epoch(), record.snapshot_request(id)) + let (Some(expected), Some(mut request)) = + (record.committed_epoch(), record.snapshot_request(id)) else { return Err("unreadable remake record".to_owned()); }; + // Voters fetched it already; a node that didn't vote fetches it now. Without it the + // switch fails, and the node rejoins (downloading the snapshot instead). + if let Some(reference) = &record.proposal.content { + request.content = Some(content::ensure(host, reference).await?); + } let translator = request .translator() .ok_or_else(|| format!("unknown remake kind {:?}", request.kind))?; diff --git a/src/remake/vote.rs b/src/remake/vote.rs new file mode 100644 index 0000000..16de87a --- /dev/null +++ b/src/remake/vote.rs @@ -0,0 +1,515 @@ +//! 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. +//! +//! 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. +//! +//! 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; + +use tracing::{info, warn}; + +use super::Host; +use super::content; +use super::records::{self, Decision, EpochRecord, Phase, Record, Register, 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 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, +} + +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, ", + 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 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. +pub(crate) fn tally( + id: &str, + record: &Record, + all: &[(String, Record)], + register: &Register, +) -> Tally { + let electorate: HashSet<&str> = record + .proposal + .electorate + .iter() + .map(String::as_str) + .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, + elsewhere, + twice, + pending: electorate.len() - yes - elsewhere - twice, + } +} + +/// 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 only = match freedom(me, epoch, all, register, started) { + Freedom::Any => None, + Freedom::Only(id) => Some(id), + Freedom::Held(_) => return None, + }; + 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)) + }) + .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; + 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 register = records::load_register(doc); + let id = choice( + &voter, + &wanted, + &all, + ®ister, + started, + unix_millis(), + grace, + )?; + let content = all + .into_iter() + .find(|(other, _)| *other == id) + .and_then(|(_, record)| record.proposal.content); + Some((id, content)) + }) + .await; + let Some((chosen, content)) = due else { + return; + }; + // Only with the content in hand: a committed remake must have it on a majority. + if let Some(reference) = &content { + if content::available_or_fetch(host, reference).is_none() { + 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 register = records::load_register(doc); + let Some(id) = choice(&voter, &epoch, &all, ®ister, started, now, grace) + .filter(|id| *id == chosen) + else { + return Ok(None); + }; + let vote = Vote { + remake: id.clone(), + at: now, + }; + records::write_vote(doc, &voter, &vote)?; + 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::{Policy, Proposal}; + use serde_json::Value as JsonValue; + + fn record(proposed_at: u64, electorate: &[&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: None, + translation: JsonValue::Null, + }, + decision: None, + abort_requested: None, + contested: false, + 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() { + 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)), + ("y".to_owned(), record(200, &ALL)), + ]; + // 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, &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 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 all = vec![ + ("x".to_owned(), aborted(record(100, &ALL))), + ("y".to_owned(), record(200, &ALL)), + ]; + 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", &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)), + ]; + 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 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") + ); + + // 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!(freedom("a", &genesis(), &all, &none, started), Freedom::Any); + } +} |