From 00f3be4e8b6ebf1f4262bc34384f82bc8d438574 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 21 Oct 2026 03:56:31 +0200 Subject: rename compaction to remake --- src/api/mod.rs | 2 +- src/compaction/api.rs | 473 ------------------------------- src/compaction/catchup.rs | 506 --------------------------------- src/compaction/coordinator.rs | 183 ------------ src/compaction/delta.rs | 387 ------------------------- src/compaction/local.rs | 320 --------------------- src/compaction/mod.rs | 154 ---------- src/compaction/quarantine.rs | 160 ----------- src/compaction/records.rs | 428 ---------------------------- src/compaction/snapshot.rs | 638 ------------------------------------------ src/compaction/switch.rs | 248 ---------------- src/document/replica.rs | 10 +- src/main.rs | 8 +- src/remake/api.rs | 474 +++++++++++++++++++++++++++++++ src/remake/catchup.rs | 506 +++++++++++++++++++++++++++++++++ src/remake/coordinator.rs | 183 ++++++++++++ src/remake/delta.rs | 387 +++++++++++++++++++++++++ src/remake/local.rs | 320 +++++++++++++++++++++ src/remake/mod.rs | 158 +++++++++++ src/remake/quarantine.rs | 160 +++++++++++ src/remake/records.rs | 428 ++++++++++++++++++++++++++++ src/remake/snapshot.rs | 637 +++++++++++++++++++++++++++++++++++++++++ src/remake/switch.rs | 248 ++++++++++++++++ src/state.rs | 12 +- src/swarm/bootstrap.rs | 2 +- src/swarm/peer.rs | 2 +- 26 files changed, 3519 insertions(+), 3515 deletions(-) delete mode 100644 src/compaction/api.rs delete mode 100644 src/compaction/catchup.rs delete mode 100644 src/compaction/coordinator.rs delete mode 100644 src/compaction/delta.rs delete mode 100644 src/compaction/local.rs delete mode 100644 src/compaction/mod.rs delete mode 100644 src/compaction/quarantine.rs delete mode 100644 src/compaction/records.rs delete mode 100644 src/compaction/snapshot.rs delete mode 100644 src/compaction/switch.rs create mode 100644 src/remake/api.rs create mode 100644 src/remake/catchup.rs create mode 100644 src/remake/coordinator.rs create mode 100644 src/remake/delta.rs create mode 100644 src/remake/local.rs create mode 100644 src/remake/mod.rs create mode 100644 src/remake/quarantine.rs create mode 100644 src/remake/records.rs create mode 100644 src/remake/snapshot.rs create mode 100644 src/remake/switch.rs (limited to 'src') diff --git a/src/api/mod.rs b/src/api/mod.rs index ac6f2d0..ea3ad41 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -37,7 +37,7 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) -> .route("/stream/{*key}", get(store::stream_store_entry)) .route("/peers", get(peers::list_peers)) .route("/peers/{peer}", delete(peers::delete_peer)) - .merge(crate::compaction::router::()); + .merge(crate::remake::router::()); if debug_api { warn!("Debug API enabled"); diff --git a/src/compaction/api.rs b/src/compaction/api.rs deleted file mode 100644 index 9fae7c9..0000000 --- a/src/compaction/api.rs +++ /dev/null @@ -1,473 +0,0 @@ -//! `/admin/compaction(s)` handlers. -//! -//! * `POST /admin/compaction` proposes a compaction of the current epoch. 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/compactions` lists all compactions. -//! * `GET /admin/compactions/{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/compactions/{id}` asks the coordinator to abort. Only possible while -//! proposed; `409` once committed (a switch can't be undone) or already over. -//! `DELETE /admin/compaction` does the same for the compaction 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, - archive_retention: Option, - settle_secs: Option, -} - -#[derive(Debug, Serialize)] -#[serde(rename_all = "camelCase")] -struct Report { - id: String, - phase: Phase, - #[serde(flatten)] - record: Record, - /// What happened to this compaction on the node answering the request. - #[serde(skip_serializing_if = "Option::is_none")] - local: Option, -} - -#[derive(Debug, thiserror::Error)] -enum StartError { - #[error("compaction {0} is already open for this epoch")] - Open(String), - #[error(transparent)] - Store(#[from] StoreError), -} - -pub(super) fn error_response( - status: StatusCode, - message: impl Into, - details: Option, -) -> 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(host: &H) -> u64 { - host.dormant_after().as_millis() as u64 -} - -fn report(host: &H, id: String, record: Record) -> Report { - Report { - phase: record.phase(unix_millis(), grace(host)), - local: host.compaction().status(&id), - id, - record, - } -} - -async fn load_report(host: &H, id: &str) -> Option { - 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 compaction leaves `proposed`. Bounded by the proposal's own expiry, -/// at which point it is reported as `expired`. -async fn wait_until_final(host: &H, id: &str) -> Option { - // 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!(compaction = %id, seen, misses, "Compaction 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( - State(host): State, - Query(query): Query, - 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 = peers - .iter() - .filter(|peer| !peer.active) - .map(|peer| peer.id.clone()) - .collect(); - if !dormant.is_empty() { - return error_response( - StatusCode::CONFLICT, - "dormant peers block compaction; forget them first (DELETE /peers/{pk})", - Some(json!({ "dormantPeers": dormant })), - ); - } - - let mut peers_considered: Vec = 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 compaction of this epoch is still open", - Some(json!({ "compaction": open })), - ); - } - Err(StartError::Store(error)) => { - error!(error = %format!("{error:#}"), "Failed to write compaction proposal"); - return error_response( - StatusCode::INTERNAL_SERVER_ERROR, - "failed to write the proposal", - None, - ); - } - }; - - info!( - compaction = %id, - settle_secs = settle.as_secs(), - "Compaction proposed" - ); - - if query.wait { - return match wait_until_final(&host, &id).await { - Some(report) => Json(report).into_response(), - None => error_response(StatusCode::NOT_FOUND, "compaction record vanished", None), - }; - } - - let location = format!("/admin/compactions/{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( - host: H, - id: String, - mut proposal: Proposal, - now: u64, - grace: u64, -) -> Result { - let record_id = id.clone(); - let record = host - .commit_doc(move |doc| -> Result { - 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 = 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(State(host): State) -> Response { - let all = host.read_doc(records::load_all).await; - let mut reports: Vec = 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( - State(host): State, - Path(id): Path, - Query(query): Query, -) -> 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 compaction", None), - } -} - -/// `DELETE /admin/compactions/{id}`. -pub(crate) async fn abort( - State(host): State, - Path(id): Path, - Query(query): Query, -) -> Response { - abort_by_id(host, id, query.wait).await -} - -/// `DELETE /admin/compaction`: aborts whichever compaction 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( - State(host): State, - Query(query): Query, -) -> 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 compaction of the local epoch is in progress", - None, - ), - } -} - -async fn abort_by_id(host: H, id: String, wait: bool) -> Response { - let Some(current) = load_report(&host, &id).await else { - return error_response(StatusCode::NOT_FOUND, "no such compaction", None); - }; - - match current.phase { - Phase::Proposed => {} - Phase::Committed => { - return error_response( - StatusCode::CONFLICT, - "already committed; the switch cannot be undone", - Some(json!({ "compaction": current })), - ) - } - Phase::Aborted | Phase::Expired => { - return error_response( - StatusCode::CONFLICT, - "already over", - Some(json!({ "compaction": 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!(compaction = %id, error = %format!("{error:#}"), "Failed to write abort request"); - return error_response( - StatusCode::INTERNAL_SERVER_ERROR, - "failed to record the abort request", - None, - ); - } - info!(compaction = %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 compaction", 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(State(host): State) -> Response { - let current = EpochRecord::from(host.current_epoch()); - Json(json!({ - "current": current, - "archived": host.compaction().archived_epochs(), - "catchUp": host.compaction().catch_up_report(), - "quarantine": host.compaction().list_quarantine(|entry| entry.id.clone()), - })) - .into_response() -} diff --git a/src/compaction/catchup.rs b/src/compaction/catchup.rs deleted file mode 100644 index 5085669..0000000 --- a/src/compaction/catchup.rs +++ /dev/null @@ -1,506 +0,0 @@ -//! 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(host: &H) { - if let Some((target, started)) = host.compaction().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.compaction().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.compaction().catch_up().is_some() { - host.compaction().set_catch_up(None); - } - return; - }; - - let mut targets: Vec = 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 = 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 - .compaction() - .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(host: &H, local: Epoch, target: Epoch, attempts: u32) { - host.compaction().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.compaction().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>), - Quarantine { - paths: Vec>, - 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(host: &H) { - let Some(rejoin) = host.compaction().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.compaction().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.compaction().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.compaction().archive( - previous, - document, - metadata.policy.archive_retention.clone(), - target.generation, - ); - } - - host.compaction().set_status( - &metadata.compaction_id, - LocalStatus { - switched_at: Some(unix_millis()), - replayed_paths: Some(stats.replayed), - failed_paths: Some(stats.failed), - quarantine: quarantine_id.clone(), - error: None, - }, - ); - host.compaction().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(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 - .compaction() - .catch_up() - .filter(|catch_up| catch_up.target == target) - .map_or(1, |catch_up| catch_up.attempts); - host.compaction().set_catch_up(Some(CatchUp { - target, - attempts, - last_attempt: Instant::now(), - error: Some(reason), - })); -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::compaction::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 { - compaction_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/compaction/coordinator.rs b/src/compaction/coordinator.rs deleted file mode 100644 index 8ce3928..0000000 --- a/src/compaction/coordinator.rs +++ /dev/null @@ -1,183 +0,0 @@ -//! 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(host: H, id: String) { - info!(compaction = %id, "Coordinating compaction"); - - loop { - tokio::time::sleep(TICK).await; - - match step(&host, &id).await { - Ok(true) => return, - Ok(false) => {} - Err(error) => { - warn!(compaction = %id, error = %format!("{error:#}"), "Coordinator step failed; retrying"); - } - } - } -} - -/// One evaluation. Returns true once the coordinator is done. -async fn step(host: &H, id: &str) -> Result { - 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!(compaction = %id, "Compaction 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( - host: &H, - id: &str, - record: &Record, - all: &[(String, Record)], - now: u64, - grace: u64, -) -> Option { - 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 compaction {other} for the same epoch")); - } - - // Peers forgotten in the meantime are ignored: an operator decided they are gone. - let peers: HashMap = 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(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.compaction().store_prebuilt(id, snapshot); - - info!(compaction = %id, to_epoch = %to_epoch, snapshot_bytes, "Compaction committed"); - Ok(()) -} - -async fn decide_abort(host: &H, id: &str, reason: String) -> Result<(), StoreError> { - warn!(compaction = %id, reason = %reason, "Compaction aborted"); - write_decision( - host, - id, - Decision::Aborted { - reason, - decided_at: records::unix_millis(), - }, - ) - .await -} - -async fn write_decision(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/compaction/delta.rs b/src/compaction/delta.rs deleted file mode 100644 index cf556b2..0000000 --- a/src/compaction/delta.rs +++ /dev/null @@ -1,387 +0,0 @@ -//! 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 compaction 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/compactions`, `_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::{COMPACTIONS, EPOCHS, META_ROOT}; -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], -) -> 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 == COMPACTIONS || 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, - /// 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, - own: &ActorId, -) -> Missing { - let own_hex = own.to_string(); - let changes: Vec<(ChangeHash, Vec, 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 = 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, bool)> = changes - .iter() - .filter(|(hash, _, _)| !excluded.contains(hash)) - .collect(); - let depended_on: HashSet = kept - .iter() - .flat_map(|(_, deps, _)| deps.iter().copied()) - .collect(); - let mut before: Vec = 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> { - let mut paths: Vec> = 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::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 { - 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 = [(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 = [(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 compaction record. - store::write_path( - &mut old, - &[META_ROOT.into()], - &json!({ "forgotten": { "pk": 1 } }), - ) - .unwrap(); - store::write_path( - &mut old, - &[META_ROOT.into(), COMPACTIONS.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(COMPACTIONS).is_none()); - } -} diff --git a/src/compaction/local.rs b/src/compaction/local.rs deleted file mode 100644 index ec40b06..0000000 --- a/src/compaction/local.rs +++ /dev/null @@ -1,320 +0,0 @@ -//! Per-node, in-memory compaction 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 Compaction { - // A std mutex: every access is short and never spans an await. - inner: Mutex, -} - -#[derive(Default)] -struct Inner { - /// Built by the coordinator at commit time; consumed by the local switch. - prebuilt: HashMap, - /// When this node first saw a compaction committed. - first_seen: HashMap, - status: HashMap, - archive: Vec, - /// Last attempt to catch up with a newer epoch. - catch_up: Option, - /// A rejoin in progress: the node syncs an empty document of a newer epoch. - rejoin: Option, - quarantine: Vec, -} - -/// 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, -} - -/// 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>, - /// 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, -} - -/// What happened to a compaction 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, - #[serde(skip_serializing_if = "Option::is_none")] - pub(crate) replayed_paths: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub(crate) failed_paths: Option, - /// 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, - /// Why the switch did not happen (yet). - #[serde(skip_serializing_if = "Option::is_none")] - pub(crate) error: Option, -} - -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 Compaction { - 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 { - self.lock().prebuilt.remove(id) - } - - /// When this node first saw the compaction 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 { - 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 { - self.lock().catch_up.clone() - } - - pub(crate) fn set_catch_up(&self, catch_up: Option) { - self.lock().catch_up = catch_up; - } - - pub(crate) fn catch_up_report(&self) -> Option { - 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 { - 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( - &self, - id: &str, - f: impl FnOnce(&Quarantined) -> T, - ) -> Option { - self.lock() - .quarantine - .iter() - .find(|entry| entry.id == id) - .map(f) - } - - pub(crate) fn list_quarantine(&self, f: impl Fn(&Quarantined) -> T) -> Vec { - self.lock().quarantine.iter().map(f).collect() - } - - pub(crate) fn take_quarantined(&self, id: &str) -> Option { - 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 { - 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, 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 compaction = Compaction::default(); - let epoch = |generation| Epoch { - generation, - id: [generation as u8; 32], - }; - - // Kept by epoch count only. - compaction.archive( - epoch(0), - AutoCommit::new(), - Retention { - epochs: Some(1), - seconds: None, - }, - 1, - ); - assert_eq!(compaction.archived_epochs().len(), 1); - compaction.prune_archive(2); - assert!(compaction.archived_epochs().is_empty()); - - // Kept by time while the epoch count has passed. - compaction.archive( - epoch(1), - AutoCommit::new(), - Retention { - epochs: Some(0), - seconds: Some(3600), - }, - 5, - ); - assert_eq!(compaction.archived_epochs().len(), 1); - - // No limits at all: not kept. - compaction.archive( - epoch(2), - AutoCommit::new(), - Retention { - epochs: None, - seconds: None, - }, - 3, - ); - assert_eq!(compaction.archived_epochs().len(), 1); - } -} diff --git a/src/compaction/mod.rs b/src/compaction/mod.rs deleted file mode 100644 index 41f011f..0000000 --- a/src/compaction/mod.rs +++ /dev/null @@ -1,154 +0,0 @@ -//! Compaction: replacing the document with a snapshot in a new epoch. -//! -//! * `snapshot`: deterministic snapshot build (step 3). -//! * `records`: the replicated compaction records under `_continuum/compactions/`. -//! * `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/compaction(s)` 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::Compaction; -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 compaction 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 compaction state. - fn compaction(&self) -> &Compaction; - - /// 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 + Send; - - /// Known (not forgotten) peers. - fn peer_summaries(&self) -> impl Future> + 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 + Send; - - /// Serves requests again after [`Host::suspend`]. - fn resume(&self) -> impl Future + 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>; - - /// Reads the document on the blocking pool. - fn read_doc(&self, f: F) -> impl Future + 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(&self, f: F) -> impl Future + 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(&self, f: F) -> impl Future> + Send - where - F: FnOnce(&mut AutoCommit) -> Result + 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( - &self, - epoch: Epoch, - f: F, - ) -> impl Future> + Send - where - F: FnOnce(&mut AutoCommit) -> Result + 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, -} - -/// Admin routes. Merged into the authenticated part of the node's router. -pub(crate) fn router() -> Router { - Router::new() - .route( - "/admin/compaction", - post(api::start::).delete(api::abort_current::), - ) - .route("/admin/compactions", get(api::list::)) - .route( - "/admin/compactions/{id}", - get(api::get::).delete(api::abort::), - ) - .route("/admin/epochs", get(api::epochs::)) - .route("/admin/quarantine", get(quarantine::list::)) - .route( - "/admin/quarantine/{id}", - get(quarantine::get::).delete(quarantine::discard::), - ) - .route( - "/admin/quarantine/{id}/document", - get(quarantine::document::), - ) - .route( - "/admin/quarantine/{id}/replay", - post(quarantine::replay::), - ) -} diff --git a/src/compaction/quarantine.rs b/src/compaction/quarantine.rs deleted file mode 100644 index ad3c616..0000000 --- a/src/compaction/quarantine.rs +++ /dev/null @@ -1,160 +0,0 @@ -//! `/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(State(host): State) -> Response { - Json(host.compaction().list_quarantine(summary)).into_response() -} - -pub(crate) async fn get(State(host): State, Path(id): Path) -> Response { - let Some((summary, document, paths)) = host.compaction().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::>() - }) - .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(State(host): State, Path(id): Path) -> Response { - let Some(mut document) = host - .compaction() - .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(State(host): State, Path(id): Path) -> Response { - if !host.is_node_ready().await { - return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None); - } - let Some(entry) = host.compaction().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.compaction().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(State(host): State, Path(id): Path) -> Response { - match host.compaction().take_quarantined(&id) { - Some(_) => { - info!(quarantine = %id, "Discarded quarantined changes"); - StatusCode::NO_CONTENT.into_response() - } - None => not_found(), - } -} diff --git a/src/compaction/records.rs b/src/compaction/records.rs deleted file mode 100644 index f533ee2..0000000 --- a/src/compaction/records.rs +++ /dev/null @@ -1,428 +0,0 @@ -//! Compaction records: `_continuum/compactions/` 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, COMPACTIONS, META_ROOT}; -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 for EpochRecord { - fn from(epoch: Epoch) -> Self { - Self { - generation: epoch.generation, - id: epoch.id_hex(), - } - } -} - -impl EpochRecord { - pub(crate) fn to_epoch(&self) -> Option { - 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, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub(crate) seconds: Option, -} - -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, - 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, -} - -impl Proposal { - pub(crate) fn base_change_hashes(&self) -> Option> { - 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, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub(crate) abort_requested: Option, -} - -#[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 compaction committed to, if committed and readable. - pub(crate) fn committed_epoch(&self) -> Option { - match &self.decision { - Some(Decision::Committed { to_epoch, .. }) => to_epoch.to_epoch(), - _ => None, - } - } - - /// Everything needed to (re)build this compaction's snapshot; None if the proposal - /// is unreadable. - pub(crate) fn snapshot_request(&self, id: &str) -> Option { - Some(SnapshotRequest { - compaction_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 { - vec![META_ROOT.to_owned(), COMPACTIONS.to_owned()] -} - -pub(crate) fn record_path(id: &str, part: &str) -> Vec { - 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 { - let value = match store::read_path(doc, &record_path(id, "")) { - Ok(Some(value)) => value, - Ok(None) => return None, - Err(error) => { - warn!(compaction = %id, error = %error, "Failed to read compaction record"); - return None; - } - }; - match serde_json::from_value(value) { - Ok(record) => Some(record), - Err(error) => { - warn!(compaction = %id, error = %error, "Skipping unreadable compaction 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( - 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/compaction/snapshot.rs b/src/compaction/snapshot.rs deleted file mode 100644 index 0d49d48..0000000 --- a/src/compaction/snapshot.rs +++ /dev/null @@ -1,638 +0,0 @@ -//! Deterministic snapshot of the document at agreed base heads. -//! -//! Every node builds the snapshot of a committed compaction 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 compaction 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/`: -//! 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 COMPACTIONS: &str = "compactions"; -pub(crate) const EPOCHS: &str = "epochs"; - -/// Everything that determines a snapshot besides the base content. -pub(crate) struct SnapshotParams<'a> { - /// ID of the committed compaction; seeds the snapshot actor. - pub(crate) compaction_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 compaction request; stored verbatim (canonicalized). - pub(crate) policy: &'a JsonValue, - /// When the compaction 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, -} - -#[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) compaction_id: String, - pub(crate) previous: Epoch, - pub(crate) base_heads: Vec, - pub(crate) policy: JsonValue, - pub(crate) proposed_at: u64, -} - -impl SnapshotRequest { - pub(crate) fn params(&self) -> SnapshotParams<'_> { - SnapshotParams { - compaction_id: &self.compaction_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) compaction_id: String, - pub(crate) previous: Epoch, - /// Highest sequence number per actor of `previous` contained in the snapshot. - pub(crate) clock: BTreeMap, - pub(crate) policy: Policy, - /// When the compaction 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, - compaction_id: String, - previous: EpochRecord, - clock: BTreeMap, - policy: Policy, - #[serde(default)] - proposed_at: u64, -} - -fn parse_metadata(doc: &AutoCommit, generation: u64) -> Result { - 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 { - compaction_id: record.compaction_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 { - 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 compaction 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 { - parse_metadata(doc, generation) -} - -/// Path of the metadata record for an epoch generation. -pub(crate) fn epoch_metadata_path(generation: u64) -> Vec { - 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 { - // 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.compaction_id, generation)); - - // Compaction records describe the old epoch's protocol and stay behind. - store::materialize_into(&base, &mut snapshot, &is_compaction_record)?; - - let mut base_heads: Vec = params.base_heads.iter().map(|h| h.to_string()).collect(); - base_heads.sort(); - - let metadata = canonical(json!({ - "generation": generation, - "compactionId": params.compaction_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 { - 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 compaction and generation. -fn snapshot_actor(compaction_id: &str, generation: u64) -> ActorId { - let digest = Sha256::new() - .chain_update(b"continuum.snapshot\0") - .chain_update(compaction_id.as_bytes()) - .chain_update(b"\0") - .chain_update(generation.to_be_bytes()) - .finalize(); - ActorId::from(&digest[..16]) -} - -fn is_compaction_record(path: &[String]) -> bool { - matches!(path, [root, records] if root == META_ROOT && records == COMPACTIONS) -} - -/// 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 = map - .into_iter() - .map(|(key, value)| (key, canonical(value))) - .collect(); - JsonValue::Object(sorted.into_iter().collect::>()) - } - 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 { - compaction_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_compaction_records_dropped() { - let mut a = doc(b"actor-a"); - store::write_path( - &mut a, - &[META_ROOT.into(), COMPACTIONS.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(COMPACTIONS).is_none()); - assert_eq!(meta["forgotten"]["pk"]["at"], json!(1)); - - let epoch = &meta[EPOCHS]["1"]; - assert_eq!(epoch["generation"], json!(1)); - assert_eq!(epoch["compactionId"], 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::compaction::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.compaction_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/compaction/switch.rs b/src/compaction/switch.rs deleted file mode 100644 index 67ff471..0000000 --- a/src/compaction/switch.rs +++ /dev/null @@ -1,248 +0,0 @@ -//! Switching this node to a committed epoch. -//! -//! The worker watches the compaction records. When a compaction 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, COMPACTIONS, META_ROOT}; -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(host: H) { - info!("Starting compaction worker"); - - let records_dir = vec![META_ROOT.to_owned(), COMPACTIONS.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; compaction worker continues on its timer only"); - changes_closed = true; - false - } - } - }; - - if should_check { - check(&host).await; - } - } -} - -async fn check(host: &H) { - let local = host.current_epoch(); - host.compaction().prune_archive(local.generation); - - // While rejoining, the document is being filled from peers: its records are not - // complete enough to act on. - if host.compaction().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!(compactions = ?ids, "Several compactions committed for this epoch; not switching on my own"); - return catchup::check(host).await; - } - }; - - if host - .compaction() - .status(&id) - .is_some_and(|status| status.error.is_some()) - { - // Our own switch failed (reported in the compaction'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.compaction().first_seen(&id).elapsed() < switch_delay { - return; - } - - match switch(host, &id, &record).await { - Ok(stats) => { - host.compaction().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!(compaction = %id, reason = %reason, "Epoch switch failed"); - host.compaction().set_status( - &id, - LocalStatus { - error: Some(reason), - ..LocalStatus::default() - }, - ); - } - } -} - -async fn switch(host: &H, id: &str, record: &Record) -> Result { - let (Some(expected), Some(request)) = (record.committed_epoch(), record.snapshot_request(id)) - else { - return Err("unreadable compaction 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!(compaction = %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 - .compaction() - .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, 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 compaction'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 compaction 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.compaction() - .archive(previous, old, retention, expected.generation); - - info!( - compaction = %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( - host: &H, - target: Epoch, - new_doc: AutoCommit, - copy_heads: Vec, -) -> Result { - 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 -} diff --git a/src/document/replica.rs b/src/document/replica.rs index 4579029..ff3af4c 100644 --- a/src/document/replica.rs +++ b/src/document/replica.rs @@ -1,7 +1,7 @@ //! The local replica: the Automerge document together with the epoch it belongs to. //! -//! Compaction (later steps) replaces the document with a snapshot whose history is -//! unrelated to the old one. Syncing two unrelated histories would make Automerge merge +//! A remake replaces the document with a new one (for compaction: a snapshot) whose +//! history is unrelated to the old one. Syncing two unrelated histories would make Automerge merge //! them, turning every key into a conflict with its own copy, so nodes on different //! epochs must never sync. Keeping the epoch and the document under one lock makes it //! impossible to label a sync message from one document with the other's epoch. @@ -13,10 +13,10 @@ use automerge::AutoCommit; /// One generation of the cluster document. /// -/// * `generation` orders epochs: every compaction increments it, so a node can tell +/// * `generation` orders epochs: every remake increments it, so a node can tell /// whether a peer is behind or ahead. /// * `id` is the hash of the epoch's snapshot change. It tells two epochs of the same -/// generation apart, which only happens if two compactions committed concurrently. +/// generation apart, which only happens if two remakes committed concurrently. #[derive(Clone, Copy, PartialEq, Eq, Hash)] pub struct Epoch { pub generation: u64, @@ -31,7 +31,7 @@ pub enum EpochRelation { PeerBehind, /// The peer is on a newer generation; we have to migrate. PeerAhead, - /// Same generation, different snapshot: competing compactions. Never sync. + /// Same generation, different snapshot: competing remakes. Never sync. Diverged, } diff --git a/src/main.rs b/src/main.rs index 1793523..ab91490 100644 --- a/src/main.rs +++ b/src/main.rs @@ -19,8 +19,8 @@ use tracing::{debug, error, info, trace}; use url::Url; mod api; -mod compaction; mod document; +mod remake; mod state; mod swarm; mod transport; @@ -155,7 +155,7 @@ async fn main() -> Result<()> { initial_sync_done: Arc::new(AtomicBool::new(false)), change_events, timing: Timing::from_keepalive(Duration::from_secs(args.keepalive_secs)), - compaction: Arc::new(compaction::Compaction::default()), + remake: Arc::new(remake::RemakeState::default()), stale: Default::default(), }; @@ -189,8 +189,8 @@ async fn main() -> Result<()> { // Switches to committed epochs and catches up with newer ones. Starts before // bootstrapping: a node joining a compacted cluster can only finish its initial sync // after fetching the current epoch from a peer. - tokio::spawn(compaction::run_worker(state.clone())); - trace!("Compaction worker task spawned"); + tokio::spawn(remake::run_worker(state.clone())); + trace!("Remake worker task spawned"); bootstrap::bootstrap(&state, &seeds) .await 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, + archive_retention: Option, + settle_secs: Option, +} + +#[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, +} + +#[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, + details: Option, +) -> 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(host: &H) -> u64 { + host.dormant_after().as_millis() as u64 +} + +fn report(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(host: &H, id: &str) -> Option { + 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(host: &H, id: &str) -> Option { + // 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( + State(host): State, + Query(query): Query, + 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 = 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 = 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( + host: H, + id: String, + mut proposal: Proposal, + now: u64, + grace: u64, +) -> Result { + let record_id = id.clone(); + let record = host + .commit_doc(move |doc| -> Result { + 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 = 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(State(host): State) -> Response { + let all = host.read_doc(records::load_all).await; + let mut reports: Vec = 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( + State(host): State, + Path(id): Path, + Query(query): Query, +) -> 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( + State(host): State, + Path(id): Path, + Query(query): Query, +) -> 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( + State(host): State, + Query(query): Query, +) -> 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(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(State(host): State) -> 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(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 = 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 = 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(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>), + Quarantine { + paths: Vec>, + 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(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(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(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(host: &H, id: &str) -> Result { + 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( + host: &H, + id: &str, + record: &Record, + all: &[(String, Record)], + now: u64, + grace: u64, +) -> Option { + 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 = 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(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(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(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], +) -> 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, + /// 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, + own: &ActorId, +) -> Missing { + let own_hex = own.to_string(); + let changes: Vec<(ChangeHash, Vec, 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 = 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, bool)> = changes + .iter() + .filter(|(hash, _, _)| !excluded.contains(hash)) + .collect(); + let depended_on: HashSet = kept + .iter() + .flat_map(|(_, deps, _)| deps.iter().copied()) + .collect(); + let mut before: Vec = 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> { + let mut paths: Vec> = 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::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 { + 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 = [(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 = [(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, +} + +#[derive(Default)] +struct Inner { + /// Built by the coordinator at commit time; consumed by the local switch. + prebuilt: HashMap, + /// When this node first saw a remake committed. + first_seen: HashMap, + status: HashMap, + archive: Vec, + /// Last attempt to catch up with a newer epoch. + catch_up: Option, + /// A rejoin in progress: the node syncs an empty document of a newer epoch. + rejoin: Option, + quarantine: Vec, +} + +/// 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, +} + +/// 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>, + /// 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, +} + +/// 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, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) replayed_paths: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) failed_paths: Option, + /// 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, + /// Why the switch did not happen (yet). + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) error: Option, +} + +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 { + 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 { + 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 { + self.lock().catch_up.clone() + } + + pub(crate) fn set_catch_up(&self, catch_up: Option) { + self.lock().catch_up = catch_up; + } + + pub(crate) fn catch_up_report(&self) -> Option { + 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 { + 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( + &self, + id: &str, + f: impl FnOnce(&Quarantined) -> T, + ) -> Option { + self.lock() + .quarantine + .iter() + .find(|entry| entry.id == id) + .map(f) + } + + pub(crate) fn list_quarantine(&self, f: impl Fn(&Quarantined) -> T) -> Vec { + self.lock().quarantine.iter().map(f).collect() + } + + pub(crate) fn take_quarantined(&self, id: &str) -> Option { + 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 { + 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, 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/`. +//! * `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 + Send; + + /// Known (not forgotten) peers. + fn peer_summaries(&self) -> impl Future> + 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 + Send; + + /// Serves requests again after [`Host::suspend`]. + fn resume(&self) -> impl Future + 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>; + + /// Reads the document on the blocking pool. + fn read_doc(&self, f: F) -> impl Future + 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(&self, f: F) -> impl Future + 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(&self, f: F) -> impl Future> + Send + where + F: FnOnce(&mut AutoCommit) -> Result + 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( + &self, + epoch: Epoch, + f: F, + ) -> impl Future> + Send + where + F: FnOnce(&mut AutoCommit) -> Result + 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, +} + +/// Admin routes. Merged into the authenticated part of the node's router. +pub(crate) fn router() -> Router { + Router::new() + .route( + "/admin/remakes", + get(api::list::) + .post(api::start::) + .delete(api::abort_current::), + ) + // The original endpoint, kept as a shortcut for a compacting remake. + .route("/admin/compaction", post(api::start::)) + .route( + "/admin/remakes/{id}", + get(api::get::).delete(api::abort::), + ) + .route("/admin/epochs", get(api::epochs::)) + .route("/admin/quarantine", get(quarantine::list::)) + .route( + "/admin/quarantine/{id}", + get(quarantine::get::).delete(quarantine::discard::), + ) + .route( + "/admin/quarantine/{id}/document", + get(quarantine::document::), + ) + .route( + "/admin/quarantine/{id}/replay", + post(quarantine::replay::), + ) +} 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(State(host): State) -> Response { + Json(host.remake().list_quarantine(summary)).into_response() +} + +pub(crate) async fn get(State(host): State, Path(id): Path) -> 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::>() + }) + .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(State(host): State, Path(id): Path) -> 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(State(host): State, Path(id): Path) -> 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(State(host): State, Path(id): Path) -> 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/` 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 for EpochRecord { + fn from(epoch: Epoch) -> Self { + Self { + generation: epoch.generation, + id: epoch.id_hex(), + } + } +} + +impl EpochRecord { + pub(crate) fn to_epoch(&self) -> Option { + 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, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) seconds: Option, +} + +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, + 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, +} + +impl Proposal { + pub(crate) fn base_change_hashes(&self) -> Option> { + 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, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) abort_requested: Option, +} + +#[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 { + 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 { + 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 { + vec![META_ROOT.to_owned(), REMAKES.to_owned()] +} + +pub(crate) fn record_path(id: &str, part: &str) -> Vec { + 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 { + 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( + 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/`: +//! 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, +} + +#[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, + 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, + 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, + policy: Policy, + #[serde(default)] + proposed_at: u64, +} + +fn parse_metadata(doc: &AutoCommit, generation: u64) -> Result { + 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 { + 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 { + parse_metadata(doc, generation) +} + +/// Path of the metadata record for an epoch generation. +pub(crate) fn epoch_metadata_path(generation: u64) -> Vec { + 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 { + // 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 = 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 { + 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 = map + .into_iter() + .map(|(key, value)| (key, canonical(value))) + .collect(); + JsonValue::Object(sorted.into_iter().collect::>()) + } + 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(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(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(host: &H, id: &str, record: &Record) -> Result { + 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, 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( + host: &H, + target: Epoch, + new_doc: AutoCommit, + copy_heads: Vec, +) -> Result { + 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 +} diff --git a/src/state.rs b/src/state.rs index 18bface..51ca58a 100644 --- a/src/state.rs +++ b/src/state.rs @@ -12,8 +12,8 @@ use tokio::sync::{broadcast, mpsc, watch, Mutex, RwLock}; use tokio::time::Instant; use tracing::{debug, error, warn}; -use crate::compaction::{Compaction, Host, PeerSummary}; use crate::document::replica::{Epoch, Replica}; +use crate::remake::{Host, PeerSummary, RemakeState}; use crate::swarm::sync::{ChangeSet, SyncEvent}; use crate::transport; use crate::transport::firewall::Firewall; @@ -298,8 +298,8 @@ pub(crate) struct AppState { pub(crate) timing: Timing, - /// Per-node compaction state (prebuilt snapshots, switch status, archive). - pub(crate) compaction: Arc, + /// Per-node remake state (prebuilt snapshots, switch status, archive). + pub(crate) remake: Arc, /// While rejoining a newer epoch, the previous document, served read-only (and /// marked stale) instead of answering 503. @@ -621,10 +621,10 @@ pub(crate) fn unix_millis() -> u64 { .unwrap_or(0) } -/// The node as seen by the compaction module. +/// The node as seen by the remake module. impl Host for AppState { - fn compaction(&self) -> &Compaction { - &self.compaction + fn remake(&self) -> &RemakeState { + &self.remake } fn node_id(&self) -> String { diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index 2907e93..47fadfa 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -54,7 +54,7 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()> if let Some(cluster_id) = state.cluster_id().await { // A node joining a compacted cluster rejoins the newer epoch first; it is // ready only once that rejoin verified the epoch. - if state.initial_sync_done.load(Ordering::Acquire) && !state.compaction.is_rejoining() { + if state.initial_sync_done.load(Ordering::Acquire) && !state.remake.is_rejoining() { state.mark_ready(cluster_id.clone()).await?; info!(cluster_id = %cluster_id, "Node is ready"); return Ok(()); diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index eb97a7e..b092b03 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -534,7 +534,7 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool .send(SyncEvent::PeerNeedsSync(peer.to_owned())); } } - // Expected after a compaction: either side switches or catches up. + // Expected after a remake: either side switches or catches up. EpochRelation::PeerAhead | EpochRelation::PeerBehind => { info!( peer = %peer, -- cgit v1.3.1