summaryrefslogtreecommitdiffstats
path: root/src/remake
diff options
context:
space:
mode:
Diffstat (limited to 'src/remake')
-rw-r--r--src/remake/api.rs260
-rw-r--r--src/remake/catchup.rs119
-rw-r--r--src/remake/content.rs95
-rw-r--r--src/remake/coordinator.rs203
-rw-r--r--src/remake/delta.rs7
-rw-r--r--src/remake/kind.rs52
-rw-r--r--src/remake/local.rs99
-rw-r--r--src/remake/mod.rs37
-rw-r--r--src/remake/records.rs257
-rw-r--r--src/remake/snapshot.rs28
-rw-r--r--src/remake/switch.rs23
-rw-r--r--src/remake/vote.rs408
12 files changed, 1310 insertions, 278 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs
index de8166c..826c81e 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -12,9 +12,10 @@
//! * `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>", …],
@@ -22,15 +23,19 @@
//! "archiveRetention": {"epochs": 2, "seconds": 2592000},
//! "settleSecs": 480, "switchDelaySecs": 60 }`
//!
-//! `?force=true` accepts a `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), `replace`, whose spec is
//! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }`
//! (see `kind::Replace`), or `import`, whose spec is
//! `{ "document": "<base64 of an Automerge document>", "rules": […] }` (see
-//! `kind::Import`; `GET /admin/document` exports one). Reports leave the document out
-//! (`contentBytes`). Requests may be up to 64 MiB.
+//! `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
@@ -38,6 +43,7 @@
//! 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;
@@ -61,7 +67,8 @@ use uuid::Uuid;
use super::coordinator;
use super::local::LocalStatus;
use super::records::{
- self, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, Vote, unix_millis,
+ self, AbortRequest, ContentRef, Decision, EpochRecord, Phase, Policy, Proposal, Record,
+ Retention, Vote, unix_millis,
};
use super::{Host, kind, vote};
use crate::document::store::StoreError;
@@ -77,9 +84,8 @@ pub(crate) struct WaitQuery {
pub(crate) struct StartQuery {
#[serde(default)]
wait: bool,
- /// Allows a switch delay shorter than one keepalive interval.
- #[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).
@@ -149,10 +155,57 @@ struct StartRequest {
/// 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 {
@@ -160,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>,
@@ -173,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)]
@@ -195,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,
}
@@ -279,6 +332,10 @@ async fn start_kind<H: Host>(
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 is_octet_stream(headers) {
match query.import_request(&body) {
@@ -369,7 +426,7 @@ async fn start_kind<H: Host>(
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(),
@@ -380,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,
@@ -394,7 +451,7 @@ async fn start_kind<H: Host>(
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(),
@@ -414,9 +471,17 @@ async fn start_kind<H: Host>(
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
@@ -426,6 +491,7 @@ async fn start_kind<H: Host>(
id.clone(),
proposal,
base,
+ force.supersede,
now,
grace,
));
@@ -450,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);
}
@@ -492,19 +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, &register),
+ })
+ .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, &register, started) {
+ vote::Freedom::Any => {}
+ vote::Freedom::Only(id) => {
+ return Err(StartError::VoteHeld(format!(
+ "it may have voted for remake {id} before its start"
+ )));
+ }
+ vote::Freedom::Held(reason) => return Err(StartError::VoteHeld(reason)),
}
// The base is the requested heads, or everything before the proposal (the
@@ -534,12 +682,16 @@ async fn propose<H: Host>(
proposal.electorate = electorate;
records::write_proposal(doc, &record_id, &proposal)?;
- let own_vote = Vote { at: now };
- records::write_vote(doc, &record_id, &me, &own_vote)?;
+ let own_vote = Vote {
+ remake: record_id.clone(),
+ at: now,
+ };
+ records::write_vote(doc, &me, &own_vote)?;
Ok(Record {
proposal,
decision: None,
abort_requested: None,
+ contested: false,
votes: [(me, own_vote)].into(),
})
})
@@ -585,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)
@@ -624,6 +772,7 @@ 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,
@@ -631,7 +780,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
Some(json!({ "remake": current })),
);
}
- Phase::Aborted | Phase::Expired => {
+ Phase::Aborted => {
return error_response(
StatusCode::CONFLICT,
"already over",
@@ -641,7 +790,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
}
// 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(),
@@ -674,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 db25251..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,13 +27,13 @@
//! 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};
@@ -44,12 +48,23 @@ use crate::document::replica::Epoch;
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
+ });
- let Some(generation) = ahead.iter().map(|(_, epoch)| epoch.generation).max() else {
+ // 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) = 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.
@@ -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 5885884..9bcf8ba 100644
--- a/src/remake/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -6,48 +6,85 @@
//! * an operator requested an abort,
//! * the local epoch changed,
//! * a competing remake of the same epoch committed,
-//! * a majority is out of reach (too many members voted for a competing remake), or
-//! * the voting period (`settleSecs`) passed without a majority.
+//! * the voting period (`settleSecs`) passed without a majority, or
+//! * a majority is out of reach (too many members voted for a competing remake).
+//!
+//! 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.
//!
//! Committing builds the snapshot at the base heads and records the new epoch, so
//! every node can verify its own build against it.
//!
-//! The task lives only in memory. If the coordinator dies, nobody decides and the
-//! proposal expires (its votes become void).
+//! 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::time::Duration;
use tracing::{info, warn};
use super::Host;
-use super::records::{self, Decision, Record};
+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,15 +95,26 @@ 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) {
+ 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);
}
- let tally = vote::tally(id, &record, &all, now, grace);
+ let tally = vote::tally(id, &record, &all, &register);
+ // 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);
+ }
if tally.reached() {
info!(remake = %id, votes = %tally, "Majority reached");
commit(host, id, &record).await?;
@@ -76,49 +124,24 @@ async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
decide_abort(host, id, format!("a majority is out of reach: {tally}")).await?;
return Ok(true);
}
- if now >= record.proposal.settle_until {
- decide_abort(
- host,
- id,
- format!("no majority in the voting period: {tally}"),
- )
- .await?;
- return Ok(true);
- }
Ok(false)
}
-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 the vote".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) == records::Phase::Committed
- }) {
- return Some(format!("competing remake {other} committed"));
- }
-
- 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,
@@ -126,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;
@@ -144,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, &register);
+ 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 d318c00..c8f6b7c 100644
--- a/src/remake/delta.rs
+++ b/src/remake/delta.rs
@@ -27,7 +27,7 @@ use automerge::{ActorId, AutoCommit, ChangeHash, PatchAction, Prop};
use tracing::warn;
use super::kind::{Translation, Translator};
-use super::snapshot::{EPOCHS, META_ROOT, REMAKES};
+use super::snapshot::{EPOCHS, META_ROOT, REMAKES, VOTES};
use crate::document::store::{self, StoreError};
#[derive(Debug, Default, Clone)]
@@ -135,7 +135,10 @@ fn log_failure(path: &[String], error: &StoreError) {
}
fn is_protocol_path(path: &[String]) -> bool {
- matches!(path, [root, part, ..] if root == META_ROOT && (part == REMAKES || part == EPOCHS))
+ let [root, part, ..] = path else {
+ return false;
+ };
+ root == META_ROOT && [REMAKES, VOTES, EPOCHS].contains(&part.as_str())
}
/// True for the root and for `_continuum` itself.
diff --git a/src/remake/kind.rs b/src/remake/kind.rs
index 685fa28..8131f7f 100644
--- a/src/remake/kind.rs
+++ b/src/remake/kind.rs
@@ -28,11 +28,12 @@ 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,
@@ -64,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>;
@@ -107,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)
@@ -199,8 +200,10 @@ impl RemakeKind for Replace {
));
}
+ let content =
+ serde_json::to_vec(&canonical(spec.document)).map_err(|error| error.to_string())?;
Ok(Prepared {
- content: canonical(spec.document),
+ content: Some(content),
translation: prepare_rules(spec.rules)?,
})
}
@@ -208,12 +211,15 @@ impl RemakeKind for Replace {
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 {
@@ -300,11 +306,8 @@ struct ImportSpec {
rules: Vec<RuleSpec>,
}
-/// The document in `content` (base64 of a saved Automerge document).
-fn load_content(content: &JsonValue) -> Result<AutoCommit, String> {
- let JsonValue::String(encoded) = content else {
- return Err("the document must be a base64 string".to_owned());
- };
+/// 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}"))?;
@@ -320,17 +323,17 @@ impl RemakeKind for Import {
fn propose(&self, spec: &JsonValue) -> Result<Prepared, String> {
let spec: ImportSpec =
serde_json::from_value(spec.clone()).map_err(|error| error.to_string())?;
- let imported = load_content(&JsonValue::String(spec.document))?;
+ 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 gets
- // replicated with the proposal.
+ // 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: JsonValue::String(BASE64.encode(content.save())),
+ content: Some(content.save()),
translation: prepare_rules(spec.rules)?,
})
}
@@ -338,10 +341,11 @@ impl RemakeKind for Import {
fn build(
&self,
_old_at_base: &AutoCommit,
- content: &JsonValue,
+ content: Option<&[u8]>,
target: &mut AutoCommit,
) -> Result<(), StoreError> {
- let imported = load_content(content).map_err(|reason| anyhow::anyhow!(reason))?;
+ 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)
}
@@ -412,7 +416,7 @@ mod tests {
fn build(
&self,
_old: &AutoCommit,
- _content: &JsonValue,
+ _content: Option<&[u8]>,
_target: &mut AutoCommit,
) -> Result<(), StoreError> {
Ok(())
@@ -559,7 +563,7 @@ mod tests {
let build = || {
let mut target = AutoCommit::new().with_actor(automerge::ActorId::from(&b"fixed"[..]));
Import
- .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));
target
@@ -591,7 +595,7 @@ mod tests {
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 d2ea2c2..04ac0b0 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -1,12 +1,13 @@
//! 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;
@@ -16,12 +17,23 @@ use super::records::{Retention, unix_millis};
use super::snapshot::Snapshot;
use crate::document::replica::Epoch;
-#[derive(Default)]
pub(crate) struct RemakeState {
+ /// When this node started (Unix milliseconds): votes it cast before may be lost
+ /// (`vote`).
+ started_at: u64,
// A std mutex: every access is short and never spans an await.
inner: Mutex<Inner>,
}
+impl Default for RemakeState {
+ fn default() -> Self {
+ Self {
+ started_at: unix_millis(),
+ inner: Mutex::default(),
+ }
+ }
+}
+
#[derive(Default)]
struct Inner {
/// Built by the coordinator at commit time; consumed by the local switch.
@@ -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 0895931..ca17956 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -7,6 +7,7 @@
//! * `records`: the replicated remake records under `_continuum/remakes/<id>`.
//! * `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.
@@ -20,6 +21,7 @@
mod api;
mod catchup;
+mod content;
mod coordinator;
mod delta;
mod kind;
@@ -68,6 +70,21 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
/// 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.
@@ -124,10 +141,30 @@ 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
diff --git a/src/remake/records.rs b/src/remake/records.rs
index ed018ca..cd3e5be 100644
--- a/src/remake/records.rs
+++ b/src/remake/records.rs
@@ -7,7 +7,16 @@
//! * `decision`: written once, by the coordinator only (committed or aborted).
//! * `abortRequested`: written by any node on an operator's request. It is an input to
//! the coordinator's decision, not a decision itself.
-//! * `votes/<pk>`: each written by that member only (`vote`).
+//!
+//! 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,
@@ -24,7 +33,7 @@ use serde_json::Value as JsonValue;
use tracing::warn;
use super::kind::DEFAULT_KIND;
-use super::snapshot::{META_ROOT, REMAKES, SnapshotRequest};
+use super::snapshot::{META_ROOT, REMAKES, SnapshotRequest, VOTES};
use crate::document::replica::Epoch;
use crate::document::store::{self, StoreError};
@@ -137,14 +146,37 @@ pub(crate) struct Proposal {
/// 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()
}
@@ -178,13 +210,18 @@ pub(crate) struct AbortRequest {
pub(crate) at: u64,
}
-/// A member's vote for a remake.
-#[derive(Clone, Debug, Serialize, Deserialize)]
+/// A member's vote: its entry in the votes register.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Vote {
+ pub(crate) remake: String,
pub(crate) at: u64,
}
+/// The votes register: every concurrent value of each member's entry, by member public
+/// key. Two values mean the member voted again without seeing its first vote (`vote`).
+pub(crate) type Register = BTreeMap<String, Vec<Vote>>;
+
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Record {
@@ -193,8 +230,17 @@ pub(crate) struct Record {
pub(crate) decision: Option<Decision>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) abort_requested: Option<AbortRequest>,
- /// By member public key.
- #[serde(default)]
+ /// 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>,
}
@@ -229,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(),
@@ -239,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(),
})
}
@@ -269,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, &register)))
+ })
+ .collect()
+}
+
+fn with_votes(id: &str, mut record: Record, register: &Register) -> Record {
+ record.votes = register
+ .iter()
+ .filter_map(|(voter, votes)| {
+ let vote = votes.iter().find(|vote| vote.remake == id)?;
+ Some((voter.clone(), vote.clone()))
+ })
+ .collect();
+ record
+}
+
+fn load_record(doc: &AutoCommit, id: &str) -> Option<Record> {
let value = match store::read_path(doc, &record_path(id, "")) {
Ok(Some(value)) => value,
Ok(None) => return None,
@@ -279,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, &register_path(&voter)) {
+ Ok(values) => values,
+ Err(error) => {
+ warn!(voter = %voter, error = %error, "Failed to read vote");
+ return None;
+ }
+ };
+ let votes: Vec<Vote> = values
+ .into_iter()
+ .filter_map(|value| match serde_json::from_value(value) {
+ Ok(vote) => Some(vote),
+ Err(error) => {
+ warn!(voter = %voter, error = %error, "Skipping unreadable vote");
+ None
+ }
+ })
+ .collect();
+ (!votes.is_empty()).then_some((voter, votes))
+ })
.collect()
}
@@ -331,16 +458,10 @@ pub(crate) fn write_abort_request(
write_part(doc, id, "abortRequested", request)
}
-pub(crate) fn write_vote(
- doc: &mut AutoCommit,
- id: &str,
- voter: &str,
- vote: &Vote,
-) -> Result<(), StoreError> {
+/// Writes `voter`'s entry in the votes register (replacing its earlier one).
+pub(crate) fn write_vote(doc: &mut AutoCommit, voter: &str, vote: &Vote) -> Result<(), StoreError> {
let json = serde_json::to_value(vote).map_err(anyhow::Error::from)?;
- let mut path = record_path(id, "votes");
- path.push(voter.to_owned());
- store::write_path(doc, &path, &json)
+ store::write_path(doc, &register_path(voter), &json)
}
pub(crate) fn decode_hex32(hex: &str) -> Option<[u8; 32]> {
@@ -390,7 +511,7 @@ mod tests {
electorate: vec!["node-a".into(), "node-b".into()],
quorum: 2,
kind: DEFAULT_KIND.to_owned(),
- content: JsonValue::Null,
+ content: None,
translation: JsonValue::Null,
}
}
@@ -418,7 +539,12 @@ mod tests {
decided_at: 11_000,
};
write_decision(&mut doc, "c-1", &decision).unwrap();
- write_vote(&mut doc, "c-1", "node-b", &Vote { at: 7 }).unwrap();
+ let vote = |remake: &str| Vote {
+ remake: remake.into(),
+ at: 7,
+ };
+ write_vote(&mut doc, "node-b", &vote("c-1")).unwrap();
+ write_vote(&mut doc, "node-c", &vote("other")).unwrap();
doc.commit();
let all = load_all(&doc);
@@ -434,11 +560,41 @@ mod tests {
}
#[test]
+ fn concurrent_votes_are_kept_apart() {
+ let mut a = AutoCommit::new();
+ let vote = |remake: &str, at| Vote {
+ remake: remake.into(),
+ at,
+ };
+ write_vote(&mut a, "node-a", &vote("x", 1)).unwrap();
+ a.commit();
+ let mut b = a.fork();
+ // node-b votes in both, never seeing its other vote (a lost vote).
+ write_vote(&mut a, "node-b", &vote("x", 2)).unwrap();
+ write_vote(&mut b, "node-b", &vote("y", 3)).unwrap();
+ a.commit();
+ b.commit();
+ a.merge(&mut b).unwrap();
+
+ let register = load_register(&a);
+ assert_eq!(register["node-a"], [vote("x", 1)]);
+ let mut twice = register["node-b"].clone();
+ twice.sort_by_key(|vote| vote.at);
+ assert_eq!(twice, [vote("x", 2), vote("y", 3)]);
+
+ // Seen, a vote is replaced.
+ write_vote(&mut a, "node-b", &vote("z", 4)).unwrap();
+ a.commit();
+ assert_eq!(load_register(&a)["node-b"], [vote("z", 4)]);
+ }
+
+ #[test]
fn phases() {
let mut record = Record {
proposal: proposal(10_000),
decision: None,
abort_requested: None,
+ contested: false,
votes: BTreeMap::new(),
};
assert_eq!(record.phase(10_000, 500), Phase::Proposed);
@@ -452,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 efe7e37..e50a2a2 100644
--- a/src/remake/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -35,6 +35,8 @@ use crate::document::store::{self, StoreError};
pub(crate) const META_ROOT: &str = "_continuum";
pub(crate) const REMAKES: &str = "remakes";
+/// The votes register (`records`).
+pub(crate) const VOTES: &str = "votes";
pub(crate) const EPOCHS: &str = "epochs";
/// Everything that determines a snapshot besides the base content.
@@ -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])?;
}
@@ -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 e9f49ee..cf5ea75 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -26,6 +26,8 @@ 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;
@@ -82,7 +84,16 @@ async fn check<H: Host>(host: &H) {
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()
@@ -94,6 +105,10 @@ 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 => {
// Majority voting prevents this; never guess which one is right. Peers
@@ -157,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
index d73326d..16de87a 100644
--- a/src/remake/vote.rs
+++ b/src/remake/vote.rs
@@ -3,25 +3,32 @@
//!
//! The electorate is fixed in the proposal: the members that renewed their alive stamp
//! within the coordinator's `--electorate-window-days` (`swarm::members::electorate`),
-//! the coordinator always included. Each member votes for at most one open remake of
-//! its epoch: the first it sees (the oldest proposal, if it sees several at once). Two
-//! competing remakes, even in different partitions, can't both reach a majority, so at
-//! most one commits. A vote for a remake that aborted or expired is void; the member
-//! may vote again.
+//! the coordinator always included.
//!
-//! Votes live in the record (`votes/<pk>`, written only by that member). A node votes
-//! once it is ready, on its current epoch, and only if it is in the electorate.
+//! Votes live in the votes register (`_continuum/votes/<pk>`, written only by that
+//! member): one entry per member for the whole epoch (the snapshot leaves it behind).
+//! A member votes for at most one remake of its epoch, the first it sees (the oldest
+//! proposal, if it sees several at once), and votes again only once that remake is
+//! aborted. Neither expiry nor a deleted record frees a vote. Two competing remakes,
+//! even in different partitions, can't both reach a majority, so at most one commits.
//!
-//! Gap: votes are not persisted locally. A node whose vote reached only the coordinator
-//! and that then restarts and resyncs from nodes that never saw the vote may vote again
-//! for a competitor.
+//! A node votes once it is ready, on its current epoch, only if it is in the electorate,
+//! and only once it holds the remake's content (`content`), so a committed remake has
+//! its content on a majority.
+//!
+//! A vote can be lost: a node whose vote reached only the coordinator restarts and
+//! resyncs from nodes that never saw it. Should it vote again for a competitor, its
+//! entry ends up with two concurrent values, and it counts for no remake. To avoid
+//! that, a node doesn't vote while more than one undecided remake it could have voted
+//! for is older than its start, and with one, votes only for that one.
use std::collections::HashSet;
use tracing::{info, warn};
use super::Host;
-use super::records::{self, EpochRecord, Phase, Record, Vote, unix_millis};
+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.
@@ -36,8 +43,10 @@ pub(crate) struct Tally {
pub(crate) quorum: usize,
/// Voted for it.
pub(crate) yes: usize,
- /// Voted for a competing open remake.
+ /// Voted for a competing remake that isn't aborted.
pub(crate) elsewhere: usize,
+ /// Voted for two remakes: counts for neither.
+ pub(crate) twice: usize,
/// Not voted (yet).
pub(crate) pending: usize,
}
@@ -57,14 +66,51 @@ impl std::fmt::Display for Tally {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "{} of {} voted for it, {} for a competing remake, {} not yet; {} needed",
- self.yes, self.electorate, self.elsewhere, self.pending, self.quorum
- )
+ "{} of {} voted for it, {} for a competing remake, ",
+ self.yes, self.electorate, self.elsewhere
+ )?;
+ if self.twice > 0 {
+ write!(f, "{} for two, ", self.twice)?;
+ }
+ write!(f, "{} not yet; {} needed", self.pending, self.quorum)
+ }
+}
+
+/// What a member's entry in the register amounts to.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum Held<'a> {
+ /// No vote, or only for aborted remakes.
+ Free,
+ For(&'a str),
+ /// For two remakes, neither aborted.
+ Twice,
+}
+
+/// What `votes` (one member's entry) amount to among `all` records.
+pub(crate) fn held<'a>(votes: &'a [Vote], all: &[(String, Record)]) -> Held<'a> {
+ let aborted = |remake: &str| {
+ all.iter().any(|(id, record)| {
+ id == remake && matches!(record.decision, Some(Decision::Aborted { .. }))
+ })
+ };
+ let mut kept = votes
+ .iter()
+ .map(|vote| vote.remake.as_str())
+ .filter(|remake| !aborted(remake));
+ let Some(first) = kept.next() else {
+ return Held::Free;
+ };
+ if kept.all(|other| other == first) {
+ Held::For(first)
+ } else {
+ Held::Twice
}
}
-fn open(record: &Record, now: u64, grace: u64) -> bool {
- record.phase(now, grace).is_open()
+fn held_by<'a>(member: &str, register: &'a Register, all: &[(String, Record)]) -> Held<'a> {
+ register
+ .get(member)
+ .map_or(Held::Free, |votes| held(votes, all))
}
/// The tally of remake `id` among `all` records.
@@ -72,8 +118,7 @@ pub(crate) fn tally(
id: &str,
record: &Record,
all: &[(String, Record)],
- now: u64,
- grace: u64,
+ register: &Register,
) -> Tally {
let electorate: HashSet<&str> = record
.proposal
@@ -81,51 +126,95 @@ pub(crate) fn tally(
.iter()
.map(String::as_str)
.collect();
- let yes: HashSet<&str> = record
- .votes
- .keys()
- .map(String::as_str)
- .filter(|voter| electorate.contains(voter))
- .collect();
- let elsewhere: HashSet<&str> = all
- .iter()
- .filter(|(other, candidate)| {
- other != id
- && candidate.proposal.from_epoch == record.proposal.from_epoch
- && open(candidate, now, grace)
- })
- .flat_map(|(_, candidate)| candidate.votes.keys().map(String::as_str))
- .filter(|voter| electorate.contains(voter) && !yes.contains(voter))
- .collect();
+ let (mut yes, mut elsewhere, mut twice) = (0, 0, 0);
+ for member in &electorate {
+ match held_by(member, register, all) {
+ Held::For(remake) if remake == id => yes += 1,
+ Held::For(_) => elsewhere += 1,
+ Held::Twice => twice += 1,
+ Held::Free => {}
+ }
+ }
Tally {
electorate: electorate.len(),
quorum: record.proposal.quorum,
- yes: yes.len(),
- elsewhere: elsewhere.len(),
- pending: electorate.len() - yes.len() - elsewhere.len(),
+ yes,
+ elsewhere,
+ twice,
+ pending: electorate.len() - yes - elsewhere - twice,
}
}
-/// The remake `me` should vote for now, if any: none if it already voted for an open
-/// remake of `epoch`; otherwise the oldest proposed one whose electorate includes it.
+/// What `me` may vote for in `epoch`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum Freedom {
+ Any,
+ /// Only this remake: the one it may have voted for before its start.
+ Only(String),
+ /// Nothing, for this reason.
+ Held(String),
+}
+
+/// What `me` may vote for in `epoch`. Remakes proposed before `started` (this node's
+/// start, plus a margin for clocks) may hold a vote of ours that we lost in a restart.
+pub(crate) fn freedom(
+ me: &str,
+ epoch: &EpochRecord,
+ all: &[(String, Record)],
+ register: &Register,
+ started: u64,
+) -> Freedom {
+ match held_by(me, register, all) {
+ Held::Free => {}
+ Held::For(remake) => {
+ return Freedom::Held(format!("this node voted for remake {remake}"));
+ }
+ Held::Twice => {
+ return Freedom::Held("this node voted for two remakes".to_owned());
+ }
+ }
+ let mut older = all
+ .iter()
+ .filter(|(_, record)| {
+ &record.proposal.from_epoch == epoch
+ && record.decision.is_none()
+ && record.proposal.proposed_at <= started
+ && record.proposal.electorate.iter().any(|member| member == me)
+ })
+ .map(|(id, _)| id);
+ match (older.next(), older.next()) {
+ (None, _) => Freedom::Any,
+ (Some(id), None) => Freedom::Only(id.clone()),
+ (Some(_), Some(_)) => Freedom::Held(
+ "several undecided remakes are older than this node's start; it may have voted \
+ for one before"
+ .to_owned(),
+ ),
+ }
+}
+
+/// The remake `me` should vote for now, if any: the oldest proposed one of `epoch`
+/// whose electorate includes it, as far as it is free to vote (`freedom`).
pub(crate) fn choice(
me: &str,
epoch: &EpochRecord,
all: &[(String, Record)],
+ register: &Register,
+ started: u64,
now: u64,
grace: u64,
) -> Option<String> {
- let of_epoch = || {
- all.iter()
- .filter(move |(_, record)| &record.proposal.from_epoch == epoch)
+ let only = match freedom(me, epoch, all, register, started) {
+ Freedom::Any => None,
+ Freedom::Only(id) => Some(id),
+ Freedom::Held(_) => return None,
};
- if of_epoch().any(|(_, record)| open(record, now, grace) && record.votes.contains_key(me)) {
- return None;
- }
- of_epoch()
- .filter(|(_, record)| {
- record.phase(now, grace) == Phase::Proposed
+ all.iter()
+ .filter(|(id, record)| {
+ &record.proposal.from_epoch == epoch
+ && record.phase(now, grace) == Phase::Proposed
&& record.proposal.electorate.iter().any(|member| member == me)
+ && only.as_ref().is_none_or(|only| only == id)
})
.min_by(|(a_id, a), (b_id, b)| {
(a.proposal.proposed_at, a_id).cmp(&(b.proposal.proposed_at, b_id))
@@ -142,23 +231,38 @@ pub(super) async fn cast<H: Host>(host: &H) {
let me = host.node_id();
let epoch = EpochRecord::from(host.current_epoch());
let grace = host.dormant_after().as_millis() as u64;
+ let started = host.remake().started_at() + grace;
// Usually there is nothing to vote on: look first, without writing.
let (voter, wanted) = (me.clone(), epoch.clone());
let due = host
.read_doc(move |doc| {
- choice(
+ let all = records::load_all(doc);
+ let register = records::load_register(doc);
+ let id = choice(
&voter,
&wanted,
- &records::load_all(doc),
+ &all,
+ &register,
+ started,
unix_millis(),
grace,
- )
- .is_some()
+ )?;
+ let content = all
+ .into_iter()
+ .find(|(other, _)| *other == id)
+ .and_then(|(_, record)| record.proposal.content);
+ Some((id, content))
})
.await;
- if !due {
+ 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.
@@ -167,10 +271,17 @@ pub(super) async fn cast<H: Host>(host: &H) {
.commit_doc(move |doc| -> Result<Option<String>, StoreError> {
let now = unix_millis();
let all = records::load_all(doc);
- let Some(id) = choice(&voter, &epoch, &all, now, grace) else {
+ let register = records::load_register(doc);
+ let Some(id) = choice(&voter, &epoch, &all, &register, started, now, grace)
+ .filter(|id| *id == chosen)
+ else {
return Ok(None);
};
- records::write_vote(doc, &id, &voter, &Vote { at: now })?;
+ let vote = Vote {
+ remake: id.clone(),
+ at: now,
+ };
+ records::write_vote(doc, &voter, &vote)?;
Ok(Some(id))
})
.await;
@@ -185,11 +296,10 @@ pub(super) async fn cast<H: Host>(host: &H) {
mod tests {
use super::*;
use crate::document::replica::Epoch;
- use crate::remake::records::{Decision, Policy, Proposal};
+ use crate::remake::records::{Policy, Proposal};
use serde_json::Value as JsonValue;
- use std::collections::BTreeMap;
- fn record(proposed_at: u64, electorate: &[&str], votes: &[&str]) -> Record {
+ fn record(proposed_at: u64, electorate: &[&str]) -> Record {
Record {
proposal: Proposal {
coordinator: electorate[0].to_owned(),
@@ -206,19 +316,48 @@ mod tests {
electorate: electorate.iter().map(|m| (*m).to_owned()).collect(),
quorum: quorum(electorate.len()),
kind: "compact".to_owned(),
- content: JsonValue::Null,
+ content: None,
translation: JsonValue::Null,
},
decision: None,
abort_requested: None,
- votes: votes
- .iter()
- .map(|v| ((*v).to_owned(), Vote { at: proposed_at }))
- .collect::<BTreeMap<_, _>>(),
+ 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() {
@@ -232,60 +371,145 @@ mod tests {
#[test]
fn competing_remakes_split_the_vote() {
let all = vec![
- ("x".to_owned(), record(100, &ALL, &["a", "b"])),
- ("y".to_owned(), record(200, &ALL, &["c", "z"])), // z isn't in the electorate
+ ("x".to_owned(), record(100, &ALL)),
+ ("y".to_owned(), record(200, &ALL)),
];
- let x = tally("x", &all[0].1, &all, 150, 0);
+ // z isn't in the electorate.
+ let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["y"]), ("z", &["y"])]);
+ let x = tally("x", &all[0].1, &all, &votes);
assert_eq!((x.yes, x.elsewhere, x.pending, x.quorum), (2, 1, 2, 3));
assert!(!x.reached() && !x.out_of_reach());
- let y = tally("y", &all[1].1, &all, 150, 0);
+ let y = tally("y", &all[1].1, &all, &votes);
assert_eq!((y.yes, y.elsewhere, y.pending), (1, 2, 2));
assert!(!y.out_of_reach());
// d votes for x: x has its majority, y can't get one any more.
- let all = vec![
- ("x".to_owned(), record(100, &ALL, &["a", "b", "d"])),
- ("y".to_owned(), record(200, &ALL, &["c"])),
- ];
- assert!(tally("x", &all[0].1, &all, 150, 0).reached());
- assert!(tally("y", &all[1].1, &all, 150, 0).out_of_reach());
+ let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["y"]), ("d", &["x"])]);
+ assert!(tally("x", &all[0].1, &all, &votes).reached());
+ assert!(tally("y", &all[1].1, &all, &votes).out_of_reach());
}
#[test]
fn votes_for_aborted_remakes_are_void() {
- let mut aborted = record(100, &ALL, &["a", "b", "c"]);
- aborted.decision = Some(Decision::Aborted {
- reason: "test".into(),
- decided_at: 150,
- });
let all = vec![
- ("x".to_owned(), aborted),
- ("y".to_owned(), record(200, &ALL, &["d"])),
+ ("x".to_owned(), aborted(record(100, &ALL))),
+ ("y".to_owned(), record(200, &ALL)),
];
- let y = tally("y", &all[1].1, &all, 250, 0);
+ let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["x"]), ("d", &["y"])]);
+ let y = tally("y", &all[1].1, &all, &votes);
assert_eq!((y.yes, y.elsewhere, y.pending), (1, 0, 4));
// c may vote again.
assert_eq!(
- choice("c", &Epoch::GENESIS.into(), &all, 250, 0).as_deref(),
+ choice("c", &genesis(), &all, &votes, STARTED, 250, 0).as_deref(),
Some("y")
);
}
#[test]
+ fn expiry_frees_no_vote() {
+ // x's coordinator is gone: x expired, but a, b and c still hold their votes.
+ let all = vec![
+ ("x".to_owned(), record(100, &ALL)),
+ ("y".to_owned(), record(5_000, &ALL)),
+ ];
+ let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["x"])]);
+ assert_eq!(all[0].1.phase(5_000, 0), Phase::Expired);
+ assert_eq!(
+ choice("a", &genesis(), &all, &votes, STARTED, 5_000, 0),
+ None
+ );
+ let y = tally("y", &all[1].1, &all, &votes);
+ assert_eq!((y.yes, y.elsewhere, y.pending), (0, 3, 2));
+ assert!(y.out_of_reach());
+ }
+
+ #[test]
+ fn voting_twice_counts_for_neither() {
+ let all = vec![
+ ("x".to_owned(), record(100, &ALL)),
+ ("y".to_owned(), record(200, &ALL)),
+ ];
+ let votes = register(&[("a", &["x", "y"]), ("b", &["x"]), ("c", &["x", "x"])]);
+ let x = tally("x", &all[0].1, &all, &votes);
+ assert_eq!((x.yes, x.elsewhere, x.twice, x.pending), (2, 0, 1, 2));
+ assert_eq!(
+ x.to_string(),
+ "2 of 5 voted for it, 0 for a competing remake, 1 for two, 2 not yet; 3 needed"
+ );
+ assert!(matches!(
+ freedom("a", &genesis(), &all, &votes, STARTED),
+ Freedom::Held(_)
+ ));
+
+ // Once y is aborted, a's vote for x stands.
+ let all = vec![
+ ("x".to_owned(), record(100, &ALL)),
+ ("y".to_owned(), aborted(record(200, &ALL))),
+ ];
+ assert_eq!(tally("x", &all[0].1, &all, &votes).yes, 3);
+ }
+
+ #[test]
fn one_vote_per_epoch_for_the_oldest() {
let all = vec![
- ("y".to_owned(), record(200, &ALL, &[])),
- ("x".to_owned(), record(100, &ALL, &[])),
+ ("y".to_owned(), record(200, &ALL)),
+ ("x".to_owned(), record(100, &ALL)),
];
- let epoch: EpochRecord = Epoch::GENESIS.into();
- assert_eq!(choice("a", &epoch, &all, 150, 0).as_deref(), Some("x"));
- assert_eq!(choice("outsider", &epoch, &all, 150, 0), None);
+ let none = register(&[]);
+ assert_eq!(
+ 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);
+ }
- let voted = vec![
- ("y".to_owned(), record(200, &ALL, &["a"])),
- ("x".to_owned(), record(100, &ALL, &[])),
+ #[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!(choice("a", &epoch, &voted, 150, 0), None);
+ assert_eq!(freedom("a", &genesis(), &all, &none, started), Freedom::Any);
}
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-25 00:30:07 +0000