diff options
Diffstat (limited to 'src/remake')
| -rw-r--r-- | src/remake/api.rs | 293 | ||||
| -rw-r--r-- | src/remake/catchup.rs | 34 | ||||
| -rw-r--r-- | src/remake/coordinator.rs | 32 | ||||
| -rw-r--r-- | src/remake/delta.rs | 17 | ||||
| -rw-r--r-- | src/remake/kind.rs | 60 | ||||
| -rw-r--r-- | src/remake/local.rs | 25 | ||||
| -rw-r--r-- | src/remake/mod.rs | 1 | ||||
| -rw-r--r-- | src/remake/records.rs | 95 | ||||
| -rw-r--r-- | src/remake/snapshot.rs | 58 | ||||
| -rw-r--r-- | src/remake/switch.rs | 7 | ||||
| -rw-r--r-- | src/remake/vote.rs | 2 |
11 files changed, 488 insertions, 136 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs index 826c81e..0428a43 100644 --- a/src/remake/api.rs +++ b/src/remake/api.rs @@ -17,7 +17,8 @@ //! 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): +//! Request body for POST (all optional; `baseHeads` not for `compact`, whose base is +//! fixed when it commits): //! `{ "kind": "compact", "spec": null, "baseHeads": ["<hex>", …], //! "replayWindow": {"epochs": 1, "seconds": 604800}, //! "archiveRetention": {"epochs": 2, "seconds": 2592000}, @@ -72,6 +73,7 @@ use super::records::{ }; use super::{Host, kind, vote}; use crate::document::store::StoreError; +use crate::force::Force; #[derive(Debug, Default, Deserialize)] pub(crate) struct WaitQuery { @@ -79,12 +81,20 @@ pub(crate) struct WaitQuery { wait: bool, } +#[derive(Debug, Deserialize)] +pub(crate) struct AbortQuery { + #[serde(default)] + wait: bool, + /// The checks to override (`force`): `coordinatorActive`. + force: Option<String>, +} + #[derive(Debug, Default, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct StartQuery { #[serde(default)] wait: bool, - /// The checks to override (`Force`). + /// 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: @@ -159,41 +169,63 @@ struct StartRequest { 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, +/// What `?force=` may override on a proposal (`force`). +const FORCE: &[&str] = &["switchDelaySecs", "supersede", "coordinatorActive"]; +/// What `?force=` may override on an abort. +const ABORT_FORCE: &[&str] = &["coordinatorActive"]; + +/// When a node other than the coordinator may abort an expired remake: a whole grace +/// period after it expired. Its coordinator stops committing at expiry +/// (`coordinator::decide`), so an abort from elsewhere and a commit can only cross if +/// their clocks are more than a grace period apart. +fn abortable_at(record: &Record, grace: u64) -> u64 { + record + .proposal + .settle_until + .saturating_add(grace.saturating_mul(2)) } -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) +/// Why this node may not abort expired `record` now, and the `force` name that would +/// override it (none for one that only time cures). +fn abort_obstacle( + record: &Record, + me: &str, + active_peers: &HashSet<String>, + force: &Force, + now: u64, + grace: u64, +) -> Option<(String, Option<&'static str>)> { + let at = abortable_at(record, grace); + if now < at { + return Some(( + format!( + "only its coordinator may abort it until a grace period after it expired ({}s from now)", + (at - now).div_ceil(1000) + ), + None, + )); + } + let coordinator = &record.proposal.coordinator; + if coordinator != me && active_peers.contains(coordinator) && !force.allows("coordinatorActive") + { + return Some(( + format!( + "its coordinator {coordinator} is an active peer, so it is stuck rather than gone (look at its log); add force=coordinatorActive to abort it anyway" + ), + Some("coordinatorActive"), + )); } + None +} + +/// The active peers this node sees. +async fn active_peers<H: Host>(host: &H) -> HashSet<String> { + host.peer_summaries() + .await + .into_iter() + .filter(|peer| peer.active) + .map(|peer| peer.id) + .collect() } /// An expired remake of the epoch a proposal would supersede. @@ -201,11 +233,20 @@ impl Force { struct ExpiredRemake { id: String, coordinator: String, + coordinator_active: bool, /// When it expired (Unix milliseconds). since: u64, + /// When another node may abort it (`abortable_at`). + abortable_at: u64, tally: vote::Tally, } +/// Superseding expired remakes, as the proposal's `force` asks. +struct Supersede { + force: Force, + active_peers: HashSet<String>, +} + #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct Report { @@ -222,9 +263,15 @@ 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>), + /// Expired remakes still hold votes, and aren't superseded: not asked to + /// (`force=supersede`), or not possible yet (`abort_obstacle`). + #[error("{message}")] + Expired { + message: String, + remakes: Vec<ExpiredRemake>, + /// The `force` name that would override it, if any. + force: Option<&'static str>, + }, /// 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), @@ -332,7 +379,7 @@ 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()) { + let force = match Force::parse(query.force.as_deref(), FORCE) { Ok(force) => force, Err(reason) => return error_response(StatusCode::BAD_REQUEST, reason, None), }; @@ -395,6 +442,15 @@ async fn start_kind<H: Host>( } }; + if kind.base_at_commit() && request.base_heads.is_some() { + return error_response( + StatusCode::BAD_REQUEST, + format!( + "a {kind_name} remake fixes its base when it commits; baseHeads is not accepted" + ), + None, + ); + } let base = match &request.base_heads { None => None, Some(heads) => { @@ -426,7 +482,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 && force.switch_delay => { + Some(delay) if delay < keepalive && force.allows("switchDelaySecs") => { warn!( switch_delay_secs = delay.as_secs(), keepalive_secs = keepalive.as_secs(), @@ -454,7 +510,8 @@ async fn start_kind<H: Host>( let mut proposal = Proposal { coordinator: host.node_id(), from_epoch: host.current_epoch().into(), - base_heads: Vec::new(), + // Filled in when the proposal is written, unless the kind fixes it at the commit. + base_heads: None, policy: Policy { replay_window: request .replay_window @@ -491,7 +548,10 @@ async fn start_kind<H: Host>( id.clone(), proposal, base, - force.supersede, + Supersede { + force: force.clone(), + active_peers: active_peers(&host).await, + }, now, grace, )); @@ -516,33 +576,28 @@ 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 + Err(StartError::Expired { + message, + remakes, + force, + }) => { + let expired: Vec<JsonValue> = remakes .iter() .map(|remake| { json!({ "id": remake.id, "coordinator": remake.coordinator, - "coordinatorActive": remake.coordinator == me - || active.contains(&remake.coordinator), + "coordinatorActive": remake.coordinator_active, "expiredAt": remake.since, + "abortableAt": remake.abortable_at, "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" })), + message, + Some(json!({ "expired": expired, "force": force })), ); } Err(error @ StartError::VoteHeld(_)) => { @@ -590,7 +645,7 @@ async fn propose<H: Host>( id: String, mut proposal: Proposal, base: Option<Vec<ChangeHash>>, - supersede: bool, + supersede: Supersede, now: u64, grace: u64, ) -> Result<Record, StartError> { @@ -598,6 +653,7 @@ async fn propose<H: Host>( let electorate_host = host.clone(); let me = host.node_id(); let started = host.remake().started_at() + grace; + let base_at_commit = kind::by_name(&proposal.kind).is_some_and(|kind| kind.base_at_commit()); let record = host .commit_doc(move |doc| -> Result<Record, StartError> { let mut all = records::load_all(doc); @@ -620,18 +676,50 @@ async fn propose<H: Host>( .map(|(id, _)| id) .collect(); if !expired.is_empty() { - if !supersede { - let details = all - .iter() + let remakes = || { + all.iter() .filter(|(id, _)| expired.contains(&id)) .map(|(id, record)| ExpiredRemake { id: id.clone(), coordinator: record.proposal.coordinator.clone(), + coordinator_active: supersede + .active_peers + .contains(&record.proposal.coordinator), since: record.proposal.settle_until.saturating_add(grace), + abortable_at: abortable_at(record, grace), tally: vote::tally(id, record, &all, ®ister), }) - .collect(); - return Err(StartError::Expired(details)); + .collect() + }; + if !supersede.force.allows("supersede") { + return Err(StartError::Expired { + message: "an expired remake of this epoch still holds votes: abort it, \ + or add ?force=supersede to abort it and propose in its place" + .to_owned(), + remakes: remakes(), + force: Some("supersede"), + }); + } + let obstacle = + all.iter() + .filter(|(id, _)| expired.contains(&id)) + .find_map(|(id, record)| { + let (why, force) = abort_obstacle( + record, + &me, + &supersede.active_peers, + &supersede.force, + now, + grace, + )?; + Some((format!("can't supersede remake {id}: {why}"), force)) + }); + if let Some((message, force)) = obstacle { + return Err(StartError::Expired { + message, + remakes: remakes(), + force, + }); } let decision = Decision::Aborted { reason: format!("superseded by remake {record_id}"), @@ -656,22 +744,22 @@ async fn propose<H: Host>( } // The base is the requested heads, or everything before the proposal (the - // proposal itself is never in it). - let base = match base { - Some(base) => { - if let Some(unknown) = base - .iter() - .find(|head| doc.get_change_by_hash(head).is_none()) - { - return Err(StartError::UnknownBase(unknown.to_string())); + // proposal itself is never in it); or, for a compaction, left to the commit. + if !base_at_commit { + let base = match base { + Some(base) => { + if let Some(unknown) = base + .iter() + .find(|head| doc.get_change_by_hash(head).is_none()) + { + return Err(StartError::UnknownBase(unknown.to_string())); + } + base } - base - } - None => doc.get_heads(), - }; - let mut heads: Vec<String> = base.iter().map(|h| h.to_string()).collect(); - heads.sort(); - proposal.base_heads = heads; + None => doc.get_heads(), + }; + proposal.base_heads = Some(records::encode_heads(&base)); + } // We vote for our own remake, so we are always in the electorate. let mut electorate = electorate_host.electorate(doc, now); @@ -731,16 +819,16 @@ pub(crate) async fn get<H: Host>( pub(crate) async fn abort<H: Host>( State(host): State<H>, Path(id): Path<String>, - Query(query): Query<WaitQuery>, + Query(query): Query<AbortQuery>, ) -> Response { - abort_by_id(host, id, query.wait).await + abort_by_id(host, id, query).await } /// `DELETE /admin/remakes`: aborts whichever remake of the local epoch is still /// 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>, + Query(query): Query<AbortQuery>, ) -> Response { let local = EpochRecord::from(host.current_epoch()); let open = host @@ -756,7 +844,7 @@ pub(crate) async fn abort_current<H: Host>( .await; match open { - Some(id) => abort_by_id(host, id, query.wait).await, + Some(id) => abort_by_id(host, id, query).await, None => error_response( StatusCode::NOT_FOUND, "no remake of the local epoch is in progress", @@ -765,14 +853,19 @@ pub(crate) async fn abort_current<H: Host>( } } -async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { +async fn abort_by_id<H: Host>(host: H, id: String, query: AbortQuery) -> Response { + let force = match Force::parse(query.force.as_deref(), ABORT_FORCE) { + Ok(force) => force, + Err(reason) => return error_response(StatusCode::BAD_REQUEST, reason, None), + }; + let wait = query.wait; let Some(current) = load_report(&host, &id).await else { return error_response(StatusCode::NOT_FOUND, "no such remake", None); }; match current.phase { Phase::Proposed => {} - Phase::Expired => return abort_expired(&host, &id).await, + Phase::Expired => return abort_expired(&host, current, &force).await, Phase::Committed => { return error_response( StatusCode::CONFLICT, @@ -823,9 +916,30 @@ 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 { +/// Aborts expired remake `current` from this node: its coordinator is gone, and can't +/// commit it any more (`coordinator`), unless `abort_obstacle` says otherwise. Answers +/// with the final report. +async fn abort_expired<H: Host>(host: &H, current: Report, force: &Force) -> Response { + let grace = grace(host); + if let Some((why, force)) = abort_obstacle( + ¤t.record, + &host.node_id(), + &active_peers(host).await, + force, + unix_millis(), + grace, + ) { + return error_response( + StatusCode::CONFLICT, + format!("can't abort it yet: {why}"), + Some(json!({ + "remake": current, + "abortableAt": abortable_at(¤t.record, grace), + "force": force, + })), + ); + } + let id = current.id.as_str(); let record_id = id.to_owned(); let decision = Decision::Aborted { reason: format!( @@ -834,11 +948,12 @@ async fn abort_expired<H: Host>(host: &H, id: &str) -> Response { ), 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); + let now = unix_millis(); + let expired = records::load(doc, &record_id).is_some_and(|record| { + record.phase(now, grace) == Phase::Expired && now >= abortable_at(&record, grace) + }); if expired { records::write_decision(doc, &record_id, &decision)?; } diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index cad5a22..631fb25 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -31,7 +31,7 @@ use std::collections::{BTreeMap, HashMap}; use std::convert::Infallible; use std::time::Duration; -use automerge::{AutoCommit, ChangeHash}; +use automerge::{ActorId, AutoCommit, ChangeHash}; use tokio::time::Instant; use tracing::{debug, error, info, warn}; @@ -177,8 +177,12 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) { host.set_stale_view(Some((local, copy))); } let was_ready = host.suspend().await; + let actor = ActorId::random(); + host.remake().add_own_actor(actor.clone()); let old = match host - .switch_epoch(target, |_live| Ok::<_, Infallible>(AutoCommit::new())) + .switch_epoch(target, move |_live| { + Ok::<_, Infallible>(AutoCommit::new().with_actor(actor)) + }) .await { Ok(old) => old, @@ -276,7 +280,9 @@ async fn finish<H: Host>(host: &H) { // Everything heavy runs on a copy, without a lock. let copy = host.read_doc(|doc| doc.clone()).await; - let prepared = tokio::task::spawn_blocking(move || prepare(copy, old, previous, target)).await; + let own = host.remake().own_actors(); + let prepared = + tokio::task::spawn_blocking(move || prepare(copy, old, previous, target, own)).await; let (old, metadata, plan, own_missing, had_history) = match prepared { Ok(Prepared::Ready { @@ -376,6 +382,8 @@ async fn finish<H: Host>(host: &H) { }, ); host.remake().set_catch_up(None); + let actor = host.read_doc(|doc| doc.get_actor().clone()).await; + host.remake().reset_own_actors(actor); if was_ready { host.resume().await; } @@ -395,7 +403,15 @@ async fn finish<H: Host>(host: &H) { } /// Verifies the synced document and decides about the old document's own changes. -fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: Epoch) -> Prepared { +/// `own`: this node's actors (`RemakeState::own_actors`); the old document's own actor +/// counts in any case. +fn prepare( + mut copy: AutoCommit, + mut old: AutoCommit, + previous: Epoch, + target: Epoch, + mut own: Vec<ActorId>, +) -> Prepared { if copy.get_change_by_hash(&ChangeHash(target.id)).is_none() { return Prepared::NotYet(old); } @@ -421,7 +437,7 @@ fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: E .ok() .filter(|next| next.previous == previous); - let own = old.get_actor().clone(); + own.push(old.get_actor().clone()); let clock = next .as_ref() .map_or_else(BTreeMap::new, |next| next.clock.clone()); @@ -573,7 +589,7 @@ mod tests { #[test] fn replays_within_the_window() { let (old, swarm, target) = scenario(&json!({})); - match prepare(swarm, old, Epoch::GENESIS, target) { + match prepare(swarm, old, Epoch::GENESIS, target, Vec::new()) { Prepared::Ready { plan: Plan::Replay(paths, _), own_missing, @@ -589,7 +605,7 @@ mod tests { #[test] fn quarantines_outside_the_window() { let (old, swarm, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } })); - match prepare(swarm, old, Epoch::GENESIS, target) { + match prepare(swarm, old, Epoch::GENESIS, target, Vec::new()) { Prepared::Ready { plan: Plan::Quarantine { paths, .. }, .. @@ -602,7 +618,7 @@ mod tests { fn new_node_has_nothing_to_replay() { let (_, swarm, target) = scenario(&json!({})); assert!(matches!( - prepare(swarm, AutoCommit::new(), Epoch::GENESIS, target), + prepare(swarm, AutoCommit::new(), Epoch::GENESIS, target, Vec::new()), Prepared::Ready { plan: Plan::Nothing, had_history: false, @@ -615,7 +631,7 @@ mod tests { fn waits_for_the_target_epoch() { let (old, _, target) = scenario(&json!({})); assert!(matches!( - prepare(AutoCommit::new(), old, Epoch::GENESIS, target), + prepare(AutoCommit::new(), old, Epoch::GENESIS, target, Vec::new()), Prepared::NotYet(_) )); } diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs index 9bcf8ba..427de08 100644 --- a/src/remake/coordinator.rs +++ b/src/remake/coordinator.rs @@ -12,7 +12,7 @@ //! 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. +//! a grace period later, 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. @@ -141,7 +141,23 @@ fn obstacle(id: &str, record: &Record, all: &[(String, Record)]) -> Option<Strin } async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), StoreError> { - let Some(mut request) = record.snapshot_request(id) else { + let content = match &record.proposal.content { + Some(reference) => match content::ensure(host, reference).await { + Ok(bytes) => Some(bytes), + Err(reason) => return decide_abort(host, id, reason).await, + }, + None => None, + }; + + // Clone under the read lock, build without any lock. + let mut copy = host.read_doc(|doc| doc.clone()).await; + // A base left to the commit is what this node has now. Every node that sees the + // decision has it too: the decision change depends on it. + let (base, fixed_now) = match &record.proposal.base_heads { + None => (Some(copy.get_heads()), true), + Some(heads) => (records::decode_heads(heads), false), + }; + let Some(mut request) = base.and_then(|base| record.snapshot_request_at(id, base)) else { return decide_abort( host, id, @@ -149,15 +165,8 @@ 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; + let base_heads = fixed_now.then(|| records::encode_heads(&request.base_heads)); + request.content = content; let mut snapshot = match snapshot::build_detached(copy, request).await { Ok((snapshot, _copy)) => snapshot, Err(error) => { @@ -170,6 +179,7 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor let decision = Decision::Committed { to_epoch: to_epoch.into(), + base_heads, snapshot_bytes, decided_at: records::unix_millis(), }; diff --git a/src/remake/delta.rs b/src/remake/delta.rs index c8f6b7c..51cc409 100644 --- a/src/remake/delta.rs +++ b/src/remake/delta.rs @@ -159,22 +159,23 @@ pub(crate) struct Missing { /// Splits `doc`'s history against a snapshot `clock`: a change (actor, seq) is in the /// snapshot iff seq <= clock[actor]. /// -/// Only changes this node wrote itself (actor `own`) are worth replaying: other nodes' -/// changes after the base were replayed by those nodes when they switched, and replaying -/// them again would overwrite newer writes in the new epoch with stale ones. +/// Only changes this node wrote itself (any of the actors in `own`: one per run and +/// epoch) are worth replaying when it rejoins: other nodes' changes after the base were +/// replayed by the nodes that switched, and replaying them again would overwrite newer +/// writes in the new epoch with stale ones. pub(crate) fn missing_from_snapshot( doc: &mut AutoCommit, clock: &BTreeMap<String, u64>, - own: &ActorId, + own: &[ActorId], ) -> Missing { - let own_hex = own.to_string(); + let own_hex: HashSet<String> = own.iter().map(ActorId::to_string).collect(); let changes: Vec<(ChangeHash, Vec<ChangeHash>, bool)> = doc .get_changes(&[]) .iter() .map(|change| { let actor = change.actor_id().to_string(); let in_snapshot = clock.get(&actor).is_some_and(|&seq| change.seq() <= seq); - let own_missing = actor == own_hex && !in_snapshot; + let own_missing = own_hex.contains(&actor) && !in_snapshot; (change.hash(), change.deps().to_vec(), own_missing) }) .collect(); @@ -370,7 +371,7 @@ mod tests { doc.commit(); let after = doc.get_heads(); - let missing = missing_from_snapshot(&mut doc, &clock, &own); + let missing = missing_from_snapshot(&mut doc, &clock, std::slice::from_ref(&own)); assert_eq!(missing.own, 1); assert_eq!(missing.before, with_foreign); assert_ne!(missing.before, base); @@ -381,7 +382,7 @@ mod tests { // Nothing of ours missing: nothing to replay. let full: BTreeMap<String, u64> = [(own.to_string(), 2)].into_iter().collect(); - let none = missing_from_snapshot(&mut doc, &full, &own); + let none = missing_from_snapshot(&mut doc, &full, std::slice::from_ref(&own)); assert_eq!(none.own, 0); let mut heads = after.clone(); heads.sort(); diff --git a/src/remake/kind.rs b/src/remake/kind.rs index 8131f7f..42bdd88 100644 --- a/src/remake/kind.rs +++ b/src/remake/kind.rs @@ -17,6 +17,9 @@ //! Protocol data under `_continuum` is not a kind's business: the generic part carries //! it forward and replays it unchanged. +use std::path::Path; + +use anyhow::{Context, anyhow, bail}; use automerge::AutoCommit; use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; use serde::{Deserialize, Serialize}; @@ -24,6 +27,7 @@ use serde_json::Value as JsonValue; use super::snapshot::{META_ROOT, canonical}; use crate::document::store::{self, StoreError}; +use crate::files::{FileKind, Files}; /// Name of the kind used when a request doesn't name one. pub(crate) const DEFAULT_KIND: &str = "compact"; @@ -71,6 +75,14 @@ pub(crate) trait RemakeKind: Send + Sync + 'static { /// Every node: where a user path written after the base goes. fn translate(&self, path: &[String], translation: &JsonValue) -> Translation; + + /// Whether the coordinator fixes the base when it commits instead of when it + /// proposes (no `baseHeads` then). Right for a kind whose content is the old + /// document at the base: the later the base, the fewer writes every switching node + /// replays as fresh values (whole lists and texts, concurrently on every node). + fn base_at_commit(&self) -> bool { + false + } } /// The kind with this name. @@ -83,6 +95,50 @@ pub(crate) fn by_name(name: &str) -> Option<&'static dyn RemakeKind> { } } +/// Builds a new swarm's genesis content into `target` (`--genesis --remake SPEC`), the +/// way a remake of that kind builds a new epoch from an empty one: +/// * `import:PATH`: an Automerge document, e.g. a backup (its content, without history +/// and protocol data); +/// * `replace:PATH`: a JSON document, the store root; +/// * `compact`: nothing (an empty document compacts to itself). +pub(crate) fn remake_genesis( + spec: &str, + files: &Files, + target: &mut AutoCommit, +) -> anyhow::Result<()> { + let (name, path) = match spec.split_once(':') { + Some((name, path)) => (name, Some(Path::new(path))), + None => (spec, None), + }; + let kind = by_name(name).ok_or_else(|| anyhow!("unknown remake kind {name:?}"))?; + let request = match (kind.name(), path) { + ("compact", None) => return Ok(()), + ("compact", Some(_)) => bail!("compact takes no file"), + (_, None) => bail!("{name} needs a file: {name}:PATH"), + (name, Some(path)) => { + let bytes = files + .read(FileKind::Remake, path) + .with_context(|| format!("Reading {}", path.display()))? + .with_context(|| format!("{} doesn't exist", path.display()))?; + match name { + "import" => serde_json::json!({ "document": BASE64.encode(&bytes) }), + "replace" => { + let document: JsonValue = serde_json::from_slice(&bytes) + .with_context(|| format!("{} is not JSON", path.display()))?; + serde_json::json!({ "document": document }) + } + other => bail!("{other} can't make a genesis document"), + } + } + }; + let prepared = kind + .propose(&request) + .map_err(|reason| anyhow!("{name}: {reason}"))?; + kind.build(&AutoCommit::new(), prepared.content.as_deref(), target)?; + target.commit(); + Ok(()) +} + /// Paths owned by the protocol rather than by any kind. pub(crate) fn is_reserved(path: &[String]) -> bool { path.first().is_some_and(|root| root == META_ROOT) @@ -97,6 +153,10 @@ impl RemakeKind for Compact { "compact" } + fn base_at_commit(&self) -> bool { + true + } + fn propose(&self, spec: &JsonValue) -> Result<Prepared, String> { match spec { JsonValue::Null => Ok(Prepared::default()), diff --git a/src/remake/local.rs b/src/remake/local.rs index 04ac0b0..65a2735 100644 --- a/src/remake/local.rs +++ b/src/remake/local.rs @@ -9,7 +9,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::{Arc, Mutex, MutexGuard}; use std::time::Duration; -use automerge::AutoCommit; +use automerge::{ActorId, AutoCommit}; use serde::Serialize; use tokio::time::Instant; @@ -55,6 +55,11 @@ struct Inner { content_fetches: HashSet<[u8; 32]>, /// Remakes this node coordinates now (`coordinator`). coordinating: HashSet<String>, + /// The actors this node writes or wrote with in the current document: this run's, + /// and those of earlier runs it knows of (`--document-file`). Its own writes, for a + /// rejoin. A switch or rejoin starts the list over with the new document's actor, so + /// it only grows with restarts within one epoch. + own_actors: Vec<ActorId>, } /// Content no remake refers to any more is kept this long after it was stored, for @@ -147,6 +152,24 @@ impl RemakeState { self.started_at } + /// This node writes with `actor` from now on (or did, in an earlier run). + pub(crate) fn add_own_actor(&self, actor: ActorId) { + let mut inner = self.lock(); + if !inner.own_actors.contains(&actor) { + inner.own_actors.push(actor); + } + } + + /// The current document is a new one, written with `actor` only (a switch or rejoin + /// is done). + pub(crate) fn reset_own_actors(&self, actor: ActorId) { + self.lock().own_actors = vec![actor]; + } + + pub(crate) fn own_actors(&self) -> Vec<ActorId> { + self.lock().own_actors.clone() + } + /// 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()) diff --git a/src/remake/mod.rs b/src/remake/mod.rs index ca17956..9085475 100644 --- a/src/remake/mod.rs +++ b/src/remake/mod.rs @@ -32,6 +32,7 @@ pub(crate) mod snapshot; mod switch; mod vote; +pub(crate) use kind::remake_genesis; pub(crate) use local::RemakeState; pub(crate) use switch::run_worker; diff --git a/src/remake/records.rs b/src/remake/records.rs index cd3e5be..283f612 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -8,9 +8,10 @@ //! * `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 +//! Once a remake has expired, its coordinator no longer commits it, and a grace period +//! later any node may abort it on an operator's request (`api`). Should clocks disagree +//! by more than that grace period, 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. //! @@ -130,7 +131,9 @@ pub(crate) struct Proposal { pub(crate) coordinator: String, pub(crate) from_epoch: EpochRecord, /// Hex change hashes, sorted. The proposal record itself is not part of the base. - pub(crate) base_heads: Vec<String>, + /// None for a kind whose base is fixed when it commits (`Decision::Committed`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) base_heads: Option<Vec<String>>, pub(crate) policy: Policy, pub(crate) settle_ms: u64, pub(crate) proposed_at: u64, @@ -181,13 +184,19 @@ fn default_kind() -> String { DEFAULT_KIND.to_owned() } -impl Proposal { - pub(crate) fn base_change_hashes(&self) -> Option<Vec<ChangeHash>> { - self.base_heads - .iter() - .map(|hex| decode_hex32(hex).map(ChangeHash)) - .collect() - } +/// Hex change hashes as hashes; None if one is unreadable. +pub(crate) fn decode_heads(heads: &[String]) -> Option<Vec<ChangeHash>> { + heads + .iter() + .map(|hex| decode_hex32(hex).map(ChangeHash)) + .collect() +} + +/// Hashes as sorted hex, as records keep them. +pub(crate) fn encode_heads(heads: &[ChangeHash]) -> Vec<String> { + let mut hex: Vec<String> = heads.iter().map(ChangeHash::to_string).collect(); + hex.sort(); + hex } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -196,6 +205,9 @@ pub(crate) enum Decision { #[serde(rename_all = "camelCase")] Committed { to_epoch: EpochRecord, + /// The base, when the proposal leaves it to the commit (sorted hex). + #[serde(default, skip_serializing_if = "Option::is_none")] + base_heads: Option<Vec<String>>, snapshot_bytes: u64, decided_at: u64, }, @@ -275,13 +287,39 @@ impl Record { } } + /// The base: the proposal's, or for a kind that fixes it at the commit, the + /// decision's. None until committed then, or if unreadable. + pub(crate) fn base(&self) -> Option<Vec<ChangeHash>> { + let heads = match (&self.proposal.base_heads, &self.decision) { + (Some(heads), _) => heads, + ( + None, + Some(Decision::Committed { + base_heads: Some(heads), + .. + }), + ) => heads, + _ => return None, + }; + decode_heads(heads) + } + /// Everything needed to (re)build this remake's snapshot but the content (which the - /// caller adds, `content::ensure`); None if the proposal is unreadable. + /// caller adds, `content::ensure`); None if the proposal or base is unreadable. pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> { + self.snapshot_request_at(id, self.base()?) + } + + /// The same, at `base` (the coordinator fixing a base at the commit). + pub(crate) fn snapshot_request_at( + &self, + id: &str, + base: Vec<ChangeHash>, + ) -> Option<SnapshotRequest> { Some(SnapshotRequest { remake_id: id.to_owned(), previous: self.proposal.from_epoch.to_epoch()?, - base_heads: self.proposal.base_change_hashes()?, + base_heads: base, policy: serde_json::to_value(&self.proposal.policy).ok()?, proposed_at: self.proposal.proposed_at, kind: self.proposal.kind.clone(), @@ -499,7 +537,7 @@ mod tests { Proposal { coordinator: "node-a".into(), from_epoch: Epoch::GENESIS.into(), - base_heads: vec!["ab".repeat(32)], + base_heads: Some(vec!["ab".repeat(32)]), policy: Policy { replay_window: Policy::default_replay_window(), archive_retention: Policy::default_archive_retention(), @@ -535,6 +573,7 @@ mod tests { generation: 1, id: "cd".repeat(32), }, + base_heads: None, snapshot_bytes: 123, decided_at: 11_000, }; @@ -552,7 +591,7 @@ mod tests { let (id, record) = &all[0]; assert_eq!(id, "c-1"); assert_eq!(record.proposal.policy, p.policy); - assert_eq!(record.proposal.base_change_hashes().unwrap().len(), 1); + assert_eq!(record.base().unwrap().len(), 1); assert!(record.abort_requested.is_some()); assert_eq!(record.votes.keys().collect::<Vec<_>>(), ["node-b"]); assert_eq!(record.proposal.quorum, 2); @@ -608,6 +647,31 @@ mod tests { } #[test] + fn a_base_fixed_at_the_commit() { + let mut record = Record { + proposal: Proposal { + base_heads: None, + ..proposal(10_000) + }, + decision: None, + abort_requested: None, + contested: false, + votes: BTreeMap::new(), + }; + assert_eq!(record.base(), None); + record.decision = Some(Decision::Committed { + to_epoch: EpochRecord { + generation: 1, + id: "cd".repeat(32), + }, + base_heads: Some(vec!["ef".repeat(32)]), + snapshot_bytes: 1, + decided_at: 2, + }); + assert_eq!(record.base(), Some(vec![ChangeHash([0xef; 32])])); + } + + #[test] fn disagreeing_decisions_are_contested() { let mut base = AutoCommit::new(); write_proposal(&mut base, "c-1", &proposal(10_000)).unwrap(); @@ -618,6 +682,7 @@ mod tests { generation: 1, id: "cd".repeat(32), }, + base_heads: None, snapshot_bytes: 1, decided_at: 2, }; diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs index e50a2a2..20569e0 100644 --- a/src/remake/snapshot.rs +++ b/src/remake/snapshot.rs @@ -38,6 +38,8 @@ pub(crate) const REMAKES: &str = "remakes"; /// The votes register (`records`). pub(crate) const VOTES: &str = "votes"; pub(crate) const EPOCHS: &str = "epochs"; +/// The swarm ID, written once by the genesis node (`swarm::bootstrap`). +pub(crate) const SWARM: &str = "swarm"; /// Everything that determines a snapshot besides the base content. pub(crate) struct SnapshotParams<'a> { @@ -224,6 +226,38 @@ pub(crate) fn verify_document(doc: &mut AutoCommit, epoch: Epoch) -> Result<Epoc parse_metadata(&at_snapshot, epoch.generation) } +/// The epoch `doc` is on: genesis if it holds no epoch metadata; otherwise the newest +/// generation it has metadata for, whose snapshot change is the document's only change +/// without dependencies (verified). For a document loaded from a backup. +pub(crate) fn epoch_of(doc: &mut AutoCommit) -> Result<Epoch, String> { + let generation = store::keys_at(doc, &[META_ROOT.to_owned(), EPOCHS.to_owned()]) + .map_err(|error| error.to_string())? + .iter() + .filter_map(|key| key.parse::<u64>().ok()) + .max(); + let Some(generation) = generation.filter(|generation| *generation > 0) else { + return Ok(Epoch::GENESIS); + }; + let roots: Vec<ChangeHash> = doc + .get_changes(&[]) + .iter() + .filter(|change| change.deps().is_empty()) + .map(|change| change.hash()) + .collect(); + let [root] = roots.as_slice() else { + return Err(format!( + "generation {generation} needs exactly one change without dependencies (its snapshot), found {}", + roots.len() + )); + }; + let epoch = Epoch { + generation, + id: root.0, + }; + verify_document(doc, epoch)?; + Ok(epoch) +} + /// Metadata of an earlier generation, as the current document holds it. Snapshots /// carry the metadata of all earlier epochs forward (only remake records stay /// behind), so a node many epochs behind can still find the epoch that followed its own. @@ -521,6 +555,27 @@ mod tests { } #[test] + fn epoch_of_a_backup() { + let mut a = doc(b"actor-a"); + store::write_path(&mut a, &["k".into()], &json!(1)).unwrap(); + a.commit(); + assert_eq!(epoch_of(&mut a), Ok(Epoch::GENESIS)); + + let heads = a.get_heads(); + let policy = json!({}); + let mut snap = build(&mut a, ¶ms(&heads, &policy)).unwrap(); + // Saved and loaded, with a write since, as a backup would be. + let mut later = snap + .doc + .fork() + .with_actor(ActorId::from(b"actor-b".as_slice())); + store::write_path(&mut later, &["k".into()], &json!(2)).unwrap(); + later.commit(); + let mut backup = AutoCommit::load(&later.save()).unwrap(); + assert_eq!(epoch_of(&mut backup), Ok(snap.epoch)); + } + + #[test] fn metadata_written_and_remake_records_dropped() { let mut a = doc(b"actor-a"); store::write_path( @@ -695,7 +750,8 @@ mod tests { assert!(verify_document(&mut fetched, wrong).is_err()); let own = late.get_actor().clone(); - let missing = delta::missing_from_snapshot(&mut late, &metadata.clock, &own); + let missing = + delta::missing_from_snapshot(&mut late, &metadata.clock, std::slice::from_ref(&own)); assert_eq!(missing.own, 1); let heads = late.get_heads(); let mut adopted = fetched.with_actor(ActorId::from(&b"late-node-next"[..])); diff --git a/src/remake/switch.rs b/src/remake/switch.rs index cf5ea75..1a4dbf9 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -205,6 +205,10 @@ async fn switch<H: Host>( .take_prebuilt(id) .filter(|snapshot| snapshot.epoch == expected); + let actor = ActorId::random(); + host.remake().add_own_actor(actor.clone()); + let new_actor = actor.clone(); + // 2. Build (or reuse), verify and replay without holding any lock. let (new_doc, copy_heads, mut stats) = tokio::task::spawn_blocking( move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> { @@ -222,7 +226,7 @@ async fn switch<H: Host>( } // A fresh actor: this node's writes in the new epoch. - let mut doc = snapshot.doc.with_actor(ActorId::random()); + let mut doc = snapshot.doc.with_actor(actor); // Keep this remake's record visible in the new epoch, for its history. if carry_record { @@ -242,6 +246,7 @@ async fn switch<H: Host>( // 3. Swap, replaying what arrived since the copy. let (old, late) = swap_in(host, expected, new_doc, copy_heads, late_translator).await?; stats.absorb(late); + host.remake().reset_own_actors(new_actor); // 4. Writes the remake had no place for wait for an operator (the old document // holds them); the old document is kept as configured either way. diff --git a/src/remake/vote.rs b/src/remake/vote.rs index 16de87a..b4936dc 100644 --- a/src/remake/vote.rs +++ b/src/remake/vote.rs @@ -304,7 +304,7 @@ mod tests { proposal: Proposal { coordinator: electorate[0].to_owned(), from_epoch: Epoch::GENESIS.into(), - base_heads: Vec::new(), + base_heads: Some(Vec::new()), policy: Policy { replay_window: Policy::default_replay_window(), archive_retention: Policy::default_archive_retention(), |