diff options
Diffstat (limited to 'src/remake')
| -rw-r--r-- | src/remake/api.rs | 474 | ||||
| -rw-r--r-- | src/remake/catchup.rs | 506 | ||||
| -rw-r--r-- | src/remake/coordinator.rs | 183 | ||||
| -rw-r--r-- | src/remake/delta.rs | 387 | ||||
| -rw-r--r-- | src/remake/local.rs | 320 | ||||
| -rw-r--r-- | src/remake/mod.rs | 158 | ||||
| -rw-r--r-- | src/remake/quarantine.rs | 160 | ||||
| -rw-r--r-- | src/remake/records.rs | 428 | ||||
| -rw-r--r-- | src/remake/snapshot.rs | 637 | ||||
| -rw-r--r-- | src/remake/switch.rs | 248 |
10 files changed, 3501 insertions, 0 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs new file mode 100644 index 0000000..b9c0625 --- /dev/null +++ b/src/remake/api.rs @@ -0,0 +1,474 @@ +//! `/admin/remakes` handlers. +//! +//! * `POST /admin/remakes` proposes a remake of the current epoch (`POST +//! /admin/compaction` is a shortcut, kept from before). The receiving +//! node becomes its coordinator. `202` with the report, or with `?wait=true` the final +//! report once decided. A dropped request does not abort anything. `?force=true` +//! accepts a `settleSecs` below the dormancy threshold (tests, single-site clusters). +//! * `GET /admin/remakes` lists all remakes. +//! * `GET /admin/remakes/{id}` reports one; `?wait=true` long-polls until decided +//! (bounded by the proposal's own expiry). +//! * `GET /admin/epochs` shows the local epoch, the archived previous ones and whether +//! the node is catching up with a newer epoch from a peer. +//! * `DELETE /admin/remakes/{id}` asks the coordinator to abort. Only possible while +//! proposed; `409` once committed (a switch can't be undone) or already over. +//! `DELETE /admin/remakes` does the same for the remake of the local epoch that +//! is in progress, without having to know its ID. +//! +//! Request body for POST (all optional): +//! `{ "replayWindow": {"epochs": 1, "seconds": 604800}, +//! "archiveRetention": {"epochs": 2, "seconds": 2592000}, +//! "settleSecs": 480 }` + +use std::time::Duration; + +use axum::{ + body::Bytes, + extract::{Path, Query, State}, + http::{header::LOCATION, StatusCode}, + response::{IntoResponse, Response}, + Json, +}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value as JsonValue}; +use tokio::sync::broadcast; +use tracing::{error, info, warn}; +use uuid::Uuid; + +use super::coordinator; +use super::local::LocalStatus; +use super::records::{ + self, unix_millis, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, +}; +use super::Host; +use crate::document::store::StoreError; + +#[derive(Debug, Default, Deserialize)] +pub(crate) struct WaitQuery { + #[serde(default)] + wait: bool, +} + +#[derive(Debug, Default, Deserialize)] +pub(crate) struct StartQuery { + #[serde(default)] + wait: bool, + /// Allows a settle period shorter than the dormancy threshold. + #[serde(default)] + force: bool, +} + +#[derive(Debug, Default, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct StartRequest { + replay_window: Option<Retention>, + archive_retention: Option<Retention>, + settle_secs: Option<u64>, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct Report { + id: String, + phase: Phase, + #[serde(flatten)] + record: Record, + /// What happened to this remake on the node answering the request. + #[serde(skip_serializing_if = "Option::is_none")] + local: Option<LocalStatus>, +} + +#[derive(Debug, thiserror::Error)] +enum StartError { + #[error("remake {0} is already open for this epoch")] + Open(String), + #[error(transparent)] + Store(#[from] StoreError), +} + +pub(super) fn error_response( + status: StatusCode, + message: impl Into<String>, + details: Option<JsonValue>, +) -> Response { + let mut body = json!({ "error": message.into() }); + if let (Some(JsonValue::Object(extra)), Some(object)) = (details, body.as_object_mut()) { + object.extend(extra); + } + (status, Json(body)).into_response() +} + +fn grace<H: Host>(host: &H) -> u64 { + host.dormant_after().as_millis() as u64 +} + +fn report<H: Host>(host: &H, id: String, record: Record) -> Report { + Report { + phase: record.phase(unix_millis(), grace(host)), + local: host.remake().status(&id), + id, + record, + } +} + +async fn load_report<H: Host>(host: &H, id: &str) -> Option<Report> { + let wanted = id.to_owned(); + let record = host + .read_doc(move |doc| records::load(doc, &wanted)) + .await?; + Some(report(host, id.to_owned(), record)) +} + +/// Waits until the remake leaves `proposed`. Bounded by the proposal's own expiry, +/// at which point it is reported as `expired`. +async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> { + // A record that was readable may be briefly unreadable or absent while concurrent + // changes merge; give up only if that persists. + const MAX_MISSES: u32 = 5; + + let mut changes = host.subscribe_changes(); + let mut seen = false; + let mut misses = 0; + loop { + let Some(report) = load_report(host, id).await else { + misses += 1; + warn!(remake = %id, seen, misses, "Remake record not readable while waiting"); + if !seen || misses >= MAX_MISSES { + return None; + } + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + }; + seen = true; + misses = 0; + if report.phase.is_final() { + return Some(report); + } + // Re-read on every published change, and at least once a second: expiry is a + // matter of time, not of document changes. + if let Ok(Err(broadcast::error::RecvError::Closed)) = + tokio::time::timeout(Duration::from_secs(1), changes.recv()).await + { + tokio::time::sleep(Duration::from_secs(1)).await; + } + } +} + +pub(crate) async fn start<H: Host>( + State(host): State<H>, + Query(query): Query<StartQuery>, + body: Bytes, +) -> Response { + if !host.is_node_ready().await { + return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None); + } + + let request: StartRequest = if body.is_empty() { + StartRequest::default() + } else { + match serde_json::from_slice(&body) { + Ok(request) => request, + Err(error) => { + return error_response( + StatusCode::BAD_REQUEST, + format!("invalid request: {error}"), + None, + ) + } + } + }; + + let dormant_after = host.dormant_after(); + let settle = request + .settle_secs + .map(Duration::from_secs) + .unwrap_or(dormant_after * 2); + if settle < dormant_after && query.force { + warn!( + settle_secs = settle.as_secs_f64(), + dormant_after_secs = dormant_after.as_secs(), + "Forced settle period shorter than the dormancy threshold; a partition may go unnoticed" + ); + } else if settle < dormant_after { + return error_response( + StatusCode::BAD_REQUEST, + format!( + "settleSecs must be at least {} (the dormancy threshold), so a partition is detected before deciding; add ?force=true to override", + dormant_after.as_secs() + ), + None, + ); + } + + // Fail fast instead of proposing something that will abort anyway. + let peers = host.peer_summaries().await; + let dormant: Vec<String> = peers + .iter() + .filter(|peer| !peer.active) + .map(|peer| peer.id.clone()) + .collect(); + if !dormant.is_empty() { + return error_response( + StatusCode::CONFLICT, + "dormant peers block remake; forget them first (DELETE /peers/{pk})", + Some(json!({ "dormantPeers": dormant })), + ); + } + + let mut peers_considered: Vec<String> = peers.into_iter().map(|peer| peer.id).collect(); + peers_considered.sort(); + + let now = unix_millis(); + let settle_ms = settle.as_millis() as u64; + let id = Uuid::new_v4().to_string(); + let grace = grace(&host); + + let mut proposal = Proposal { + coordinator: host.node_id(), + from_epoch: host.current_epoch().into(), + base_heads: Vec::new(), + policy: Policy { + replay_window: request + .replay_window + .unwrap_or_else(Policy::default_replay_window), + archive_retention: request + .archive_retention + .unwrap_or_else(Policy::default_archive_retention), + }, + settle_ms, + proposed_at: now, + settle_until: now + settle_ms, + peers_considered, + }; + + // 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 + // leave a proposal without a coordinator. Dropping the handle doesn't cancel it. + let proposing = tokio::spawn(propose(host.clone(), id.clone(), proposal, now, grace)); + let written = match proposing.await { + Ok(written) => written, + Err(error) => { + error!(error = %error, "Proposal task failed"); + return error_response( + StatusCode::INTERNAL_SERVER_ERROR, + "failed to write the proposal", + None, + ); + } + }; + + let record = match written { + Ok(record) => record, + Err(StartError::Open(open)) => { + return error_response( + StatusCode::CONFLICT, + "another remake of this epoch is still open", + Some(json!({ "remake": open })), + ); + } + Err(StartError::Store(error)) => { + error!(error = %format!("{error:#}"), "Failed to write remake proposal"); + return error_response( + StatusCode::INTERNAL_SERVER_ERROR, + "failed to write the proposal", + None, + ); + } + }; + + info!( + remake = %id, + settle_secs = settle.as_secs(), + "Remake proposed" + ); + + if query.wait { + return match wait_until_final(&host, &id).await { + Some(report) => Json(report).into_response(), + None => error_response(StatusCode::NOT_FOUND, "remake record vanished", None), + }; + } + + let location = format!("/admin/remakes/{id}"); + ( + StatusCode::ACCEPTED, + [(LOCATION, location)], + Json(report(&host, id, record)), + ) + .into_response() +} + +/// Writes the proposal (unless another one is open for the epoch) and starts its +/// coordinator. Runs as its own task, independent of the HTTP request. +async fn propose<H: Host>( + host: H, + id: String, + mut proposal: Proposal, + now: u64, + grace: u64, +) -> Result<Record, StartError> { + let record_id = id.clone(); + let record = host + .commit_doc(move |doc| -> Result<Record, StartError> { + if let Some((open, _)) = records::load_all(doc).into_iter().find(|(_, record)| { + record.proposal.from_epoch == proposal.from_epoch + && record.phase(now, grace).is_open() + }) { + return Err(StartError::Open(open)); + } + + // The base is everything before the proposal; the proposal itself is not in it. + let mut heads: Vec<String> = doc.get_heads().iter().map(|h| h.to_string()).collect(); + heads.sort(); + proposal.base_heads = heads; + + records::write_proposal(doc, &record_id, &proposal)?; + Ok(Record { + proposal, + decision: None, + abort_requested: None, + }) + }) + .await?; + + tokio::spawn(coordinator::run(host, id)); + Ok(record) +} + +pub(crate) async fn list<H: Host>(State(host): State<H>) -> Response { + let all = host.read_doc(records::load_all).await; + let mut reports: Vec<Report> = all + .into_iter() + .map(|(id, record)| report(&host, id, record)) + .collect(); + reports.sort_by_key(|report| report.record.proposal.proposed_at); + Json(reports).into_response() +} + +pub(crate) async fn get<H: Host>( + State(host): State<H>, + Path(id): Path<String>, + Query(query): Query<WaitQuery>, +) -> Response { + let found = if query.wait { + wait_until_final(&host, &id).await + } else { + load_report(&host, &id).await + }; + match found { + Some(report) => Json(report).into_response(), + None => error_response(StatusCode::NOT_FOUND, "no such remake", None), + } +} + +/// `DELETE /admin/remakes/{id}`. +pub(crate) async fn abort<H: Host>( + State(host): State<H>, + Path(id): Path<String>, + Query(query): Query<WaitQuery>, +) -> Response { + abort_by_id(host, id, query.wait).await +} + +/// `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). +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 + }) + .min_by_key(|(_, record)| record.proposal.proposed_at) + .map(|(id, _)| id) + }) + .await; + + match open { + Some(id) => abort_by_id(host, id, query.wait).await, + None => error_response( + StatusCode::NOT_FOUND, + "no remake of the local epoch is in progress", + None, + ), + } +} + +async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { + 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::Committed => { + return error_response( + StatusCode::CONFLICT, + "already committed; the switch cannot be undone", + Some(json!({ "remake": current })), + ) + } + Phase::Aborted | Phase::Expired => { + return error_response( + StatusCode::CONFLICT, + "already over", + Some(json!({ "remake": current })), + ) + } + } + + // The coordinator applies the request on its next check (within about a second). + // If it is gone, the proposal expires anyway. + if current.record.abort_requested.is_none() { + let request = AbortRequest { + by: host.node_id(), + at: unix_millis(), + }; + let record_id = id.clone(); + if let Err(error) = host + .commit_doc(move |doc| records::write_abort_request(doc, &record_id, &request)) + .await + { + error!(remake = %id, error = %format!("{error:#}"), "Failed to write abort request"); + return error_response( + StatusCode::INTERNAL_SERVER_ERROR, + "failed to record the abort request", + None, + ); + } + info!(remake = %id, "Abort requested"); + } + + let report = if wait { + wait_until_final(&host, &id).await + } else { + load_report(&host, &id).await + }; + match report { + Some(report) if report.phase.is_final() => Json(report).into_response(), + Some(report) => (StatusCode::ACCEPTED, 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 { + let current = EpochRecord::from(host.current_epoch()); + Json(json!({ + "current": current, + "archived": host.remake().archived_epochs(), + "catchUp": host.remake().catch_up_report(), + "quarantine": host.remake().list_quarantine(|entry| entry.id.clone()), + })) + .into_response() +} diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs new file mode 100644 index 0000000..22aa0f8 --- /dev/null +++ b/src/remake/catchup.rs @@ -0,0 +1,506 @@ +//! Catching up with a newer epoch: the node rejoins. +//! +//! 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 +//! +//! 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; +//! 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 cluster never saw (none, for a new node): +//! * the metadata of the epoch that followed the node's own records which changes +//! went into it (per-actor clock); the rest are this node's missing changes; +//! * within that epoch's replay window they are replayed (state per path, as in a +//! switch), otherwise they are quarantined for an operator +//! (`/admin/quarantine`), as are changes that can't be told apart; +//! 5. archives the old document and serves requests again. A node that was never +//! ready (a new node) is made ready by bootstrap, which waits for the rejoin. +//! +//! 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::convert::Infallible; +use std::time::Duration; + +use automerge::{AutoCommit, ChangeHash}; +use tokio::time::Instant; +use tracing::{error, info, warn}; +use uuid::Uuid; + +use super::delta::{self, ReplayStats}; +use super::local::{CatchUp, LocalStatus, Quarantined, Rejoin}; +use super::records::unix_millis; +use super::snapshot::{self, EpochMetadata}; +use super::Host; +use crate::document::replica::Epoch; + +/// Pause before trying the same target again after a failed rejoin. +const RETRY: Duration = Duration::from_secs(30); +/// Lower bound for how long a rejoin may take to complete a sync. +const MIN_REJOIN_TIMEOUT: Duration = Duration::from_secs(60); + +/// 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.initial_sync_done() { + finish(host).await; + } else if started.elapsed() > timeout { + if let Some(rejoin) = host.remake().take_rejoin() { + let reason = format!( + "no peer on {target} completed a sync within {}s", + timeout.as_secs() + ); + abandon(host, rejoin, reason).await; + } + } + return; + } + + let local = host.current_epoch(); + let mut ahead: Vec<(String, Epoch)> = host + .peer_summaries() + .await + .into_iter() + .filter(|peer| peer.active) + .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)); + + let Some(generation) = ahead.iter().map(|(_, epoch)| epoch.generation).max() else { + if host.remake().catch_up().is_some() { + host.remake().set_catch_up(None); + } + return; + }; + + let mut targets: Vec<Epoch> = ahead + .iter() + .filter(|(_, epoch)| epoch.generation == generation) + .map(|(_, epoch)| *epoch) + .collect(); + targets.sort_by_key(|epoch| epoch.id); + targets.dedup(); + let [target] = targets.as_slice() else { + let listed: Vec<String> = targets.iter().map(ToString::to_string).collect(); + warn!(epochs = ?listed, "Peers disagree about the newest epoch; not catching up"); + return; + }; + let target = *target; + + // Back off between attempts for the same target. + let previous = host + .remake() + .catch_up() + .filter(|catch_up| catch_up.target == target); + if previous + .as_ref() + .is_some_and(|catch_up| catch_up.last_attempt.elapsed() < RETRY) + { + return; + } + let attempts = previous.map_or(0, |catch_up| catch_up.attempts) + 1; + + start(host, local, target, attempts).await; +} + +async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) { + host.remake().set_catch_up(Some(CatchUp { + target, + attempts, + last_attempt: Instant::now(), + error: None, + })); + + // Keep answering reads, from a copy of the current document marked stale, before + // the node stops being ready: no window in which reads get 503. + if host.is_node_ready().await { + let copy = host.read_doc(|doc| doc.clone()).await; + host.set_stale_view(Some((local, copy))); + } + let was_ready = host.suspend().await; + let old = match host + .switch_epoch(target, |_live| Ok::<_, Infallible>(AutoCommit::new())) + .await + { + Ok(old) => old, + Err(never) => match never {}, + }; + // After the swap: messages of the old epoch are dropped from now on, so nothing + // can mark the new, empty document as caught up by mistake. A stale "done" from + // just before is caught in `finish` (the snapshot change is missing then). + host.reset_initial_sync(); + + info!(from = %local, to = %target, attempt = attempts, "Rejoining newer epoch"); + host.remake().start_rejoin(Rejoin { + previous: local, + target, + old, + started: Instant::now(), + was_ready, + }); +} + +/// What to do with this node's own changes the new epoch lacks. +enum Plan { + Nothing, + Replay(Vec<Vec<String>>), + Quarantine { + paths: Vec<Vec<String>>, + reason: String, + }, +} + +enum Prepared { + /// Caught up with someone, but not with the target epoch yet: keep waiting. + NotYet(AutoCommit), + Failed(AutoCommit, String), + Ready { + old: AutoCommit, + metadata: EpochMetadata, + plan: Plan, + own_missing: usize, + had_history: bool, + }, +} + +async fn finish<H: Host>(host: &H) { + let Some(rejoin) = host.remake().take_rejoin() else { + return; + }; + let Rejoin { + previous, + target, + old, + started, + was_ready, + } = rejoin; + + // 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 (old, metadata, plan, own_missing, had_history) = match prepared { + Ok(Prepared::Ready { + old, + metadata, + plan, + own_missing, + had_history, + }) => (old, metadata, plan, own_missing, had_history), + Ok(Prepared::NotYet(old)) => { + host.reset_initial_sync(); + host.remake().start_rejoin(Rejoin { + previous, + target, + old, + started, + was_ready, + }); + return; + } + Ok(Prepared::Failed(old, reason)) => { + let rejoin = Rejoin { + previous, + target, + old, + started, + was_ready, + }; + return abandon(host, rejoin, reason).await; + } + Err(join_error) => { + // The old document is gone with the panicked task; nothing to restore. + error!(error = %join_error, "Rejoin task failed; staying on the new epoch"); + if was_ready { + host.resume().await; + } + host.set_stale_view(None); + return; + } + }; + + let mut stats = ReplayStats::default(); + let mut quarantine_id = None; + let mut keep_archive = true; + let mut old = Some(old); + + match plan { + Plan::Nothing => {} + Plan::Replay(paths) => { + let document = old.take().expect("old document"); + let (replayed, document) = host + .commit_doc(move |doc| { + let stats = delta::replay_paths(&document, doc, &paths); + Ok::<_, Infallible>((stats, document)) + }) + .await + .unwrap_or_else(|never| match never {}); + stats = replayed; + old = Some(document); + } + Plan::Quarantine { paths, reason } => { + let id = Uuid::new_v4().to_string(); + warn!(quarantine = %id, epoch = %previous, paths = paths.len(), reason = %reason, "Quarantined local changes"); + host.remake().add_quarantine(Quarantined { + id: id.clone(), + epoch: previous, + target, + created_at: unix_millis(), + reason, + paths, + document: old.take().expect("old document"), + }); + quarantine_id = Some(id); + keep_archive = false; // the quarantine holds the document + } + } + + if let (true, true, Some(document)) = (had_history, keep_archive, old) { + host.remake().archive( + previous, + document, + metadata.policy.archive_retention.clone(), + target.generation, + ); + } + + host.remake().set_status( + &metadata.remake_id, + LocalStatus { + switched_at: Some(unix_millis()), + replayed_paths: Some(stats.replayed), + failed_paths: Some(stats.failed), + quarantine: quarantine_id.clone(), + error: None, + }, + ); + host.remake().set_catch_up(None); + if was_ready { + host.resume().await; + } + // After resuming, so reads move from the stale view to the live document directly. + host.set_stale_view(None); + + info!( + from = %previous, + to = %target, + own_changes = own_missing, + replayed = stats.replayed, + failed = stats.failed, + quarantine = ?quarantine_id, + secs = started.elapsed().as_secs(), + "Rejoined newer epoch" + ); +} + +/// 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 { + if copy.get_change_by_hash(&ChangeHash(target.id)).is_none() { + return Prepared::NotYet(old); + } + let metadata = match snapshot::verify_document(&mut copy, target) { + Ok(metadata) => metadata, + Err(reason) => return Prepared::Failed(old, reason), + }; + + let old_heads = old.get_heads(); + if old_heads.is_empty() { + return Prepared::Ready { + old, + metadata, + plan: Plan::Nothing, + own_missing: 0, + had_history: false, + }; + } + + // The epoch that followed ours records (per actor) which of our changes it took. + // Snapshots carry earlier epochs' metadata forward, so this works across any gap. + let next = snapshot::read_epoch_metadata(©, previous.generation + 1) + .ok() + .filter(|next| next.previous == previous); + + let own = old.get_actor().clone(); + let clock = next + .as_ref() + .map_or_else(BTreeMap::new, |next| next.clock.clone()); + let missing = delta::missing_from_snapshot(&mut old, &clock, &own); + if missing.own == 0 { + return Prepared::Ready { + old, + metadata, + plan: Plan::Nothing, + own_missing: 0, + had_history: true, + }; + } + + let paths = delta::changed_paths(&mut old, &missing.before, &old_heads); + let gap = target.generation - previous.generation; + + let plan = match &next { + None => Plan::Quarantine { + paths, + reason: format!( + "the metadata of the epoch after {previous} is missing, so it is unknown which of {} local changes the cluster has", + missing.own + ), + }, + Some(next) => { + let away_ms = unix_millis().saturating_sub(next.proposed_at); + if next.policy.replay_window.covers(gap, away_ms) { + Plan::Replay(paths) + } else { + Plan::Quarantine { + paths, + reason: format!( + "{} local changes are outside the replay window ({gap} epochs behind, cluster moved on {}s ago)", + missing.own, + away_ms / 1000 + ), + } + } + } + }; + + Prepared::Ready { + old, + metadata, + plan, + own_missing: missing.own, + had_history: true, + } +} + +/// Restores the old document and epoch, and retries later. +async fn abandon<H: Host>(host: &H, rejoin: Rejoin, reason: String) { + let Rejoin { + previous, + target, + old, + was_ready, + .. + } = rejoin; + warn!(target = %target, reason = %reason, "Rejoin failed; restoring the previous epoch"); + + if let Err(never) = host + .switch_epoch(previous, move |_live| Ok::<_, Infallible>(old)) + .await + { + match never {} + } + // A node that was never ready must sync again before bootstrap may finish. + host.reset_initial_sync(); + if was_ready { + host.resume().await; + } + host.set_stale_view(None); + + let attempts = host + .remake() + .catch_up() + .filter(|catch_up| catch_up.target == target) + .map_or(1, |catch_up| catch_up.attempts); + host.remake().set_catch_up(Some(CatchUp { + target, + attempts, + last_attempt: Instant::now(), + error: Some(reason), + })); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::remake::snapshot::{build, SnapshotParams}; + use automerge::{transaction::Transactable, ActorId, ROOT}; + use serde_json::{json, Value as JsonValue}; + + /// An old-epoch document with one own change after the base, and the cluster's + /// document of the next epoch (snapshot plus one foreign change). + fn scenario(policy: &JsonValue) -> (AutoCommit, AutoCommit, Epoch) { + let mut old = AutoCommit::new().with_actor(ActorId::from(&b"late-node"[..])); + old.put(ROOT, "base", 1_i64).unwrap(); + old.commit(); + let base = old.get_heads(); + + let snapshot = build( + &mut old, + &SnapshotParams { + remake_id: "c-1", + previous: Epoch::GENESIS, + base_heads: &base, + policy, + proposed_at: 0, + }, + ) + .unwrap(); + let target = snapshot.epoch; + + let mut cluster = snapshot.doc.with_actor(ActorId::from(&b"cluster-node"[..])); + cluster.put(ROOT, "theirs", 1_i64).unwrap(); + cluster.commit(); + + old.put(ROOT, "mine", 1_i64).unwrap(); + old.commit(); + (old, cluster, target) + } + + #[test] + fn replays_within_the_window() { + let (old, cluster, target) = scenario(&json!({})); + match prepare(cluster, old, Epoch::GENESIS, target) { + Prepared::Ready { + plan: Plan::Replay(paths), + own_missing, + .. + } => { + assert_eq!(own_missing, 1); + assert_eq!(paths, vec![vec!["mine".to_owned()]]); + } + _ => panic!("expected a replay"), + } + } + + #[test] + fn quarantines_outside_the_window() { + let (old, cluster, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } })); + match prepare(cluster, old, Epoch::GENESIS, target) { + Prepared::Ready { + plan: Plan::Quarantine { paths, .. }, + .. + } => assert_eq!(paths, vec![vec!["mine".to_owned()]]), + _ => panic!("expected a quarantine"), + } + } + + #[test] + fn new_node_has_nothing_to_replay() { + let (_, cluster, target) = scenario(&json!({})); + assert!(matches!( + prepare(cluster, AutoCommit::new(), Epoch::GENESIS, target), + Prepared::Ready { + plan: Plan::Nothing, + had_history: false, + .. + } + )); + } + + #[test] + fn waits_for_the_target_epoch() { + let (old, _, target) = scenario(&json!({})); + assert!(matches!( + prepare(AutoCommit::new(), old, Epoch::GENESIS, target), + Prepared::NotYet(_) + )); + } +} diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs new file mode 100644 index 0000000..b84b0a9 --- /dev/null +++ b/src/remake/coordinator.rs @@ -0,0 +1,183 @@ +//! The coordinator: the node that received the proposal watches the cluster during the +//! settle period and then writes the decision. +//! +//! It aborts if, at any point before deciding: +//! * an operator requested an abort, +//! * a competing proposal for the same epoch appeared (both coordinators abort), +//! * a peer it considered went dormant (a partition may hide a competitor), +//! * the local epoch changed, or +//! * the decision would come later than the settle period plus the grace margin +//! (other nodes report the proposal as expired by then). +//! +//! Otherwise it builds the snapshot at the base heads and commits, recording the new +//! epoch so every node can verify its own build against it. +//! +//! The task lives only in memory. If the coordinator dies, nobody decides and the +//! proposal expires. + +use std::collections::HashMap; +use std::time::Duration; + +use tracing::{info, warn}; + +use super::records::{self, Decision, Record}; +use super::snapshot; +use super::Host; +use crate::document::store::StoreError; + +const TICK: Duration = Duration::from_secs(1); + +pub(crate) async fn run<H: Host>(host: H, id: String) { + info!(remake = %id, "Coordinating remake"); + + loop { + tokio::time::sleep(TICK).await; + + match step(&host, &id).await { + Ok(true) => return, + Ok(false) => {} + Err(error) => { + warn!(remake = %id, error = %format!("{error:#}"), "Coordinator step failed; retrying"); + } + } + } +} + +/// 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))) + .await; + + let Some(record) = record else { + warn!(remake = %id, "Remake record vanished; stopping coordinator"); + return Ok(true); + }; + if record.decision.is_some() { + return Ok(true); + } + + let now = records::unix_millis(); + let grace = host.dormant_after().as_millis() as u64; + + if let Some(reason) = abort_reason(host, id, &record, &all, now, grace).await { + decide_abort(host, id, reason).await?; + return Ok(true); + } + + if now < record.proposal.settle_until { + return Ok(false); + } + + commit(host, id, &record).await?; + Ok(true) +} + +async fn abort_reason<H: Host>( + host: &H, + id: &str, + record: &Record, + all: &[(String, Record)], + now: u64, + grace: u64, +) -> Option<String> { + let proposal = &record.proposal; + + if let Some(request) = &record.abort_requested { + return Some(format!("aborted by operator via {}", request.by)); + } + + if records::EpochRecord::from(host.current_epoch()) != proposal.from_epoch { + return Some("local epoch changed during settle".to_owned()); + } + + if let Some((other, _)) = all.iter().find(|(other, candidate)| { + other != id + && candidate.proposal.from_epoch == proposal.from_epoch + && candidate.phase(now, grace).is_open() + }) { + return Some(format!("competing remake {other} for the same epoch")); + } + + // Peers forgotten in the meantime are ignored: an operator decided they are gone. + let peers: HashMap<String, bool> = host + .peer_summaries() + .await + .into_iter() + .map(|peer| (peer.id, peer.active)) + .collect(); + let dormant: Vec<&str> = proposal + .peers_considered + .iter() + .filter(|peer| peers.get(peer.as_str()) == Some(&false)) + .map(String::as_str) + .collect(); + if !dormant.is_empty() { + return Some(format!( + "peers went dormant during settle: {}", + dormant.join(", ") + )); + } + + if now > proposal.settle_until.saturating_add(grace) { + return Some("decision deadline passed".to_owned()); + } + + None +} + +async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), StoreError> { + let Some(request) = record.snapshot_request(id) else { + return decide_abort( + host, + id, + "unreadable epoch or base heads in proposal".into(), + ) + .await; + }; + + // Clone under the read lock, build without any lock. + let copy = host.read_doc(|doc| doc.clone()).await; + let mut snapshot = match snapshot::build_detached(copy, request).await { + Ok((snapshot, _copy)) => snapshot, + Err(error) => { + return decide_abort(host, id, format!("snapshot build failed: {error}")).await; + } + }; + + let to_epoch = snapshot.epoch; + let snapshot_bytes = snapshot.doc.save().len() as u64; + + let decision = Decision::Committed { + to_epoch: to_epoch.into(), + 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"); + 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 +} + +async fn write_decision<H: Host>(host: &H, id: &str, decision: Decision) -> Result<(), StoreError> { + let id = id.to_owned(); + host.commit_doc(move |doc| records::write_decision(doc, &id, &decision)) + .await +} diff --git a/src/remake/delta.rs b/src/remake/delta.rs new file mode 100644 index 0000000..b07c862 --- /dev/null +++ b/src/remake/delta.rs @@ -0,0 +1,387 @@ +//! Replaying a node's changes after the base into the new epoch. +//! +//! Changes from the old epoch can't be applied to the new document (their history is +//! unrelated), so they are replayed as state: for every path that changed between two +//! sets of heads, the new document gets whatever the old document's merged view holds +//! there now (or a delete if nothing). Replayed values are fresh operations, so they win +//! over what the snapshot holds at the same path. +//! +//! Granularity: +//! * Map entries are replayed per key, so changes to sibling keys made on different +//! nodes don't overwrite each other. +//! * Lists and text are replayed whole: concurrent edits of the same list or text on two +//! nodes around a remake resolve as last-writer-wins. +//! * Edits inside a map that lost a merge conflict produce no patch in Automerge and are +//! not replayed. Such maps only exist briefly between concurrent creations. +//! +//! The protocol's own metadata (`_continuum/remakes`, `_continuum/epochs`) is never +//! replayed: the snapshot owns it. + +use std::collections::{BTreeMap, HashSet}; + +use automerge::{ActorId, AutoCommit, ChangeHash, PatchAction, Prop}; +use tracing::warn; + +use super::snapshot::{EPOCHS, META_ROOT, REMAKES}; +use crate::document::store::{self, StoreError}; + +#[derive(Debug, Default, Clone, Copy)] +pub(crate) struct ReplayStats { + pub(crate) replayed: usize, + pub(crate) failed: usize, +} + +/// Replays into `dst` everything that changed in `src` between `before` and `after`, +/// as one commit. Both head sets must be in `src`. +pub(crate) fn replay( + src: &mut AutoCommit, + before: &[ChangeHash], + after: &[ChangeHash], + dst: &mut AutoCommit, +) -> ReplayStats { + if before == after { + return ReplayStats::default(); + } + + let paths = changed_paths(src, before, after); + replay_paths(src, dst, &paths) +} + +/// Replays the state of `paths` (from [`changed_paths`]) from `src` into `dst`, as one +/// commit. Lets callers find the paths outside a lock and copy them inside it. +pub(crate) fn replay_paths( + src: &AutoCommit, + dst: &mut AutoCommit, + paths: &[Vec<String>], +) -> ReplayStats { + let mut stats = ReplayStats::default(); + for path in paths { + replay_path(src, dst, path, &mut stats); + } + let _ = dst.commit(); + stats +} + +fn replay_path(src: &AutoCommit, dst: &mut AutoCommit, path: &[String], stats: &mut ReplayStats) { + if is_protocol_path(path) { + return; + } + + // Replacing an ancestor of the protocol metadata as a whole would clobber it; replay + // its children one by one instead. + if contains_protocol_path(path) { + let mut keys = store::keys_at(src, path).unwrap_or_default(); + keys.extend(store::keys_at(dst, path).unwrap_or_default()); + keys.sort(); + keys.dedup(); + for key in keys { + let mut child = path.to_vec(); + child.push(key); + replay_path(src, dst, &child, stats); + } + return; + } + + match store::copy_path(src, dst, path) { + Ok(()) => stats.replayed += 1, + Err(error) => { + log_failure(path, &error); + stats.failed += 1; + } + } +} + +fn log_failure(path: &[String], error: &StoreError) { + warn!(path = ?path, error = %error, "Could not replay change into the new epoch"); +} + +fn is_protocol_path(path: &[String]) -> bool { + matches!(path, [root, part, ..] if root == META_ROOT && (part == REMAKES || part == EPOCHS)) +} + +/// True for the root and for `_continuum` itself. +fn contains_protocol_path(path: &[String]) -> bool { + matches!(path, [] | [_]) && path.first().is_none_or(|root| root == META_ROOT) +} + +/// What a late node has that a snapshot lacks. +#[derive(Debug)] +pub(crate) struct Missing { + /// Heads of everything except `own` changes the snapshot lacks (and changes built on + /// them). Replaying from these heads to the document's heads replays exactly those. + pub(crate) before: Vec<ChangeHash>, + /// Changes by `own` that the snapshot lacks. + pub(crate) own: usize, +} + +/// 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. +pub(crate) fn missing_from_snapshot( + doc: &mut AutoCommit, + clock: &BTreeMap<String, u64>, + own: &ActorId, +) -> Missing { + let own_hex = own.to_string(); + 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; + (change.hash(), change.deps().to_vec(), own_missing) + }) + .collect(); + + // Excluded: own missing changes and everything that (transitively) depends on them. + // A fixpoint, so it doesn't rely on the order `get_changes` returns. + let mut excluded: HashSet<ChangeHash> = changes + .iter() + .filter(|(_, _, own_missing)| *own_missing) + .map(|(hash, _, _)| *hash) + .collect(); + let own_count = excluded.len(); + loop { + let before = excluded.len(); + for (hash, deps, _) in &changes { + if !excluded.contains(hash) && deps.iter().any(|dep| excluded.contains(dep)) { + excluded.insert(*hash); + } + } + if excluded.len() == before { + break; + } + } + + // Heads of the kept part: kept changes no kept change depends on. + let kept: Vec<&(ChangeHash, Vec<ChangeHash>, bool)> = changes + .iter() + .filter(|(hash, _, _)| !excluded.contains(hash)) + .collect(); + let depended_on: HashSet<ChangeHash> = kept + .iter() + .flat_map(|(_, deps, _)| deps.iter().copied()) + .collect(); + let mut before: Vec<ChangeHash> = kept + .iter() + .map(|(hash, _, _)| *hash) + .filter(|hash| !depended_on.contains(hash)) + .collect(); + before.sort(); + + Missing { + before, + own: own_count, + } +} + +/// The minimal set of store paths covering every change between `before` and `after`: +/// sorted, and no path is a prefix of another. +pub(crate) fn changed_paths( + doc: &mut AutoCommit, + before: &[ChangeHash], + after: &[ChangeHash], +) -> Vec<Vec<String>> { + let mut paths: Vec<Vec<String>> = doc.diff(before, after).iter().map(patch_path).collect(); + paths.sort(); + paths.dedup(); + + // Sorted order puts a prefix right before its extensions, so comparing with the + // last kept path is enough. + let mut minimal: Vec<Vec<String>> = Vec::new(); + for path in paths { + if minimal.last().is_some_and(|kept| path.starts_with(kept)) { + continue; + } + minimal.push(path); + } + minimal +} + +fn patch_path(patch: &automerge::Patch) -> Vec<String> { + let mut path = Vec::new(); + for (_, prop) in &patch.path { + match prop { + Prop::Map(key) => path.push(key.clone()), + // Inside a list: the list is replayed as a whole. + Prop::Seq(_) => return path, + } + } + + match &patch.action { + PatchAction::PutMap { key, .. } | PatchAction::DeleteMap { key, .. } => { + path.push(key.clone()) + } + PatchAction::Increment { + prop: Prop::Map(key), + .. + } => path.push(key.clone()), + // List and text edits: the containing object. + _ => {} + } + path +} + +#[cfg(test)] +mod tests { + use super::*; + use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, ROOT}; + use serde_json::json; + + fn view(doc: &AutoCommit) -> serde_json::Value { + store::read_path(doc, &[]).unwrap().unwrap() + } + + #[test] + fn minimal_paths() { + let mut doc = AutoCommit::new(); + doc.put(ROOT, "untouched", 1_i64).unwrap(); + let before = doc.get_heads(); + + let m = doc.put_object(ROOT, "m", ObjType::Map).unwrap(); + doc.put(&m, "a", 1_i64).unwrap(); + doc.put(ROOT, "x", 1_i64).unwrap(); + doc.put(ROOT, "c", ScalarValue::counter(0)).unwrap(); + doc.commit(); + let middle = doc.get_heads(); + doc.increment(ROOT, "c", 2).unwrap(); + let list = doc.put_object(ROOT, "l", ObjType::List).unwrap(); + doc.insert(&list, 0, "e").unwrap(); + doc.commit(); + let after = doc.get_heads(); + + let all = changed_paths(&mut doc, &before, &after); + assert_eq!( + all, + vec![ + vec!["c".to_owned()], + vec!["l".to_owned()], + vec!["m".to_owned()], + vec!["x".to_owned()], + ] + ); + + let later = changed_paths(&mut doc, &middle, &after); + assert_eq!(later, vec![vec!["c".to_owned()], vec!["l".to_owned()]]); + } + + #[test] + fn replays_state_per_key() { + // Base shared by the old document and the snapshot. + let mut old = AutoCommit::new(); + let m = old.put_object(ROOT, "m", ObjType::Map).unwrap(); + old.put(&m, "keep", "base").unwrap(); + old.put(&m, "change", "base").unwrap(); + old.put(ROOT, "gone", true).unwrap(); + old.commit(); + let base = old.get_heads(); + + let mut new = AutoCommit::new(); + store::materialize_into(&old, &mut new, &|_| false).unwrap(); + // Something the new epoch already holds that the delta must not clobber. + let Some((_, m_new)) = new.get(ROOT, "m").unwrap() else { + panic!("m missing"); + }; + new.put(&m_new, "keep", "new epoch").unwrap(); + new.commit(); + + // The delta: one key changed, one key deleted. + old.put(&m, "change", "delta").unwrap(); + old.delete(ROOT, "gone").unwrap(); + old.commit(); + let after = old.get_heads(); + + let stats = replay(&mut old, &base, &after, &mut new); + assert_eq!(stats.failed, 0); + assert_eq!( + view(&new), + json!({ "m": { "keep": "new epoch", "change": "delta" } }) + ); + } + + #[test] + fn missing_counts_only_own_changes() { + let own = ActorId::from(&b"own-actor"[..]); + let other = ActorId::from(&b"other-actor"[..]); + + let mut doc = AutoCommit::new().with_actor(own.clone()); + doc.put(ROOT, "base", 1_i64).unwrap(); + doc.commit(); + let base = doc.get_heads(); + + // The snapshot contains the base only. + let clock: BTreeMap<String, u64> = [(own.to_string(), 1)].into_iter().collect(); + + // A foreign change after the base (replayed by its own node) ... + let mut foreign = doc.fork().with_actor(other); + foreign.put(ROOT, "theirs", 1_i64).unwrap(); + foreign.commit(); + doc.merge(&mut foreign).unwrap(); + let with_foreign = doc.get_heads(); + + // ... and one of ours. + doc.put(ROOT, "mine", 1_i64).unwrap(); + doc.commit(); + let after = doc.get_heads(); + + let missing = missing_from_snapshot(&mut doc, &clock, &own); + assert_eq!(missing.own, 1); + assert_eq!(missing.before, with_foreign); + assert_ne!(missing.before, base); + assert_eq!( + changed_paths(&mut doc, &missing.before, &after), + vec![vec!["mine".to_owned()]] + ); + + // 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); + assert_eq!(none.own, 0); + let mut heads = after.clone(); + heads.sort(); + assert_eq!(none.before, heads); + } + + #[test] + fn protocol_metadata_is_never_replayed() { + let mut old = AutoCommit::new(); + old.put(ROOT, "x", 1_i64).unwrap(); + old.commit(); + let base = old.get_heads(); + + let mut new = AutoCommit::new(); + store::write_path( + &mut new, + &[META_ROOT.into(), EPOCHS.into(), "1".into()], + &json!({ "generation": 1 }), + ) + .unwrap(); + new.commit(); + + // The old epoch replaces `_continuum` wholesale and writes a remake record. + store::write_path( + &mut old, + &[META_ROOT.into()], + &json!({ "forgotten": { "pk": 1 } }), + ) + .unwrap(); + store::write_path( + &mut old, + &[META_ROOT.into(), REMAKES.into(), "c".into()], + &json!({ "x": 1 }), + ) + .unwrap(); + old.commit(); + let after = old.get_heads(); + + replay(&mut old, &base, &after, &mut new); + let meta = &view(&new)[META_ROOT]; + assert_eq!(meta[EPOCHS]["1"]["generation"], json!(1)); + assert_eq!(meta["forgotten"]["pk"], json!(1)); + assert!(meta.get(REMAKES).is_none()); + } +} diff --git a/src/remake/local.rs b/src/remake/local.rs new file mode 100644 index 0000000..35a4ff1 --- /dev/null +++ b/src/remake/local.rs @@ -0,0 +1,320 @@ +//! 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. +//! +//! 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 automerge::AutoCommit; +use serde::Serialize; +use tokio::time::Instant; + +use super::records::{unix_millis, Retention}; +use super::snapshot::Snapshot; +use crate::document::replica::Epoch; + +#[derive(Default)] +pub(crate) struct RemakeState { + // A std mutex: every access is short and never spans an await. + inner: Mutex<Inner>, +} + +#[derive(Default)] +struct Inner { + /// Built by the coordinator at commit time; consumed by the local switch. + prebuilt: HashMap<String, Snapshot>, + /// When this node first saw a remake committed. + first_seen: HashMap<String, Instant>, + status: HashMap<String, LocalStatus>, + archive: Vec<Archived>, + /// Last attempt to catch up with a newer epoch. + catch_up: Option<CatchUp>, + /// A rejoin in progress: the node syncs an empty document of a newer epoch. + rejoin: Option<Rejoin>, + quarantine: Vec<Quarantined>, +} + +/// Catching up with a newer epoch: attempts and the last failure. +#[derive(Clone, Debug)] +pub(crate) struct CatchUp { + pub(crate) target: Epoch, + pub(crate) attempts: u32, + pub(crate) last_attempt: Instant, + pub(crate) error: Option<String>, +} + +/// The node left `previous` and is syncing `target` from scratch. +pub(crate) struct Rejoin { + pub(crate) previous: Epoch, + pub(crate) target: Epoch, + /// The document of `previous`, until it is replayed, quarantined or restored. + pub(crate) old: AutoCommit, + pub(crate) started: Instant, + /// The node served requests before; it does again once the rejoin is over. + pub(crate) was_ready: bool, +} + +/// Changes of this node that were not replayed into the current epoch. +pub(crate) struct Quarantined { + pub(crate) id: String, + /// The epoch the changes were made in, and the one the node rejoined. + pub(crate) epoch: Epoch, + pub(crate) target: Epoch, + pub(crate) created_at: u64, + pub(crate) reason: String, + /// Store paths the changes touched (minimal, prefix-free). + pub(crate) paths: Vec<Vec<String>>, + /// The old document; `paths` are read from its merged view. + pub(crate) document: AutoCommit, +} + +#[derive(Clone, Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct CatchUpReport { + pub(crate) target: String, + pub(crate) attempts: u32, + /// True while the node syncs the target epoch (not serving requests). + pub(crate) rejoining: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) error: Option<String>, +} + +/// What happened to a remake on this node. +#[derive(Clone, Debug, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct LocalStatus { + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) switched_at: Option<u64>, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) replayed_paths: Option<usize>, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) failed_paths: Option<usize>, + /// ID of the quarantine entry holding this node's changes that weren't replayed. + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) quarantine: Option<String>, + /// Why the switch did not happen (yet). + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) error: Option<String>, +} + +struct Archived { + epoch: Epoch, + // Kept for auditing and recovery by hand; not read by the protocol. + #[allow(dead_code)] + doc: AutoCommit, + archived_at: u64, + retention: Retention, +} + +/// Summary of an archived epoch. +#[derive(Clone, Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ArchivedEpoch { + pub(crate) generation: u64, + pub(crate) id: String, + pub(crate) archived_at: u64, +} + +impl RemakeState { + fn lock(&self) -> MutexGuard<'_, Inner> { + self.inner + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + pub(crate) fn store_prebuilt(&self, id: &str, snapshot: Snapshot) { + self.lock().prebuilt.insert(id.to_owned(), snapshot); + } + + pub(crate) fn take_prebuilt(&self, id: &str) -> Option<Snapshot> { + self.lock().prebuilt.remove(id) + } + + /// When this node first saw the remake committed (now, on the first call). + pub(crate) fn first_seen(&self, id: &str) -> Instant { + *self + .lock() + .first_seen + .entry(id.to_owned()) + .or_insert_with(Instant::now) + } + + pub(crate) fn status(&self, id: &str) -> Option<LocalStatus> { + self.lock().status.get(id).cloned() + } + + pub(crate) fn set_status(&self, id: &str, status: LocalStatus) { + self.lock().status.insert(id.to_owned(), status); + } + + /// Keeps the previous epoch's document, then drops what has outlived its retention. + pub(crate) fn archive( + &self, + epoch: Epoch, + doc: AutoCommit, + retention: Retention, + current_generation: u64, + ) { + let mut inner = self.lock(); + inner.archive.push(Archived { + epoch, + doc, + archived_at: unix_millis(), + retention, + }); + prune(&mut inner.archive, current_generation); + } + + pub(crate) fn prune_archive(&self, current_generation: u64) { + prune(&mut self.lock().archive, current_generation); + } + + pub(crate) fn catch_up(&self) -> Option<CatchUp> { + self.lock().catch_up.clone() + } + + pub(crate) fn set_catch_up(&self, catch_up: Option<CatchUp>) { + self.lock().catch_up = catch_up; + } + + pub(crate) fn catch_up_report(&self) -> Option<CatchUpReport> { + let inner = self.lock(); + let rejoining = inner.rejoin.is_some(); + inner.catch_up.as_ref().map(|catch_up| CatchUpReport { + target: catch_up.target.to_string(), + attempts: catch_up.attempts, + rejoining, + error: catch_up.error.clone(), + }) + } + + pub(crate) fn is_rejoining(&self) -> bool { + self.lock().rejoin.is_some() + } + + /// Progress of the rejoin in progress: (target, started). + pub(crate) fn rejoin_progress(&self) -> Option<(Epoch, Instant)> { + self.lock() + .rejoin + .as_ref() + .map(|rejoin| (rejoin.target, rejoin.started)) + } + + pub(crate) fn start_rejoin(&self, rejoin: Rejoin) { + self.lock().rejoin = Some(rejoin); + } + + pub(crate) fn take_rejoin(&self) -> Option<Rejoin> { + self.lock().rejoin.take() + } + + pub(crate) fn add_quarantine(&self, quarantined: Quarantined) { + self.lock().quarantine.push(quarantined); + } + + /// Runs `f` on the quarantined entry with this ID. + pub(crate) fn with_quarantined<T>( + &self, + id: &str, + f: impl FnOnce(&Quarantined) -> T, + ) -> Option<T> { + self.lock() + .quarantine + .iter() + .find(|entry| entry.id == id) + .map(f) + } + + pub(crate) fn list_quarantine<T>(&self, f: impl Fn(&Quarantined) -> T) -> Vec<T> { + self.lock().quarantine.iter().map(f).collect() + } + + pub(crate) fn take_quarantined(&self, id: &str) -> Option<Quarantined> { + let mut inner = self.lock(); + let index = inner.quarantine.iter().position(|entry| entry.id == id)?; + Some(inner.quarantine.remove(index)) + } + + /// Puts an entry back, e.g. after a failed replay. + pub(crate) fn restore_quarantined(&self, quarantined: Quarantined) { + self.lock().quarantine.push(quarantined); + } + + pub(crate) fn archived_epochs(&self) -> Vec<ArchivedEpoch> { + self.lock() + .archive + .iter() + .map(|archived| ArchivedEpoch { + generation: archived.epoch.generation, + id: archived.epoch.id_hex(), + archived_at: archived.archived_at, + }) + .collect() + } +} + +/// An archive is kept until every limit of its retention has passed. +fn prune(archive: &mut Vec<Archived>, current_generation: u64) { + let now = unix_millis(); + archive.retain(|archived| { + archived.retention.covers( + current_generation.saturating_sub(archived.epoch.generation), + now.saturating_sub(archived.archived_at), + ) + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn archive_retention() { + let remake = RemakeState::default(); + let epoch = |generation| Epoch { + generation, + id: [generation as u8; 32], + }; + + // Kept by epoch count only. + remake.archive( + epoch(0), + AutoCommit::new(), + Retention { + epochs: Some(1), + seconds: None, + }, + 1, + ); + assert_eq!(remake.archived_epochs().len(), 1); + remake.prune_archive(2); + assert!(remake.archived_epochs().is_empty()); + + // Kept by time while the epoch count has passed. + remake.archive( + epoch(1), + AutoCommit::new(), + Retention { + epochs: Some(0), + seconds: Some(3600), + }, + 5, + ); + assert_eq!(remake.archived_epochs().len(), 1); + + // No limits at all: not kept. + remake.archive( + epoch(2), + AutoCommit::new(), + Retention { + epochs: None, + seconds: None, + }, + 3, + ); + assert_eq!(remake.archived_epochs().len(), 1); + } +} diff --git a/src/remake/mod.rs b/src/remake/mod.rs new file mode 100644 index 0000000..8cca57c --- /dev/null +++ b/src/remake/mod.rs @@ -0,0 +1,158 @@ +//! Remakes: coordinated moves of the whole cluster to a new document, a new epoch. +//! Compaction is one: the new document has the old one's content without its history. +//! +//! * `snapshot`: deterministic build of the next epoch's initial document. +//! * `records`: the replicated remake records under `_continuum/remakes/<id>`. +//! * `coordinator`: proposal, settle period and decision on the node that was asked. +//! * `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. +//! * `quarantine`: inspecting, replaying or discarding quarantined changes. +//! * `delta`: replaying this node's changes since the base into the new epoch. +//! * `local`: per-node in-memory state (prebuilt snapshots, switch status, archive). +//! * `api`: `/admin/remakes` handlers. +//! +//! The rest of the node is reached exclusively through [`Host`], so this module never +//! depends on `AppState`, peers or sync internals. + +mod api; +mod catchup; +mod coordinator; +mod delta; +mod local; +mod quarantine; +mod records; +pub(crate) mod snapshot; +mod switch; + +pub(crate) use local::RemakeState; +pub(crate) use switch::run_worker; + +use std::future::Future; +use std::sync::Arc; +use std::time::Duration; + +use automerge::AutoCommit; +use axum::{ + routing::{get, post}, + Router, +}; +use tokio::sync::broadcast; + +use crate::document::replica::Epoch; +use crate::swarm::sync::ChangeSet; + +/// What remake needs from the node. Implemented by `AppState`. +/// +/// Used generically (`H: Host`), never as `dyn Host`, so the generic methods are fine. +pub(crate) trait Host: Clone + Send + Sync + 'static { + /// This node's in-memory remake state. + fn remake(&self) -> &RemakeState; + + /// This node's peer ID. + fn node_id(&self) -> String; + + /// The epoch of the local replica. + fn current_epoch(&self) -> Epoch; + + /// Lower bound and default basis for the settle period. + fn dormant_after(&self) -> Duration; + + fn is_node_ready(&self) -> impl Future<Output = bool> + Send; + + /// Known (not forgotten) peers. + fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send; + + /// Stops serving requests (back to `Joining`, keeping the cluster ID), as while a + /// rejoin syncs a fresh document. Returns whether the node was serving. + fn suspend(&self) -> impl Future<Output = bool> + Send; + + /// Serves requests again after [`Host::suspend`]. + fn resume(&self) -> impl Future<Output = ()> + Send; + + /// Serves reads from this document (read-only, marked stale) while the node is not + /// ready; `None` ends it. + fn set_stale_view(&self, view: Option<(Epoch, AutoCommit)>); + + /// Whether the document caught up with a peer since the last reset. + fn initial_sync_done(&self) -> bool; + + /// Makes the next sync count as initial again. + fn reset_initial_sync(&self); + + /// Store paths touched by each published document change. + fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>>; + + /// Reads the document on the blocking pool. + fn read_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send + where + F: FnOnce(&AutoCommit) -> T + Send + 'static, + T: Send + 'static; + + /// Mutable access to the document without publishing a change (e.g. forks). + fn update_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send + where + F: FnOnce(&mut AutoCommit) -> T + Send + 'static, + T: Send + 'static; + + /// One committed change (rolled back on error), published and replicated. + fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send + where + F: FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static, + T: Send + 'static, + E: Send + 'static; + + /// Replaces the document and its epoch atomically. Under the write lock, `f` gets + /// the live document and returns the new one. Afterwards the node resets every + /// peer's sync state and publishes the change. Returns the old document. + fn switch_epoch<E, F>( + &self, + epoch: Epoch, + f: F, + ) -> impl Future<Output = Result<AutoCommit, E>> + Send + where + F: FnOnce(&mut AutoCommit) -> Result<AutoCommit, E> + Send + 'static, + E: Send + 'static; +} + +#[derive(Clone, Debug)] +pub(crate) struct PeerSummary { + pub(crate) id: String, + /// Not dormant. + pub(crate) active: bool, + /// Handshake completed. + pub(crate) established: bool, + /// The epoch the peer last advertised. + pub(crate) epoch: Option<Epoch>, +} + +/// Admin routes. Merged into the authenticated part of the node's router. +pub(crate) fn router<H: Host>() -> Router<H> { + Router::new() + .route( + "/admin/remakes", + get(api::list::<H>) + .post(api::start::<H>) + .delete(api::abort_current::<H>), + ) + // The original endpoint, kept as a shortcut for a compacting remake. + .route("/admin/compaction", post(api::start::<H>)) + .route( + "/admin/remakes/{id}", + get(api::get::<H>).delete(api::abort::<H>), + ) + .route("/admin/epochs", get(api::epochs::<H>)) + .route("/admin/quarantine", get(quarantine::list::<H>)) + .route( + "/admin/quarantine/{id}", + get(quarantine::get::<H>).delete(quarantine::discard::<H>), + ) + .route( + "/admin/quarantine/{id}/document", + get(quarantine::document::<H>), + ) + .route( + "/admin/quarantine/{id}/replay", + post(quarantine::replay::<H>), + ) +} diff --git a/src/remake/quarantine.rs b/src/remake/quarantine.rs new file mode 100644 index 0000000..5643549 --- /dev/null +++ b/src/remake/quarantine.rs @@ -0,0 +1,160 @@ +//! `/admin/quarantine`: this node's own changes that were not replayed when it rejoined +//! a newer epoch (outside the replay window, or impossible to tell apart). +//! +//! * `GET /admin/quarantine` lists the entries. +//! * `GET /admin/quarantine/{id}` shows the writes: every changed path with the value +//! the old document's merged view holds there (`deleted` if none). +//! * `GET /admin/quarantine/{id}/document` downloads the old document +//! (`AutoCommit::save()`), for inspection with other tools. +//! * `POST /admin/quarantine/{id}/replay` writes them into the current epoch anyway (as +//! fresh writes, so they win over what is there) and removes the entry. +//! * `DELETE /admin/quarantine/{id}` discards the entry. +//! +//! Quarantine lives in memory on the node that holds it. + +use std::convert::Infallible; + +use axum::{ + extract::{Path, State}, + http::{header, StatusCode}, + response::{IntoResponse, Response}, + Json, +}; +use serde::Serialize; +use serde_json::{json, Value as JsonValue}; +use tracing::info; + +use super::api::error_response; +use super::delta; +use super::local::Quarantined; +use super::Host; +use crate::document::store; + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct Summary { + id: String, + /// The epoch the changes were made in. + epoch: String, + /// The epoch the node rejoined. + target: String, + created_at: u64, + reason: String, + paths: usize, +} + +fn summary(entry: &Quarantined) -> Summary { + Summary { + id: entry.id.clone(), + epoch: entry.epoch.to_string(), + target: entry.target.to_string(), + created_at: entry.created_at, + reason: entry.reason.clone(), + paths: entry.paths.len(), + } +} + +fn not_found() -> Response { + error_response(StatusCode::NOT_FOUND, "no such quarantine entry", None) +} + +pub(crate) async fn list<H: Host>(State(host): State<H>) -> Response { + Json(host.remake().list_quarantine(summary)).into_response() +} + +pub(crate) async fn get<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response { + let Some((summary, document, paths)) = host.remake().with_quarantined(&id, |entry| { + (summary(entry), entry.document.clone(), entry.paths.clone()) + }) else { + return not_found(); + }; + + let writes = tokio::task::spawn_blocking(move || { + paths + .into_iter() + .map(|path| match store::read_path(&document, &path) { + Ok(Some(value)) => json!({ "path": path, "value": value }), + Ok(None) => json!({ "path": path, "deleted": true }), + Err(error) => json!({ "path": path, "error": error.to_string() }), + }) + .collect::<Vec<JsonValue>>() + }) + .await; + + match writes { + Ok(writes) => Json(json!({ "entry": summary, "writes": writes })).into_response(), + Err(_) => error_response( + StatusCode::INTERNAL_SERVER_ERROR, + "reading the writes failed", + None, + ), + } +} + +pub(crate) async fn document<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response { + let Some(mut document) = host + .remake() + .with_quarantined(&id, |entry| entry.document.clone()) + else { + return not_found(); + }; + + match tokio::task::spawn_blocking(move || document.save()).await { + Ok(bytes) => ( + [ + (header::CONTENT_TYPE, "application/octet-stream".to_owned()), + ( + header::CONTENT_DISPOSITION, + format!("attachment; filename=\"quarantine-{id}.automerge\""), + ), + ], + bytes, + ) + .into_response(), + Err(_) => error_response( + StatusCode::INTERNAL_SERVER_ERROR, + "saving the document failed", + None, + ), + } +} + +pub(crate) async fn replay<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response { + if !host.is_node_ready().await { + return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None); + } + let Some(entry) = host.remake().take_quarantined(&id) else { + return not_found(); + }; + + let (stats, entry) = host + .commit_doc(move |doc| { + let stats = delta::replay_paths(&entry.document, doc, &entry.paths); + Ok::<_, Infallible>((stats, entry)) + }) + .await + .unwrap_or_else(|never| match never {}); + + if stats.failed > 0 && stats.replayed == 0 { + // Nothing went in; keep the entry for another look. + host.remake().restore_quarantined(entry); + return error_response( + StatusCode::CONFLICT, + "none of the writes could be replayed", + Some(json!({ "failed": stats.failed })), + ); + } + + info!(quarantine = %id, replayed = stats.replayed, failed = stats.failed, "Replayed quarantined changes"); + Json(json!({ "replayed": stats.replayed, "failed": stats.failed })).into_response() +} + +pub(crate) async fn discard<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response { + match host.remake().take_quarantined(&id) { + Some(_) => { + info!(quarantine = %id, "Discarded quarantined changes"); + StatusCode::NO_CONTENT.into_response() + } + None => not_found(), + } +} diff --git a/src/remake/records.rs b/src/remake/records.rs new file mode 100644 index 0000000..219fc9b --- /dev/null +++ b/src/remake/records.rs @@ -0,0 +1,428 @@ +//! Remake records: `_continuum/remakes/<id>` in the replicated document. +//! +//! A record has up to three parts, each with a single kind of writer, so concurrent +//! writes never fight over the same key: +//! +//! * `proposal`: written once, by the coordinator. +//! * `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. +//! +//! A proposal whose coordinator never decides (crash, partition) is reported as +//! `expired` once its settle period plus a grace margin has passed. Expiry is derived, +//! never written, so a late but valid commit can't collide with someone's abort. +//! +//! Records edited into an unreadable shape through the store API are skipped with a +//! warning. + +use automerge::{AutoCommit, ChangeHash}; +use serde::{Deserialize, Serialize}; +use tracing::warn; + +use super::snapshot::{SnapshotRequest, META_ROOT, REMAKES}; +use crate::document::replica::Epoch; +use crate::document::store::{self, StoreError}; + +/// One day, for retention defaults. +const DAY_SECS: u64 = 24 * 60 * 60; + +pub(crate) fn unix_millis() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +// --- record types ------------------------------------------------------------ + +/// An epoch in a record: generation plus hex-encoded snapshot hash. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct EpochRecord { + pub(crate) generation: u64, + pub(crate) id: String, +} + +impl From<Epoch> for EpochRecord { + fn from(epoch: Epoch) -> Self { + Self { + generation: epoch.generation, + id: epoch.id_hex(), + } + } +} + +impl EpochRecord { + pub(crate) fn to_epoch(&self) -> Option<Epoch> { + Some(Epoch { + generation: self.generation, + id: decode_hex32(&self.id)?, + }) + } +} + +/// How long something is kept. A value is kept until every given limit has passed. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Retention { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) epochs: Option<u64>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) seconds: Option<u64>, +} + +impl Retention { + /// True while at least one given limit has not passed: `epochs` epochs and + /// `age_ms` milliseconds after the fact. + pub(crate) fn covers(&self, epochs: u64, age_ms: u64) -> bool { + self.epochs.is_some_and(|limit| epochs <= limit) + || self + .seconds + .is_some_and(|limit| age_ms <= limit.saturating_mul(1000)) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Policy { + /// How stale a returning node's changes may be and still be replayed automatically; + /// beyond it they are quarantined. + #[serde(default = "Policy::default_replay_window")] + pub(crate) replay_window: Retention, + /// How long nodes keep the old epoch's document. + #[serde(default = "Policy::default_archive_retention")] + pub(crate) archive_retention: Retention, +} + +impl Policy { + pub(crate) fn default_replay_window() -> Retention { + Retention { + epochs: Some(1), + seconds: Some(7 * DAY_SECS), + } + } + + pub(crate) fn default_archive_retention() -> Retention { + Retention { + epochs: Some(2), + seconds: Some(30 * DAY_SECS), + } + } +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +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>, + pub(crate) policy: Policy, + pub(crate) settle_ms: u64, + pub(crate) proposed_at: u64, + pub(crate) settle_until: u64, + /// Peers that must stay active during the settle period. + pub(crate) peers_considered: Vec<String>, +} + +impl Proposal { + pub(crate) fn base_change_hashes(&self) -> Option<Vec<ChangeHash>> { + self.base_heads + .iter() + .map(|hex| decode_hex32(hex).map(ChangeHash)) + .collect() + } +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(tag = "outcome", rename_all = "camelCase")] +pub(crate) enum Decision { + #[serde(rename_all = "camelCase")] + Committed { + to_epoch: EpochRecord, + snapshot_bytes: u64, + decided_at: u64, + }, + #[serde(rename_all = "camelCase")] + Aborted { reason: String, decided_at: u64 }, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct AbortRequest { + pub(crate) by: String, + pub(crate) at: u64, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Record { + pub(crate) proposal: Proposal, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) decision: Option<Decision>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) abort_requested: Option<AbortRequest>, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) enum Phase { + /// Settling; the coordinator has not decided yet. + Proposed, + /// No decision long after the settle period: the coordinator is gone. + Expired, + Committed, + Aborted, +} + +impl Phase { + /// Still able to become (or already) the next epoch. + pub(crate) fn is_open(self) -> bool { + matches!(self, Phase::Proposed | Phase::Committed) + } + + pub(crate) fn is_final(self) -> bool { + !matches!(self, Phase::Proposed) + } +} + +impl Record { + /// The epoch this remake committed to, if committed and readable. + pub(crate) fn committed_epoch(&self) -> Option<Epoch> { + match &self.decision { + Some(Decision::Committed { to_epoch, .. }) => to_epoch.to_epoch(), + _ => None, + } + } + + /// Everything needed to (re)build this remake's snapshot; None if the proposal + /// is unreadable. + pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> { + Some(SnapshotRequest { + remake_id: id.to_owned(), + previous: self.proposal.from_epoch.to_epoch()?, + base_heads: self.proposal.base_change_hashes()?, + policy: serde_json::to_value(&self.proposal.policy).ok()?, + proposed_at: self.proposal.proposed_at, + }) + } + + pub(crate) fn phase(&self, now_ms: u64, grace_ms: u64) -> Phase { + match &self.decision { + Some(Decision::Committed { .. }) => Phase::Committed, + Some(Decision::Aborted { .. }) => Phase::Aborted, + None if now_ms > self.proposal.settle_until.saturating_add(grace_ms) => Phase::Expired, + None => Phase::Proposed, + } + } +} + +// --- document access --------------------------------------------------------- + +fn records_dir() -> Vec<String> { + vec![META_ROOT.to_owned(), REMAKES.to_owned()] +} + +pub(crate) fn record_path(id: &str, part: &str) -> Vec<String> { + let mut path = records_dir(); + path.push(id.to_owned()); + if !part.is_empty() { + path.push(part.to_owned()); + } + path +} + +/// The record with this ID, if it exists and is readable. +pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> { + let value = match store::read_path(doc, &record_path(id, "")) { + Ok(Some(value)) => value, + Ok(None) => return None, + Err(error) => { + warn!(remake = %id, error = %error, "Failed to read remake record"); + return None; + } + }; + match serde_json::from_value(value) { + Ok(record) => Some(record), + Err(error) => { + warn!(remake = %id, error = %error, "Skipping unreadable remake record"); + None + } + } +} + +/// All readable records. +pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> { + store::keys_at(doc, &records_dir()) + .unwrap_or_default() + .into_iter() + .filter_map(|id| load(doc, &id).map(|record| (id, record))) + .collect() +} + +fn write_part<T: Serialize>( + doc: &mut AutoCommit, + id: &str, + part: &str, + value: &T, +) -> Result<(), StoreError> { + let json = serde_json::to_value(value).map_err(anyhow::Error::from)?; + store::write_path(doc, &record_path(id, part), &json) +} + +pub(crate) fn write_proposal( + doc: &mut AutoCommit, + id: &str, + proposal: &Proposal, +) -> Result<(), StoreError> { + write_part(doc, id, "proposal", proposal) +} + +pub(crate) fn write_decision( + doc: &mut AutoCommit, + id: &str, + decision: &Decision, +) -> Result<(), StoreError> { + write_part(doc, id, "decision", decision) +} + +pub(crate) fn write_abort_request( + doc: &mut AutoCommit, + id: &str, + request: &AbortRequest, +) -> Result<(), StoreError> { + write_part(doc, id, "abortRequested", request) +} + +fn decode_hex32(hex: &str) -> Option<[u8; 32]> { + if hex.len() != 64 || !hex.is_ascii() { + return None; + } + let mut out = [0u8; 32]; + for (i, byte) in out.iter_mut().enumerate() { + *byte = u8::from_str_radix(&hex[2 * i..2 * i + 2], 16).ok()?; + } + Some(out) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn retention_covers_until_every_limit_passed() { + let both = Retention { + epochs: Some(1), + seconds: Some(10), + }; + assert!(both.covers(1, 60_000)); // epochs still within + assert!(both.covers(5, 5_000)); // time still within + assert!(!both.covers(2, 10_001)); + let none = Retention { + epochs: None, + seconds: None, + }; + assert!(!none.covers(0, 0)); + } + + fn proposal(settle_until: u64) -> Proposal { + Proposal { + coordinator: "node-a".into(), + from_epoch: Epoch::GENESIS.into(), + base_heads: vec!["ab".repeat(32)], + policy: Policy { + replay_window: Policy::default_replay_window(), + archive_retention: Policy::default_archive_retention(), + }, + settle_ms: 1_000, + proposed_at: settle_until.saturating_sub(1_000), + settle_until, + peers_considered: vec!["node-b".into()], + } + } + + #[test] + fn round_trip_through_the_document() { + let mut doc = AutoCommit::new(); + let p = proposal(10_000); + write_proposal(&mut doc, "c-1", &p).unwrap(); + write_abort_request( + &mut doc, + "c-1", + &AbortRequest { + by: "x".into(), + at: 5, + }, + ) + .unwrap(); + let decision = Decision::Committed { + to_epoch: EpochRecord { + generation: 1, + id: "cd".repeat(32), + }, + snapshot_bytes: 123, + decided_at: 11_000, + }; + write_decision(&mut doc, "c-1", &decision).unwrap(); + doc.commit(); + + let all = load_all(&doc); + assert_eq!(all.len(), 1); + 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!(record.abort_requested.is_some()); + assert_eq!(record.phase(0, 0), Phase::Committed); + } + + #[test] + fn phases() { + let mut record = Record { + proposal: proposal(10_000), + decision: None, + abort_requested: None, + }; + assert_eq!(record.phase(10_000, 500), Phase::Proposed); + assert_eq!(record.phase(10_500, 500), Phase::Proposed); + assert_eq!(record.phase(10_501, 500), Phase::Expired); + record.decision = Some(Decision::Aborted { + reason: "test".into(), + decided_at: 1, + }); + assert_eq!(record.phase(0, 0), Phase::Aborted); + } + + #[test] + fn unreadable_records_are_skipped() { + let mut doc = AutoCommit::new(); + store::write_path( + &mut doc, + &record_path("broken", ""), + &serde_json::json!({ "proposal": "not an object" }), + ) + .unwrap(); + write_proposal(&mut doc, "fine", &proposal(1)).unwrap(); + doc.commit(); + let all = load_all(&doc); + assert_eq!(all.len(), 1); + assert_eq!(all[0].0, "fine"); + } + + #[test] + fn epoch_record_round_trip() { + let epoch = Epoch { + generation: 3, + id: [0x5a; 32], + }; + assert_eq!(EpochRecord::from(epoch).to_epoch(), Some(epoch)); + assert_eq!( + EpochRecord { + generation: 1, + id: "zz".into() + } + .to_epoch(), + None + ); + } +} diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs new file mode 100644 index 0000000..da23fdf --- /dev/null +++ b/src/remake/snapshot.rs @@ -0,0 +1,637 @@ +//! Deterministic snapshot of the document at agreed base heads. +//! +//! Every node builds the snapshot of a committed remake on its own. For that to +//! work, the snapshot must be byte-identical everywhere: same operations, same order, +//! same actor, same timestamp. Its change hash then identifies the new epoch, and a +//! node whose build produces a different hash knows it must fetch the snapshot instead. +//! +//! Determinism rests on: +//! * the content: the merged view at the base heads (`store::materialize_into`), which +//! depends only on the base, not on how a node's history got there; +//! * a fixed actor derived from the remake ID and generation; +//! * commit time 0 and no message; +//! * canonical (sorted) JSON for the metadata; +//! * the same Automerge version on every node (encodings may differ across versions). +//! +//! The snapshot carries its own metadata under `_continuum/epochs/<generation>`: +//! previous epoch, base heads, per-actor clock and retention policy. Later steps read it +//! at the snapshot change itself, so edits made afterwards through the store API can't +//! change what the protocol sees. + +// `Snapshot::clock` is kept for returning nodes (step 7); late nodes read the clock from +// the metadata instead. +#![cfg_attr(not(test), allow(dead_code))] + +use std::collections::BTreeMap; + +use automerge::{transaction::CommitOptions, ActorId, AutoCommit, ChangeHash}; +use serde_json::{json, Map as JsonMap, Value as JsonValue}; +use sha2::{Digest, Sha256}; + +use super::records::{EpochRecord, Policy}; +use crate::document::replica::Epoch; +use crate::document::store::{self, StoreError}; + +pub(crate) const META_ROOT: &str = "_continuum"; +pub(crate) const REMAKES: &str = "remakes"; +pub(crate) const EPOCHS: &str = "epochs"; + +/// Everything that determines a snapshot besides the base content. +pub(crate) struct SnapshotParams<'a> { + /// ID of the committed remake; seeds the snapshot actor. + pub(crate) remake_id: &'a str, + /// The epoch being compacted. The snapshot starts generation `previous + 1`. + pub(crate) previous: Epoch, + /// Heads of the base. Every node must hold all of these changes. + pub(crate) base_heads: &'a [ChangeHash], + /// Retention policy from the remake request; stored verbatim (canonicalized). + pub(crate) policy: &'a JsonValue, + /// When the remake was proposed (unix ms); recorded for replay windows. + pub(crate) proposed_at: u64, +} + +pub(crate) struct Snapshot { + /// The new epoch: next generation, ID = hash of the snapshot change. + pub(crate) epoch: Epoch, + /// A document consisting of exactly that one change. + pub(crate) doc: AutoCommit, + /// Highest sequence number per actor (hex) contained in the base. A change + /// (actor, seq) is part of the snapshot iff seq <= clock[actor]. + pub(crate) clock: BTreeMap<String, u64>, +} + +#[derive(Debug, thiserror::Error)] +pub(crate) enum SnapshotError { + #[error("base head {0} is not in the local document")] + MissingBase(ChangeHash), + #[error("snapshot produced {0} heads instead of exactly one")] + UnexpectedHeads(usize), + #[error(transparent)] + Doc(#[from] automerge::AutomergeError), + #[error(transparent)] + Store(#[from] StoreError), + #[error("snapshot task failed: {0}")] + Task(String), +} + +/// Owned version of [`SnapshotParams`], for moving into a blocking task. +#[derive(Clone, Debug)] +pub(crate) struct SnapshotRequest { + pub(crate) remake_id: String, + pub(crate) previous: Epoch, + pub(crate) base_heads: Vec<ChangeHash>, + pub(crate) policy: JsonValue, + pub(crate) proposed_at: u64, +} + +impl SnapshotRequest { + pub(crate) fn params(&self) -> SnapshotParams<'_> { + SnapshotParams { + remake_id: &self.remake_id, + previous: self.previous, + base_heads: &self.base_heads, + policy: &self.policy, + proposed_at: self.proposed_at, + } + } +} + +/// Builds the snapshot from `copy`, a clone of the live document, on the blocking pool. +/// No lock is held while building; the caller only needs a lock for the clone. +/// Returns the copy as well, for callers that continue working with it. +pub(crate) async fn build_detached( + mut copy: AutoCommit, + request: SnapshotRequest, +) -> Result<(Snapshot, AutoCommit), SnapshotError> { + tokio::task::spawn_blocking(move || { + let snapshot = build(&mut copy, &request.params())?; + Ok((snapshot, copy)) + }) + .await + .map_err(|error| SnapshotError::Task(error.to_string()))? +} + +/// The protocol's view of an epoch, as recorded in its snapshot. +#[derive(Clone, Debug)] +pub(crate) struct EpochMetadata { + pub(crate) remake_id: String, + pub(crate) previous: Epoch, + /// Highest sequence number per actor of `previous` contained in the snapshot. + pub(crate) clock: BTreeMap<String, u64>, + pub(crate) policy: Policy, + /// When the remake was proposed (unix ms): the point the cluster moved on. + pub(crate) proposed_at: u64, +} + +#[derive(serde::Deserialize)] +#[serde(rename_all = "camelCase")] +struct MetadataRecord { + generation: u64, + remake_id: String, + previous: EpochRecord, + clock: BTreeMap<String, u64>, + policy: Policy, + #[serde(default)] + proposed_at: u64, +} + +fn parse_metadata(doc: &AutoCommit, generation: u64) -> Result<EpochMetadata, String> { + let value = store::read_path(doc, &epoch_metadata_path(generation)) + .map_err(|error| error.to_string())? + .ok_or_else(|| format!("no epoch metadata for generation {generation}"))?; + let record: MetadataRecord = serde_json::from_value(value) + .map_err(|error| format!("invalid epoch metadata: {error}"))?; + if record.generation != generation { + return Err(format!( + "epoch metadata names generation {} instead of {generation}", + record.generation + )); + } + let previous = record + .previous + .to_epoch() + .ok_or_else(|| "invalid previous epoch in metadata".to_owned())?; + + Ok(EpochMetadata { + remake_id: record.remake_id, + previous, + clock: record.clock, + policy: record.policy, + proposed_at: record.proposed_at, + }) +} + +/// Checks that `doc` is a document of `epoch`: its history starts with exactly the +/// snapshot change `epoch.id` (no dependencies), and the metadata at that change names +/// the same generation. Returns the metadata. +/// +/// This does not prove the snapshot's content matches the old epoch; the hash commits +/// to it, and the peer is a cluster member trusted as much as for ordinary sync. +pub(crate) fn verify_document(doc: &mut AutoCommit, epoch: Epoch) -> Result<EpochMetadata, String> { + let root = ChangeHash(epoch.id); + let Some(change) = doc.get_change_by_hash(&root) else { + return Err(format!( + "document does not contain the snapshot change of {epoch}" + )); + }; + if !change.deps().is_empty() { + return Err(format!("snapshot change of {epoch} has dependencies")); + } + + let at_snapshot = doc.fork_at(&[root]).map_err(|error| error.to_string())?; + parse_metadata(&at_snapshot, epoch.generation) +} + +/// 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. +pub(crate) fn read_epoch_metadata( + doc: &AutoCommit, + generation: u64, +) -> Result<EpochMetadata, String> { + parse_metadata(doc, generation) +} + +/// Path of the metadata record for an epoch generation. +pub(crate) fn epoch_metadata_path(generation: u64) -> Vec<String> { + vec![ + META_ROOT.to_owned(), + EPOCHS.to_owned(), + generation.to_string(), + ] +} + +/// Builds the snapshot of `source` at `params.base_heads`. +pub(crate) fn build( + source: &mut AutoCommit, + params: &SnapshotParams<'_>, +) -> Result<Snapshot, SnapshotError> { + // fork_at with unknown heads would quietly build something else. + for head in params.base_heads { + if source.get_change_by_hash(head).is_none() { + return Err(SnapshotError::MissingBase(*head)); + } + } + + let mut base = source.fork_at(params.base_heads)?; + let clock = clock_of(&mut base); + let generation = params.previous.generation + 1; + + let mut snapshot = AutoCommit::new().with_actor(snapshot_actor(params.remake_id, generation)); + + // Remake records describe the old epoch's protocol and stay behind. + store::materialize_into(&base, &mut snapshot, &is_remake_record)?; + + let mut base_heads: Vec<String> = params.base_heads.iter().map(|h| h.to_string()).collect(); + base_heads.sort(); + + let metadata = canonical(json!({ + "generation": generation, + "remakeId": params.remake_id, + "previous": { + "generation": params.previous.generation, + "id": params.previous.id_hex(), + }, + "baseHeads": base_heads, + "clock": clock, + "policy": params.policy, + "proposedAt": params.proposed_at, + })); + store::write_path(&mut snapshot, &epoch_metadata_path(generation), &metadata)?; + + let _ = snapshot.commit_with(CommitOptions::default().with_time(0)); + + let heads = snapshot.get_heads(); + let [id] = heads.as_slice() else { + return Err(SnapshotError::UnexpectedHeads(heads.len())); + }; + + Ok(Snapshot { + epoch: Epoch { + generation, + id: id.0, + }, + doc: snapshot, + clock, + }) +} + +/// Highest sequence number per actor in `doc`'s history. +fn clock_of(doc: &mut AutoCommit) -> BTreeMap<String, u64> { + let mut clock = BTreeMap::new(); + for change in doc.get_changes(&[]) { + let seq = clock.entry(change.actor_id().to_string()).or_insert(0); + *seq = (*seq).max(change.seq()); + } + clock +} + +/// The same 16-byte actor on every node, unique per remake and generation. +fn snapshot_actor(remake_id: &str, generation: u64) -> ActorId { + let digest = Sha256::new() + .chain_update(b"continuum.snapshot\0") + .chain_update(remake_id.as_bytes()) + .chain_update(b"\0") + .chain_update(generation.to_be_bytes()) + .finalize(); + ActorId::from(&digest[..16]) +} + +fn is_remake_record(path: &[String]) -> bool { + matches!(path, [root, records] if root == META_ROOT && records == REMAKES) +} + +/// Rebuilds every object with sorted keys, so the result is the same whichever key +/// order serde_json happens to use (it keeps insertion order with `preserve_order`). +fn canonical(value: JsonValue) -> JsonValue { + match value { + JsonValue::Object(map) => { + let sorted: BTreeMap<String, JsonValue> = map + .into_iter() + .map(|(key, value)| (key, canonical(value))) + .collect(); + JsonValue::Object(sorted.into_iter().collect::<JsonMap<_, _>>()) + } + JsonValue::Array(items) => JsonValue::Array(items.into_iter().map(canonical).collect()), + other => other, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, Value, ROOT}; + + fn doc(actor: &[u8]) -> AutoCommit { + AutoCommit::new().with_actor(ActorId::from(actor)) + } + + fn params<'a>(heads: &'a [ChangeHash], policy: &'a JsonValue) -> SnapshotParams<'a> { + SnapshotParams { + remake_id: "c-1", + previous: Epoch::GENESIS, + base_heads: heads, + policy, + proposed_at: 0, + } + } + + fn snapshot_of(source: &mut AutoCommit) -> Snapshot { + let heads = source.get_heads(); + let policy = json!({}); + build(source, ¶ms(&heads, &policy)).expect("snapshot") + } + + /// The user-visible view, without our own metadata. + fn user_view(doc: &AutoCommit) -> JsonValue { + let mut view = store::read_path(doc, &[]).unwrap().unwrap(); + view.as_object_mut().unwrap().remove(META_ROOT); + view + } + + #[test] + fn same_view_different_history_same_snapshot() { + // One history: overwrites, increments, one-shot text. + let mut a = doc(b"actor-a"); + a.put(ROOT, "x", 1_i64).unwrap(); + a.commit(); + a.put(ROOT, "x", 2_i64).unwrap(); + a.put(ROOT, "c", ScalarValue::counter(0)).unwrap(); + a.increment(ROOT, "c", 5).unwrap(); + let text = a.put_object(ROOT, "t", ObjType::Text).unwrap(); + a.splice_text(&text, 0, 0, "hello").unwrap(); + let m = a.put_object(ROOT, "m", ObjType::Map).unwrap(); + a.put(&m, "k", "v").unwrap(); + a.commit(); + + // Another actor, other order, other steps, same result. + let mut b = doc(b"actor-b"); + let m = b.put_object(ROOT, "m", ObjType::Map).unwrap(); + b.put(&m, "k", "v").unwrap(); + let text = b.put_object(ROOT, "t", ObjType::Text).unwrap(); + b.splice_text(&text, 0, 0, "hel").unwrap(); + b.commit(); + b.splice_text(&text, 3, 0, "lo").unwrap(); + b.put(ROOT, "c", ScalarValue::counter(5)).unwrap(); + b.put(ROOT, "x", 2_i64).unwrap(); + b.commit(); + + assert_eq!(user_view(&a), user_view(&b)); + + let snap_a = snapshot_of(&mut a); + let snap_b = snapshot_of(&mut b); + + // Different histories mean different clocks, and the clock is part of the + // metadata. Equal views alone therefore give equal content but not equal IDs; + // equal bases do give equal IDs (see `rebuild_is_deterministic`). + assert_eq!(user_view(&snap_a.doc), user_view(&snap_b.doc)); + assert_eq!(user_view(&snap_a.doc), user_view(&a)); + } + + #[test] + fn rebuild_is_deterministic() { + let mut a = doc(b"actor-a"); + a.put(ROOT, "x", 1_i64).unwrap(); + let list = a.put_object(ROOT, "l", ObjType::List).unwrap(); + a.insert(&list, 0, "first").unwrap(); + let inner = a.insert_object(&list, 1, ObjType::Map).unwrap(); + a.put(&inner, "k", true).unwrap(); + a.commit(); + + // A second node holding the same changes, received in a different way. + let mut b = doc(b"actor-b"); + b.merge(&mut a.clone()).unwrap(); + + let first = snapshot_of(&mut a); + let again = snapshot_of(&mut a); + let elsewhere = snapshot_of(&mut b); + + assert_eq!(first.epoch, again.epoch); + assert_eq!(first.epoch, elsewhere.epoch); + assert_eq!(first.epoch.generation, 1); + } + + #[test] + fn conflicts_collapse_into_the_merged_view() { + let mut a = doc(b"actor-a"); + let cfg = a.put_object(ROOT, "cfg", ObjType::Map).unwrap(); + a.put(&cfg, "x", 1_i64).unwrap(); + a.commit(); + + let mut b = doc(b"actor-b"); + let cfg = b.put_object(ROOT, "cfg", ObjType::Map).unwrap(); + b.put(&cfg, "y", 2_i64).unwrap(); + b.commit(); + + a.merge(&mut b).unwrap(); + let (_, conflicts) = store::read_path_with_conflicts(&a, &[]).unwrap().unwrap(); + assert!( + !conflicts.is_empty(), + "test setup should produce a conflict" + ); + + let snap = snapshot_of(&mut a); + assert_eq!(user_view(&snap.doc), json!({ "cfg": { "x": 1, "y": 2 } })); + + let (_, conflicts) = store::read_path_with_conflicts(&snap.doc, &[]) + .unwrap() + .unwrap(); + assert!( + conflicts.is_empty(), + "snapshot still has conflicts: {conflicts:?}" + ); + } + + #[test] + fn base_heads_bound_content_and_clock() { + let mut a = doc(b"actor-a"); + for i in 0..3_i64 { + a.put(ROOT, format!("a{i}"), i).unwrap(); + a.commit(); + } + + let mut b = doc(b"actor-b"); + for i in 0..2_i64 { + b.put(ROOT, format!("b{i}"), i).unwrap(); + b.commit(); + } + a.merge(&mut b).unwrap(); + + let base = a.get_heads(); + + // After the base: must not appear in the snapshot or its clock. + a.put(ROOT, "late", true).unwrap(); + a.commit(); + + let policy = json!({}); + let snap = build(&mut a, ¶ms(&base, &policy)).unwrap(); + + let view = user_view(&snap.doc); + assert!(view.get("late").is_none()); + assert_eq!(view["a2"], json!(2)); + assert_eq!(view["b1"], json!(1)); + + let actor_a = ActorId::from(b"actor-a".as_slice()).to_string(); + let actor_b = ActorId::from(b"actor-b".as_slice()).to_string(); + assert_eq!(snap.clock.get(&actor_a), Some(&3)); + assert_eq!(snap.clock.get(&actor_b), Some(&2)); + } + + #[test] + fn metadata_written_and_remake_records_dropped() { + let mut a = doc(b"actor-a"); + store::write_path( + &mut a, + &[META_ROOT.into(), REMAKES.into(), "c-1".into()], + &json!({ "state": "committed" }), + ) + .unwrap(); + store::write_path( + &mut a, + &[META_ROOT.into(), "forgotten".into(), "pk".into()], + &json!({ "at": 1 }), + ) + .unwrap(); + a.commit(); + + let heads = a.get_heads(); + let policy = json!({ "replayWindow": { "epochs": 1 } }); + let snap = build(&mut a, ¶ms(&heads, &policy)).unwrap(); + + let meta = store::read_path(&snap.doc, &[META_ROOT.into()]) + .unwrap() + .unwrap(); + assert!(meta.get(REMAKES).is_none()); + assert_eq!(meta["forgotten"]["pk"]["at"], json!(1)); + + let epoch = &meta[EPOCHS]["1"]; + assert_eq!(epoch["generation"], json!(1)); + assert_eq!(epoch["remakeId"], json!("c-1")); + assert_eq!(epoch["previous"]["generation"], json!(0)); + assert_eq!(epoch["policy"], policy); + assert_eq!(epoch["baseHeads"].as_array().unwrap().len(), heads.len()); + } + + #[test] + fn automerge_types_survive() { + let mut a = doc(b"actor-a"); + a.put(ROOT, "c", ScalarValue::counter(3)).unwrap(); + a.increment(ROOT, "c", 4).unwrap(); + let text = a.put_object(ROOT, "t", ObjType::Text).unwrap(); + a.splice_text(&text, 0, 0, "abc").unwrap(); + a.put(ROOT, "bytes", ScalarValue::Bytes(vec![0, 1, 255])) + .unwrap(); + a.put(ROOT, "ts", ScalarValue::Timestamp(1_700_000_000_000)) + .unwrap(); + a.put(ROOT, "neg_zero", -0.0_f64).unwrap(); + a.commit(); + + let snap = snapshot_of(&mut a); + let d = &snap.doc; + + match d.get(ROOT, "c").unwrap() { + Some((Value::Scalar(s), _)) => match s.as_ref() { + ScalarValue::Counter(c) => assert_eq!(i64::from(c.clone()), 7), + other => panic!("counter became {other:?}"), + }, + other => panic!("unexpected {other:?}"), + } + + let (_, text_id) = d.get(ROOT, "t").unwrap().unwrap(); + assert_eq!(d.object_type(&text_id).unwrap(), ObjType::Text); + assert_eq!(d.text(&text_id).unwrap(), "abc"); + + match d.get(ROOT, "bytes").unwrap() { + Some((Value::Scalar(s), _)) => { + assert_eq!(s.as_ref(), &ScalarValue::Bytes(vec![0, 1, 255])) + } + other => panic!("unexpected {other:?}"), + } + match d.get(ROOT, "ts").unwrap() { + Some((Value::Scalar(s), _)) => { + assert_eq!(s.as_ref(), &ScalarValue::Timestamp(1_700_000_000_000)) + } + other => panic!("unexpected {other:?}"), + } + match d.get(ROOT, "neg_zero").unwrap() { + Some((Value::Scalar(s), _)) => match s.as_ref() { + ScalarValue::F64(f) => assert_eq!(f.to_bits(), (-0.0_f64).to_bits()), + other => panic!("float became {other:?}"), + }, + other => panic!("unexpected {other:?}"), + } + } + + #[test] + fn snapshot_is_a_single_change_with_the_epoch_id() { + let mut a = doc(b"actor-a"); + a.put(ROOT, "x", 1_i64).unwrap(); + a.commit(); + + let mut snap = snapshot_of(&mut a); + let heads = snap.doc.get_heads(); + assert_eq!(heads.len(), 1); + assert_eq!(heads[0].0, snap.epoch.id); + assert_eq!(snap.doc.get_changes(&[]).len(), 1); + } + + #[test] + fn missing_base_is_rejected() { + let mut a = doc(b"actor-a"); + a.put(ROOT, "x", 1_i64).unwrap(); + a.commit(); + + let mut other = doc(b"actor-b"); + other.put(ROOT, "y", 1_i64).unwrap(); + other.commit(); + let foreign = other.get_heads(); + + let policy = json!({}); + assert!(matches!( + build(&mut a, ¶ms(&foreign, &policy)), + Err(SnapshotError::MissingBase(_)) + )); + } + + #[test] + fn empty_document() { + let mut a = doc(b"actor-a"); + let snap = snapshot_of(&mut a); + assert_eq!(snap.epoch.generation, 1); + assert!(snap.clock.is_empty()); + assert_eq!(user_view(&snap.doc), json!({})); + } + + #[test] + fn fetched_document_is_verified_and_late_changes_replayed() { + use crate::remake::delta; + + // Old epoch: the base, then a late write by this node. + let mut late = doc(b"late-node"); + late.put(ROOT, "shared", "base").unwrap(); + late.commit(); + let base = late.get_heads(); + let policy = json!({}); + let mut snap = build(&mut late, ¶ms(&base, &policy)).unwrap(); + + // Meanwhile the cluster moved on in the new epoch. + let mut cluster = snap + .doc + .fork() + .with_actor(ActorId::from(&b"cluster-node"[..])); + cluster.put(ROOT, "new", 1_i64).unwrap(); + cluster.commit(); + let bytes = cluster.save(); + + late.put(ROOT, "mine", "late").unwrap(); + late.commit(); + + let mut fetched = AutoCommit::load(&bytes).unwrap(); + let metadata = verify_document(&mut fetched, snap.epoch).unwrap(); + assert_eq!(metadata.previous, Epoch::GENESIS); + assert_eq!(metadata.remake_id, "c-1"); + assert_eq!( + metadata.policy.replay_window, + Policy::default_replay_window() + ); + + // Documents of other epochs are refused. + let mut wrong = snap.epoch; + wrong.generation = 2; + assert!(verify_document(&mut fetched, wrong).is_err()); + let mut wrong = snap.epoch; + wrong.id = [0; 32]; + 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); + assert_eq!(missing.own, 1); + let heads = late.get_heads(); + let mut adopted = fetched.with_actor(ActorId::from(&b"late-node-next"[..])); + delta::replay(&mut late, &missing.before, &heads, &mut adopted); + assert_eq!( + user_view(&adopted), + json!({ "shared": "base", "new": 1, "mine": "late" }) + ); + } +} diff --git a/src/remake/switch.rs b/src/remake/switch.rs new file mode 100644 index 0000000..b4ab829 --- /dev/null +++ b/src/remake/switch.rs @@ -0,0 +1,248 @@ +//! Switching this node to a committed epoch. +//! +//! The worker watches the remake records. When a remake of the local epoch is +//! committed, it waits one keepalive interval (so the commit record can reach peers +//! through ordinary sync before this node stops syncing the old epoch), then: +//! +//! 1. clones the live document under the read lock; +//! 2. without any lock: takes the coordinator's prebuilt snapshot or builds it, checks +//! that its hash is the committed epoch ID, and replays the clone's changes since +//! the base into it; +//! 3. under the write lock: replays whatever arrived since the clone, swaps document and +//! epoch, and hands back the old document; +//! 4. archives the old document according to the policy. +//! +//! The write lock is held only for step 3, which is proportional to the changes made +//! during steps 1-2. +//! +//! A node whose build doesn't match the committed epoch, or that never saw the commit, +//! rejoins the new epoch from peers instead (`catchup`). + +use std::time::Duration; + +use automerge::{ActorId, AutoCommit, ChangeHash}; +use tokio::sync::broadcast; +use tracing::{error, info, warn}; + +use super::catchup; +use super::delta::{self, ReplayStats}; +use super::local::LocalStatus; +use super::records::{self, unix_millis, EpochRecord, Record}; +use super::snapshot::{self, META_ROOT, REMAKES}; +use super::Host; +use crate::document::replica::Epoch; +use crate::document::store; +use crate::swarm::sync::paths_related; + +const TICK: Duration = Duration::from_secs(1); + +pub(crate) async fn run_worker<H: Host>(host: H) { + info!("Starting remake worker"); + + let records_dir = vec![META_ROOT.to_owned(), REMAKES.to_owned()]; + let mut changes = host.subscribe_changes(); + let mut changes_closed = false; + let mut tick = tokio::time::interval(TICK); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + + loop { + // React promptly to changes of the records; the timer covers the switch delay. + let should_check = tokio::select! { + _ = tick.tick() => true, + received = changes.recv(), if !changes_closed => match received { + Ok(set) => set.paths.iter().any(|path| paths_related(path, &records_dir)), + Err(broadcast::error::RecvError::Lagged(_)) => true, + Err(broadcast::error::RecvError::Closed) => { + warn!("Change events closed; remake worker continues on its timer only"); + changes_closed = true; + false + } + } + }; + + if should_check { + check(&host).await; + } + } +} + +async fn check<H: Host>(host: &H) { + let local = host.current_epoch(); + host.remake().prune_archive(local.generation); + + // While rejoining, the document is being filled from peers: its records are not + // complete enough to act on. + if host.remake().is_rejoining() { + return catchup::check(host).await; + } + + let all = host.read_doc(records::load_all).await; + let local_record = EpochRecord::from(local); + + let committed: Vec<(String, Record)> = all + .into_iter() + .filter(|(_, record)| { + record.proposal.from_epoch == local_record && record.committed_epoch().is_some() + }) + .collect(); + + let (id, record) = match committed.as_slice() { + // Nothing committed here; maybe peers moved on without us. + [] => return catchup::check(host).await, + [(id, record)] => (id.clone(), record.clone()), + several => { + // The settle protocol prevents this; never guess which one is right. Peers + // that did switch can still be followed. + let ids: Vec<&str> = several.iter().map(|(id, _)| id.as_str()).collect(); + error!(remakes = ?ids, "Several remakes committed for this epoch; not switching on my own"); + return catchup::check(host).await; + } + }; + + if host + .remake() + .status(&id) + .is_some_and(|status| status.error.is_some()) + { + // Our own switch failed (reported in the remake's status): fetch the epoch + // from a peer that made it. + return catchup::check(host).await; + } + + let switch_delay = host.dormant_after() / 4; // one keepalive interval + if host.remake().first_seen(&id).elapsed() < switch_delay { + return; + } + + match switch(host, &id, &record).await { + Ok(stats) => { + host.remake().set_status( + &id, + LocalStatus { + switched_at: Some(unix_millis()), + replayed_paths: Some(stats.replayed), + failed_paths: Some(stats.failed), + quarantine: None, + error: None, + }, + ); + } + Err(reason) => { + error!(remake = %id, reason = %reason, "Epoch switch failed"); + host.remake().set_status( + &id, + LocalStatus { + error: Some(reason), + ..LocalStatus::default() + }, + ); + } + } +} + +async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplayStats, String> { + let (Some(expected), Some(request)) = (record.committed_epoch(), record.snapshot_request(id)) + else { + return Err("unreadable remake record".to_owned()); + }; + let previous = request.previous; + let base = request.base_heads.clone(); + let retention = record.proposal.policy.archive_retention.clone(); + // Only the coordinator carries the record into the new epoch: one writer, so the + // nodes don't all write the same path concurrently right after switching (a merge + // conflict, plus one change per node to sync). The epoch metadata in the snapshot + // keeps the essentials on every node anyway. + let carry_record = record.proposal.coordinator == host.node_id(); + let record_path = records::record_path(id, ""); + + info!(remake = %id, from = %previous, to = %expected, "Switching epoch"); + + // 1. Consistent copy, under the read lock only. + let copy = host.read_doc(|doc| doc.clone()).await; + let prebuilt = host + .remake() + .take_prebuilt(id) + .filter(|snapshot| snapshot.epoch == expected); + + // 2. Build (or reuse), verify and replay without holding any lock. + let (new_doc, copy_heads, stats) = tokio::task::spawn_blocking( + move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> { + let mut copy = copy; + let snapshot = match prebuilt { + Some(snapshot) => snapshot, + None => snapshot::build(&mut copy, &request.params()) + .map_err(|error| format!("snapshot build failed: {error}"))?, + }; + if snapshot.epoch != expected { + return Err(format!( + "built snapshot {} does not match the committed epoch {expected}", + snapshot.epoch + )); + } + + // A fresh actor: this node's writes in the new epoch. + let mut doc = snapshot.doc.with_actor(ActorId::random()); + + // Keep this remake's record visible in the new epoch, for its history. + if carry_record { + if let Err(error) = store::copy_path(©, &mut doc, &record_path) { + warn!(error = %error, "Could not carry the remake record into the new epoch"); + } + } + + let heads = copy.get_heads(); + let stats = delta::replay(&mut copy, &base, &heads, &mut doc); + Ok((doc, heads, stats)) + }, + ) + .await + .map_err(|error| format!("switch task failed: {error}"))??; + + // 3. Swap, replaying what arrived since the copy. + let old = swap_in(host, expected, new_doc, copy_heads).await?; + + // 4. Keep the old document as configured. + host.remake() + .archive(previous, old, retention, expected.generation); + + info!( + remake = %id, + epoch = %expected, + replayed = stats.replayed, + failed = stats.failed, + "Switched to the new epoch" + ); + Ok(stats) +} + +/// Under the write lock: replays into `new_doc` whatever the live document received +/// since `copy_heads` (the heads of the copy `new_doc` was prepared from), then swaps it +/// in as `target`. Returns the old document. +pub(super) async fn swap_in<H: Host>( + host: &H, + target: Epoch, + new_doc: AutoCommit, + copy_heads: Vec<ChangeHash>, +) -> Result<AutoCommit, String> { + host.switch_epoch(target, move |live| { + // Only one worker switches, but be sure the copy still describes the live document. + if !copy_heads + .iter() + .all(|head| live.get_change_by_hash(head).is_some()) + { + return Err("the document was replaced while preparing the switch".to_owned()); + } + let mut new_doc = new_doc; + let live_heads = live.get_heads(); + let late = delta::replay(live, ©_heads, &live_heads, &mut new_doc); + if late.replayed + late.failed > 0 { + info!( + replayed = late.replayed, + failed = late.failed, + "Replayed late changes" + ); + } + Ok(new_doc) + }) + .await +} |