diff options
Diffstat (limited to 'src/compaction')
| -rw-r--r-- | src/compaction/api.rs | 473 | ||||
| -rw-r--r-- | src/compaction/catchup.rs | 506 | ||||
| -rw-r--r-- | src/compaction/coordinator.rs | 183 | ||||
| -rw-r--r-- | src/compaction/delta.rs | 387 | ||||
| -rw-r--r-- | src/compaction/local.rs | 320 | ||||
| -rw-r--r-- | src/compaction/mod.rs | 154 | ||||
| -rw-r--r-- | src/compaction/quarantine.rs | 160 | ||||
| -rw-r--r-- | src/compaction/records.rs | 428 | ||||
| -rw-r--r-- | src/compaction/snapshot.rs | 638 | ||||
| -rw-r--r-- | src/compaction/switch.rs | 248 |
10 files changed, 0 insertions, 3497 deletions
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<Retention>, - archive_retention: Option<Retention>, - settle_secs: Option<u64>, -} - -#[derive(Debug, Serialize)] -#[serde(rename_all = "camelCase")] -struct Report { - id: String, - phase: Phase, - #[serde(flatten)] - record: Record, - /// What happened to this compaction on the node answering the request. - #[serde(skip_serializing_if = "Option::is_none")] - local: Option<LocalStatus>, -} - -#[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<String>, - details: Option<JsonValue>, -) -> Response { - let mut body = json!({ "error": message.into() }); - if let (Some(JsonValue::Object(extra)), Some(object)) = (details, body.as_object_mut()) { - object.extend(extra); - } - (status, Json(body)).into_response() -} - -fn grace<H: Host>(host: &H) -> u64 { - host.dormant_after().as_millis() as u64 -} - -fn report<H: Host>(host: &H, id: String, record: Record) -> Report { - Report { - phase: record.phase(unix_millis(), grace(host)), - local: host.compaction().status(&id), - id, - record, - } -} - -async fn load_report<H: Host>(host: &H, id: &str) -> Option<Report> { - let wanted = id.to_owned(); - let record = host - .read_doc(move |doc| records::load(doc, &wanted)) - .await?; - Some(report(host, id.to_owned(), record)) -} - -/// Waits until the compaction leaves `proposed`. Bounded by the proposal's own expiry, -/// at which point it is reported as `expired`. -async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> { - // A record that was readable may be briefly unreadable or absent while concurrent - // changes merge; give up only if that persists. - const MAX_MISSES: u32 = 5; - - let mut changes = host.subscribe_changes(); - let mut seen = false; - let mut misses = 0; - loop { - let Some(report) = load_report(host, id).await else { - misses += 1; - warn!(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<H: Host>( - State(host): State<H>, - Query(query): Query<StartQuery>, - body: Bytes, -) -> Response { - if !host.is_node_ready().await { - return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None); - } - - let request: StartRequest = if body.is_empty() { - StartRequest::default() - } else { - match serde_json::from_slice(&body) { - Ok(request) => request, - Err(error) => { - return error_response( - StatusCode::BAD_REQUEST, - format!("invalid request: {error}"), - None, - ) - } - } - }; - - let dormant_after = host.dormant_after(); - let settle = request - .settle_secs - .map(Duration::from_secs) - .unwrap_or(dormant_after * 2); - if settle < dormant_after && query.force { - warn!( - settle_secs = settle.as_secs_f64(), - dormant_after_secs = dormant_after.as_secs(), - "Forced settle period shorter than the dormancy threshold; a partition may go unnoticed" - ); - } else if settle < dormant_after { - return error_response( - StatusCode::BAD_REQUEST, - format!( - "settleSecs must be at least {} (the dormancy threshold), so a partition is detected before deciding; add ?force=true to override", - dormant_after.as_secs() - ), - None, - ); - } - - // Fail fast instead of proposing something that will abort anyway. - let peers = host.peer_summaries().await; - let dormant: Vec<String> = peers - .iter() - .filter(|peer| !peer.active) - .map(|peer| peer.id.clone()) - .collect(); - if !dormant.is_empty() { - return error_response( - StatusCode::CONFLICT, - "dormant peers block compaction; forget them first (DELETE /peers/{pk})", - Some(json!({ "dormantPeers": dormant })), - ); - } - - let mut peers_considered: Vec<String> = peers.into_iter().map(|peer| peer.id).collect(); - peers_considered.sort(); - - let now = unix_millis(); - let settle_ms = settle.as_millis() as u64; - let id = Uuid::new_v4().to_string(); - let grace = grace(&host); - - let mut proposal = Proposal { - coordinator: host.node_id(), - from_epoch: host.current_epoch().into(), - base_heads: Vec::new(), - policy: Policy { - replay_window: request - .replay_window - .unwrap_or_else(Policy::default_replay_window), - archive_retention: request - .archive_retention - .unwrap_or_else(Policy::default_archive_retention), - }, - settle_ms, - proposed_at: now, - settle_until: now + settle_ms, - peers_considered, - }; - - // From here on, nothing depends on the client: writing the proposal and starting the - // coordinator run in their own task, so a client that disconnects mid-request can't - // leave a proposal without a coordinator. Dropping the handle doesn't cancel it. - let proposing = tokio::spawn(propose(host.clone(), id.clone(), proposal, now, grace)); - let written = match proposing.await { - Ok(written) => written, - Err(error) => { - error!(error = %error, "Proposal task failed"); - return error_response( - StatusCode::INTERNAL_SERVER_ERROR, - "failed to write the proposal", - None, - ); - } - }; - - let record = match written { - Ok(record) => record, - Err(StartError::Open(open)) => { - return error_response( - StatusCode::CONFLICT, - "another 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<H: Host>( - host: H, - id: String, - mut proposal: Proposal, - now: u64, - grace: u64, -) -> Result<Record, StartError> { - let record_id = id.clone(); - let record = host - .commit_doc(move |doc| -> Result<Record, StartError> { - if let Some((open, _)) = records::load_all(doc).into_iter().find(|(_, record)| { - record.proposal.from_epoch == proposal.from_epoch - && record.phase(now, grace).is_open() - }) { - return Err(StartError::Open(open)); - } - - // The base is everything before the proposal; the proposal itself is not in it. - let mut heads: Vec<String> = doc.get_heads().iter().map(|h| h.to_string()).collect(); - heads.sort(); - proposal.base_heads = heads; - - records::write_proposal(doc, &record_id, &proposal)?; - Ok(Record { - proposal, - decision: None, - abort_requested: None, - }) - }) - .await?; - - tokio::spawn(coordinator::run(host, id)); - Ok(record) -} - -pub(crate) async fn list<H: Host>(State(host): State<H>) -> Response { - let all = host.read_doc(records::load_all).await; - let mut reports: Vec<Report> = all - .into_iter() - .map(|(id, record)| report(&host, id, record)) - .collect(); - reports.sort_by_key(|report| report.record.proposal.proposed_at); - Json(reports).into_response() -} - -pub(crate) async fn get<H: Host>( - State(host): State<H>, - Path(id): Path<String>, - Query(query): Query<WaitQuery>, -) -> Response { - let found = if query.wait { - wait_until_final(&host, &id).await - } else { - load_report(&host, &id).await - }; - match found { - Some(report) => Json(report).into_response(), - None => error_response(StatusCode::NOT_FOUND, "no such compaction", None), - } -} - -/// `DELETE /admin/compactions/{id}`. -pub(crate) async fn abort<H: Host>( - State(host): State<H>, - Path(id): Path<String>, - Query(query): Query<WaitQuery>, -) -> Response { - abort_by_id(host, id, query.wait).await -} - -/// `DELETE /admin/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<H: Host>( - State(host): State<H>, - Query(query): Query<WaitQuery>, -) -> Response { - let local = EpochRecord::from(host.current_epoch()); - let now = unix_millis(); - let grace = grace(&host); - let open = host - .read_doc(move |doc| { - records::load_all(doc) - .into_iter() - .filter(|(_, record)| { - record.proposal.from_epoch == local - && record.phase(now, grace) == Phase::Proposed - }) - .min_by_key(|(_, record)| record.proposal.proposed_at) - .map(|(id, _)| id) - }) - .await; - - match open { - Some(id) => abort_by_id(host, id, query.wait).await, - None => error_response( - StatusCode::NOT_FOUND, - "no compaction of the local epoch is in progress", - None, - ), - } -} - -async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { - let Some(current) = load_report(&host, &id).await else { - return error_response(StatusCode::NOT_FOUND, "no such 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<H: Host>(State(host): State<H>) -> 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<H: Host>(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<Epoch> = ahead - .iter() - .filter(|(_, epoch)| epoch.generation == generation) - .map(|(_, epoch)| *epoch) - .collect(); - targets.sort_by_key(|epoch| epoch.id); - targets.dedup(); - let [target] = targets.as_slice() else { - let listed: Vec<String> = targets.iter().map(ToString::to_string).collect(); - warn!(epochs = ?listed, "Peers disagree about the newest epoch; not catching up"); - return; - }; - let target = *target; - - // Back off between attempts for the same target. - let previous = host - .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<H: Host>(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<Vec<String>>), - Quarantine { - paths: Vec<Vec<String>>, - reason: String, - }, -} - -enum Prepared { - /// Caught up with someone, but not with the target epoch yet: keep waiting. - NotYet(AutoCommit), - Failed(AutoCommit, String), - Ready { - old: AutoCommit, - metadata: EpochMetadata, - plan: Plan, - own_missing: usize, - had_history: bool, - }, -} - -async fn finish<H: Host>(host: &H) { - let Some(rejoin) = host.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<H: Host>(host: &H, rejoin: Rejoin, reason: String) { - let Rejoin { - previous, - target, - old, - was_ready, - .. - } = rejoin; - warn!(target = %target, reason = %reason, "Rejoin failed; restoring the previous epoch"); - - if let Err(never) = host - .switch_epoch(previous, move |_live| Ok::<_, Infallible>(old)) - .await - { - match never {} - } - // A node that was never ready must sync again before bootstrap may finish. - host.reset_initial_sync(); - if was_ready { - host.resume().await; - } - host.set_stale_view(None); - - let attempts = host - .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<H: Host>(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<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> { - let wanted = id.to_owned(); - let (record, all) = host - .read_doc(move |doc| (records::load(doc, &wanted), records::load_all(doc))) - .await; - - let Some(record) = record else { - warn!(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<H: Host>( - host: &H, - id: &str, - record: &Record, - all: &[(String, Record)], - now: u64, - grace: u64, -) -> Option<String> { - let proposal = &record.proposal; - - if let Some(request) = &record.abort_requested { - return Some(format!("aborted by operator via {}", request.by)); - } - - if records::EpochRecord::from(host.current_epoch()) != proposal.from_epoch { - return Some("local epoch changed during settle".to_owned()); - } - - if let Some((other, _)) = all.iter().find(|(other, candidate)| { - other != id - && candidate.proposal.from_epoch == proposal.from_epoch - && candidate.phase(now, grace).is_open() - }) { - return Some(format!("competing compaction {other} for the same epoch")); - } - - // Peers forgotten in the meantime are ignored: an operator decided they are gone. - let peers: HashMap<String, bool> = host - .peer_summaries() - .await - .into_iter() - .map(|peer| (peer.id, peer.active)) - .collect(); - let dormant: Vec<&str> = proposal - .peers_considered - .iter() - .filter(|peer| peers.get(peer.as_str()) == Some(&false)) - .map(String::as_str) - .collect(); - if !dormant.is_empty() { - return Some(format!( - "peers went dormant during settle: {}", - dormant.join(", ") - )); - } - - if now > proposal.settle_until.saturating_add(grace) { - return Some("decision deadline passed".to_owned()); - } - - None -} - -async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), StoreError> { - let Some(request) = record.snapshot_request(id) else { - return decide_abort( - host, - id, - "unreadable epoch or base heads in proposal".into(), - ) - .await; - }; - - // Clone under the read lock, build without any lock. - let copy = host.read_doc(|doc| doc.clone()).await; - let mut snapshot = match snapshot::build_detached(copy, request).await { - Ok((snapshot, _copy)) => snapshot, - Err(error) => { - return decide_abort(host, id, format!("snapshot build failed: {error}")).await; - } - }; - - let to_epoch = snapshot.epoch; - let snapshot_bytes = snapshot.doc.save().len() as u64; - - let decision = Decision::Committed { - to_epoch: to_epoch.into(), - snapshot_bytes, - decided_at: records::unix_millis(), - }; - write_decision(host, id, decision).await?; - - // The local switch reuses it instead of building again. - host.compaction().store_prebuilt(id, snapshot); - - info!(compaction = %id, to_epoch = %to_epoch, snapshot_bytes, "Compaction committed"); - Ok(()) -} - -async fn decide_abort<H: Host>(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<H: Host>(host: &H, id: &str, decision: Decision) -> Result<(), StoreError> { - let id = id.to_owned(); - host.commit_doc(move |doc| records::write_decision(doc, &id, &decision)) - .await -} diff --git a/src/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<String>], -) -> ReplayStats { - let mut stats = ReplayStats::default(); - for path in paths { - replay_path(src, dst, path, &mut stats); - } - let _ = dst.commit(); - stats -} - -fn replay_path(src: &AutoCommit, dst: &mut AutoCommit, path: &[String], stats: &mut ReplayStats) { - if is_protocol_path(path) { - return; - } - - // Replacing an ancestor of the protocol metadata as a whole would clobber it; replay - // its children one by one instead. - if contains_protocol_path(path) { - let mut keys = store::keys_at(src, path).unwrap_or_default(); - keys.extend(store::keys_at(dst, path).unwrap_or_default()); - keys.sort(); - keys.dedup(); - for key in keys { - let mut child = path.to_vec(); - child.push(key); - replay_path(src, dst, &child, stats); - } - return; - } - - match store::copy_path(src, dst, path) { - Ok(()) => stats.replayed += 1, - Err(error) => { - log_failure(path, &error); - stats.failed += 1; - } - } -} - -fn log_failure(path: &[String], error: &StoreError) { - warn!(path = ?path, error = %error, "Could not replay change into the new epoch"); -} - -fn is_protocol_path(path: &[String]) -> bool { - matches!(path, [root, part, ..] if root == META_ROOT && (part == 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<ChangeHash>, - /// Changes by `own` that the snapshot lacks. - pub(crate) own: usize, -} - -/// Splits `doc`'s history against a snapshot `clock`: a change (actor, seq) is in the -/// snapshot iff seq <= clock[actor]. -/// -/// Only changes this node wrote itself (actor `own`) are worth replaying: other nodes' -/// changes after the base were replayed by those nodes when they switched, and replaying -/// them again would overwrite newer writes in the new epoch with stale ones. -pub(crate) fn missing_from_snapshot( - doc: &mut AutoCommit, - clock: &BTreeMap<String, u64>, - own: &ActorId, -) -> Missing { - let own_hex = own.to_string(); - let changes: Vec<(ChangeHash, Vec<ChangeHash>, bool)> = doc - .get_changes(&[]) - .iter() - .map(|change| { - let actor = change.actor_id().to_string(); - let in_snapshot = clock.get(&actor).is_some_and(|&seq| change.seq() <= seq); - let own_missing = actor == own_hex && !in_snapshot; - (change.hash(), change.deps().to_vec(), own_missing) - }) - .collect(); - - // Excluded: own missing changes and everything that (transitively) depends on them. - // A fixpoint, so it doesn't rely on the order `get_changes` returns. - let mut excluded: HashSet<ChangeHash> = changes - .iter() - .filter(|(_, _, own_missing)| *own_missing) - .map(|(hash, _, _)| *hash) - .collect(); - let own_count = excluded.len(); - loop { - let before = excluded.len(); - for (hash, deps, _) in &changes { - if !excluded.contains(hash) && deps.iter().any(|dep| excluded.contains(dep)) { - excluded.insert(*hash); - } - } - if excluded.len() == before { - break; - } - } - - // Heads of the kept part: kept changes no kept change depends on. - let kept: Vec<&(ChangeHash, Vec<ChangeHash>, bool)> = changes - .iter() - .filter(|(hash, _, _)| !excluded.contains(hash)) - .collect(); - let depended_on: HashSet<ChangeHash> = kept - .iter() - .flat_map(|(_, deps, _)| deps.iter().copied()) - .collect(); - let mut before: Vec<ChangeHash> = kept - .iter() - .map(|(hash, _, _)| *hash) - .filter(|hash| !depended_on.contains(hash)) - .collect(); - before.sort(); - - Missing { - before, - own: own_count, - } -} - -/// The minimal set of store paths covering every change between `before` and `after`: -/// sorted, and no path is a prefix of another. -pub(crate) fn changed_paths( - doc: &mut AutoCommit, - before: &[ChangeHash], - after: &[ChangeHash], -) -> Vec<Vec<String>> { - let mut paths: Vec<Vec<String>> = doc.diff(before, after).iter().map(patch_path).collect(); - paths.sort(); - paths.dedup(); - - // Sorted order puts a prefix right before its extensions, so comparing with the - // last kept path is enough. - let mut minimal: Vec<Vec<String>> = Vec::new(); - for path in paths { - if minimal.last().is_some_and(|kept| path.starts_with(kept)) { - continue; - } - minimal.push(path); - } - minimal -} - -fn patch_path(patch: &automerge::Patch) -> Vec<String> { - let mut path = Vec::new(); - for (_, prop) in &patch.path { - match prop { - Prop::Map(key) => path.push(key.clone()), - // Inside a list: the list is replayed as a whole. - Prop::Seq(_) => return path, - } - } - - match &patch.action { - PatchAction::PutMap { key, .. } | PatchAction::DeleteMap { key, .. } => { - path.push(key.clone()) - } - PatchAction::Increment { - prop: Prop::Map(key), - .. - } => path.push(key.clone()), - // List and text edits: the containing object. - _ => {} - } - path -} - -#[cfg(test)] -mod tests { - use super::*; - use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, ROOT}; - use serde_json::json; - - fn view(doc: &AutoCommit) -> serde_json::Value { - store::read_path(doc, &[]).unwrap().unwrap() - } - - #[test] - fn minimal_paths() { - let mut doc = AutoCommit::new(); - doc.put(ROOT, "untouched", 1_i64).unwrap(); - let before = doc.get_heads(); - - let m = doc.put_object(ROOT, "m", ObjType::Map).unwrap(); - doc.put(&m, "a", 1_i64).unwrap(); - doc.put(ROOT, "x", 1_i64).unwrap(); - doc.put(ROOT, "c", ScalarValue::counter(0)).unwrap(); - doc.commit(); - let middle = doc.get_heads(); - doc.increment(ROOT, "c", 2).unwrap(); - let list = doc.put_object(ROOT, "l", ObjType::List).unwrap(); - doc.insert(&list, 0, "e").unwrap(); - doc.commit(); - let after = doc.get_heads(); - - let all = changed_paths(&mut doc, &before, &after); - assert_eq!( - all, - vec![ - vec!["c".to_owned()], - vec!["l".to_owned()], - vec!["m".to_owned()], - vec!["x".to_owned()], - ] - ); - - let later = changed_paths(&mut doc, &middle, &after); - assert_eq!(later, vec![vec!["c".to_owned()], vec!["l".to_owned()]]); - } - - #[test] - fn replays_state_per_key() { - // Base shared by the old document and the snapshot. - let mut old = AutoCommit::new(); - let m = old.put_object(ROOT, "m", ObjType::Map).unwrap(); - old.put(&m, "keep", "base").unwrap(); - old.put(&m, "change", "base").unwrap(); - old.put(ROOT, "gone", true).unwrap(); - old.commit(); - let base = old.get_heads(); - - let mut new = AutoCommit::new(); - store::materialize_into(&old, &mut new, &|_| false).unwrap(); - // Something the new epoch already holds that the delta must not clobber. - let Some((_, m_new)) = new.get(ROOT, "m").unwrap() else { - panic!("m missing"); - }; - new.put(&m_new, "keep", "new epoch").unwrap(); - new.commit(); - - // The delta: one key changed, one key deleted. - old.put(&m, "change", "delta").unwrap(); - old.delete(ROOT, "gone").unwrap(); - old.commit(); - let after = old.get_heads(); - - let stats = replay(&mut old, &base, &after, &mut new); - assert_eq!(stats.failed, 0); - assert_eq!( - view(&new), - json!({ "m": { "keep": "new epoch", "change": "delta" } }) - ); - } - - #[test] - fn missing_counts_only_own_changes() { - let own = ActorId::from(&b"own-actor"[..]); - let other = ActorId::from(&b"other-actor"[..]); - - let mut doc = AutoCommit::new().with_actor(own.clone()); - doc.put(ROOT, "base", 1_i64).unwrap(); - doc.commit(); - let base = doc.get_heads(); - - // The snapshot contains the base only. - let clock: BTreeMap<String, u64> = [(own.to_string(), 1)].into_iter().collect(); - - // A foreign change after the base (replayed by its own node) ... - let mut foreign = doc.fork().with_actor(other); - foreign.put(ROOT, "theirs", 1_i64).unwrap(); - foreign.commit(); - doc.merge(&mut foreign).unwrap(); - let with_foreign = doc.get_heads(); - - // ... and one of ours. - doc.put(ROOT, "mine", 1_i64).unwrap(); - doc.commit(); - let after = doc.get_heads(); - - let missing = missing_from_snapshot(&mut doc, &clock, &own); - assert_eq!(missing.own, 1); - assert_eq!(missing.before, with_foreign); - assert_ne!(missing.before, base); - assert_eq!( - changed_paths(&mut doc, &missing.before, &after), - vec![vec!["mine".to_owned()]] - ); - - // Nothing of ours missing: nothing to replay. - let full: BTreeMap<String, u64> = [(own.to_string(), 2)].into_iter().collect(); - let none = missing_from_snapshot(&mut doc, &full, &own); - assert_eq!(none.own, 0); - let mut heads = after.clone(); - heads.sort(); - assert_eq!(none.before, heads); - } - - #[test] - fn protocol_metadata_is_never_replayed() { - let mut old = AutoCommit::new(); - old.put(ROOT, "x", 1_i64).unwrap(); - old.commit(); - let base = old.get_heads(); - - let mut new = AutoCommit::new(); - store::write_path( - &mut new, - &[META_ROOT.into(), EPOCHS.into(), "1".into()], - &json!({ "generation": 1 }), - ) - .unwrap(); - new.commit(); - - // The old epoch replaces `_continuum` wholesale and writes a 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<Inner>, -} - -#[derive(Default)] -struct Inner { - /// Built by the coordinator at commit time; consumed by the local switch. - prebuilt: HashMap<String, Snapshot>, - /// When this node first saw a compaction committed. - first_seen: HashMap<String, Instant>, - status: HashMap<String, LocalStatus>, - archive: Vec<Archived>, - /// Last attempt to catch up with a newer epoch. - catch_up: Option<CatchUp>, - /// A rejoin in progress: the node syncs an empty document of a newer epoch. - rejoin: Option<Rejoin>, - quarantine: Vec<Quarantined>, -} - -/// Catching up with a newer epoch: attempts and the last failure. -#[derive(Clone, Debug)] -pub(crate) struct CatchUp { - pub(crate) target: Epoch, - pub(crate) attempts: u32, - pub(crate) last_attempt: Instant, - pub(crate) error: Option<String>, -} - -/// The node left `previous` and is syncing `target` from scratch. -pub(crate) struct Rejoin { - pub(crate) previous: Epoch, - pub(crate) target: Epoch, - /// The document of `previous`, until it is replayed, quarantined or restored. - pub(crate) old: AutoCommit, - pub(crate) started: Instant, - /// The node served requests before; it does again once the rejoin is over. - pub(crate) was_ready: bool, -} - -/// Changes of this node that were not replayed into the current epoch. -pub(crate) struct Quarantined { - pub(crate) id: String, - /// The epoch the changes were made in, and the one the node rejoined. - pub(crate) epoch: Epoch, - pub(crate) target: Epoch, - pub(crate) created_at: u64, - pub(crate) reason: String, - /// Store paths the changes touched (minimal, prefix-free). - pub(crate) paths: Vec<Vec<String>>, - /// The old document; `paths` are read from its merged view. - pub(crate) document: AutoCommit, -} - -#[derive(Clone, Debug, Serialize)] -#[serde(rename_all = "camelCase")] -pub(crate) struct CatchUpReport { - pub(crate) target: String, - pub(crate) attempts: u32, - /// True while the node syncs the target epoch (not serving requests). - pub(crate) rejoining: bool, - #[serde(skip_serializing_if = "Option::is_none")] - pub(crate) error: Option<String>, -} - -/// What happened to a 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<u64>, - #[serde(skip_serializing_if = "Option::is_none")] - pub(crate) replayed_paths: Option<usize>, - #[serde(skip_serializing_if = "Option::is_none")] - pub(crate) failed_paths: Option<usize>, - /// ID of the quarantine entry holding this node's changes that weren't replayed. - #[serde(skip_serializing_if = "Option::is_none")] - pub(crate) quarantine: Option<String>, - /// Why the switch did not happen (yet). - #[serde(skip_serializing_if = "Option::is_none")] - pub(crate) error: Option<String>, -} - -struct Archived { - epoch: Epoch, - // Kept for auditing and recovery by hand; not read by the protocol. - #[allow(dead_code)] - doc: AutoCommit, - archived_at: u64, - retention: Retention, -} - -/// Summary of an archived epoch. -#[derive(Clone, Debug, Serialize)] -#[serde(rename_all = "camelCase")] -pub(crate) struct ArchivedEpoch { - pub(crate) generation: u64, - pub(crate) id: String, - pub(crate) archived_at: u64, -} - -impl 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<Snapshot> { - 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<LocalStatus> { - self.lock().status.get(id).cloned() - } - - pub(crate) fn set_status(&self, id: &str, status: LocalStatus) { - self.lock().status.insert(id.to_owned(), status); - } - - /// Keeps the previous epoch's document, then drops what has outlived its retention. - pub(crate) fn archive( - &self, - epoch: Epoch, - doc: AutoCommit, - retention: Retention, - current_generation: u64, - ) { - let mut inner = self.lock(); - inner.archive.push(Archived { - epoch, - doc, - archived_at: unix_millis(), - retention, - }); - prune(&mut inner.archive, current_generation); - } - - pub(crate) fn prune_archive(&self, current_generation: u64) { - prune(&mut self.lock().archive, current_generation); - } - - pub(crate) fn catch_up(&self) -> Option<CatchUp> { - self.lock().catch_up.clone() - } - - pub(crate) fn set_catch_up(&self, catch_up: Option<CatchUp>) { - self.lock().catch_up = catch_up; - } - - pub(crate) fn catch_up_report(&self) -> Option<CatchUpReport> { - let inner = self.lock(); - let rejoining = inner.rejoin.is_some(); - inner.catch_up.as_ref().map(|catch_up| CatchUpReport { - target: catch_up.target.to_string(), - attempts: catch_up.attempts, - rejoining, - error: catch_up.error.clone(), - }) - } - - pub(crate) fn is_rejoining(&self) -> bool { - self.lock().rejoin.is_some() - } - - /// Progress of the rejoin in progress: (target, started). - pub(crate) fn rejoin_progress(&self) -> Option<(Epoch, Instant)> { - self.lock() - .rejoin - .as_ref() - .map(|rejoin| (rejoin.target, rejoin.started)) - } - - pub(crate) fn start_rejoin(&self, rejoin: Rejoin) { - self.lock().rejoin = Some(rejoin); - } - - pub(crate) fn take_rejoin(&self) -> Option<Rejoin> { - self.lock().rejoin.take() - } - - pub(crate) fn add_quarantine(&self, quarantined: Quarantined) { - self.lock().quarantine.push(quarantined); - } - - /// Runs `f` on the quarantined entry with this ID. - pub(crate) fn with_quarantined<T>( - &self, - id: &str, - f: impl FnOnce(&Quarantined) -> T, - ) -> Option<T> { - self.lock() - .quarantine - .iter() - .find(|entry| entry.id == id) - .map(f) - } - - pub(crate) fn list_quarantine<T>(&self, f: impl Fn(&Quarantined) -> T) -> Vec<T> { - self.lock().quarantine.iter().map(f).collect() - } - - pub(crate) fn take_quarantined(&self, id: &str) -> Option<Quarantined> { - let mut inner = self.lock(); - let index = inner.quarantine.iter().position(|entry| entry.id == id)?; - Some(inner.quarantine.remove(index)) - } - - /// Puts an entry back, e.g. after a failed replay. - pub(crate) fn restore_quarantined(&self, quarantined: Quarantined) { - self.lock().quarantine.push(quarantined); - } - - pub(crate) fn archived_epochs(&self) -> Vec<ArchivedEpoch> { - self.lock() - .archive - .iter() - .map(|archived| ArchivedEpoch { - generation: archived.epoch.generation, - id: archived.epoch.id_hex(), - archived_at: archived.archived_at, - }) - .collect() - } -} - -/// An archive is kept until every limit of its retention has passed. -fn prune(archive: &mut Vec<Archived>, current_generation: u64) { - let now = unix_millis(); - archive.retain(|archived| { - archived.retention.covers( - current_generation.saturating_sub(archived.epoch.generation), - now.saturating_sub(archived.archived_at), - ) - }); -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn archive_retention() { - let 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/<id>`. -//! * `coordinator`: proposal, settle period and decision on the node that was asked. -//! * `switch`: switching this node to a committed epoch (worker). -//! * `catchup`: rejoining a newer epoch from peers (late, new or failed nodes), then -//! replaying or quarantining this node's own missing changes. -//! * `quarantine`: inspecting, replaying or discarding quarantined changes. -//! * `delta`: replaying this node's changes since the base into the new epoch. -//! * `local`: per-node in-memory state (prebuilt snapshots, switch status, archive). -//! * `api`: `/admin/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<Output = bool> + Send; - - /// Known (not forgotten) peers. - fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send; - - /// Stops serving requests (back to `Joining`, keeping the cluster ID), as while a - /// rejoin syncs a fresh document. Returns whether the node was serving. - fn suspend(&self) -> impl Future<Output = bool> + Send; - - /// Serves requests again after [`Host::suspend`]. - fn resume(&self) -> impl Future<Output = ()> + Send; - - /// Serves reads from this document (read-only, marked stale) while the node is not - /// ready; `None` ends it. - fn set_stale_view(&self, view: Option<(Epoch, AutoCommit)>); - - /// Whether the document caught up with a peer since the last reset. - fn initial_sync_done(&self) -> bool; - - /// Makes the next sync count as initial again. - fn reset_initial_sync(&self); - - /// Store paths touched by each published document change. - fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>>; - - /// Reads the document on the blocking pool. - fn read_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send - where - F: FnOnce(&AutoCommit) -> T + Send + 'static, - T: Send + 'static; - - /// Mutable access to the document without publishing a change (e.g. forks). - fn update_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send - where - F: FnOnce(&mut AutoCommit) -> T + Send + 'static, - T: Send + 'static; - - /// One committed change (rolled back on error), published and replicated. - fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send - where - F: FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static, - T: Send + 'static, - E: Send + 'static; - - /// Replaces the document and its epoch atomically. Under the write lock, `f` gets - /// the live document and returns the new one. Afterwards the node resets every - /// peer's sync state and publishes the change. Returns the old document. - fn switch_epoch<E, F>( - &self, - epoch: Epoch, - f: F, - ) -> impl Future<Output = Result<AutoCommit, E>> + Send - where - F: FnOnce(&mut AutoCommit) -> Result<AutoCommit, E> + Send + 'static, - E: Send + 'static; -} - -#[derive(Clone, Debug)] -pub(crate) struct PeerSummary { - pub(crate) id: String, - /// Not dormant. - pub(crate) active: bool, - /// Handshake completed. - pub(crate) established: bool, - /// The epoch the peer last advertised. - pub(crate) epoch: Option<Epoch>, -} - -/// Admin routes. Merged into the authenticated part of the node's router. -pub(crate) fn router<H: Host>() -> Router<H> { - Router::new() - .route( - "/admin/compaction", - post(api::start::<H>).delete(api::abort_current::<H>), - ) - .route("/admin/compactions", get(api::list::<H>)) - .route( - "/admin/compactions/{id}", - get(api::get::<H>).delete(api::abort::<H>), - ) - .route("/admin/epochs", get(api::epochs::<H>)) - .route("/admin/quarantine", get(quarantine::list::<H>)) - .route( - "/admin/quarantine/{id}", - get(quarantine::get::<H>).delete(quarantine::discard::<H>), - ) - .route( - "/admin/quarantine/{id}/document", - get(quarantine::document::<H>), - ) - .route( - "/admin/quarantine/{id}/replay", - post(quarantine::replay::<H>), - ) -} diff --git a/src/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<H: Host>(State(host): State<H>) -> Response { - Json(host.compaction().list_quarantine(summary)).into_response() -} - -pub(crate) async fn get<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response { - let Some((summary, document, paths)) = host.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::<Vec<JsonValue>>() - }) - .await; - - match writes { - Ok(writes) => Json(json!({ "entry": summary, "writes": writes })).into_response(), - Err(_) => error_response( - StatusCode::INTERNAL_SERVER_ERROR, - "reading the writes failed", - None, - ), - } -} - -pub(crate) async fn document<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response { - let Some(mut document) = host - .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<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response { - if !host.is_node_ready().await { - return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None); - } - let Some(entry) = host.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<H: Host>(State(host): State<H>, Path(id): Path<String>) -> 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/<id>` in the replicated document. -//! -//! A record has up to three parts, each with a single kind of writer, so concurrent -//! writes never fight over the same key: -//! -//! * `proposal`: written once, by the coordinator. -//! * `decision`: written once, by the coordinator only (committed or aborted). -//! * `abortRequested`: written by any node on an operator's request. It is an input to -//! the coordinator's decision, not a decision itself. -//! -//! A proposal whose coordinator never decides (crash, partition) is reported as -//! `expired` once its settle period plus a grace margin has passed. Expiry is derived, -//! never written, so a late but valid commit can't collide with someone's abort. -//! -//! Records edited into an unreadable shape through the store API are skipped with a -//! warning. - -use automerge::{AutoCommit, ChangeHash}; -use serde::{Deserialize, Serialize}; -use tracing::warn; - -use super::snapshot::{SnapshotRequest, 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<Epoch> for EpochRecord { - fn from(epoch: Epoch) -> Self { - Self { - generation: epoch.generation, - id: epoch.id_hex(), - } - } -} - -impl EpochRecord { - pub(crate) fn to_epoch(&self) -> Option<Epoch> { - Some(Epoch { - generation: self.generation, - id: decode_hex32(&self.id)?, - }) - } -} - -/// How long something is kept. A value is kept until every given limit has passed. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub(crate) struct Retention { - #[serde(default, skip_serializing_if = "Option::is_none")] - pub(crate) epochs: Option<u64>, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub(crate) seconds: Option<u64>, -} - -impl Retention { - /// True while at least one given limit has not passed: `epochs` epochs and - /// `age_ms` milliseconds after the fact. - pub(crate) fn covers(&self, epochs: u64, age_ms: u64) -> bool { - self.epochs.is_some_and(|limit| epochs <= limit) - || self - .seconds - .is_some_and(|limit| age_ms <= limit.saturating_mul(1000)) - } -} - -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub(crate) struct Policy { - /// How stale a returning node's changes may be and still be replayed automatically; - /// beyond it they are quarantined. - #[serde(default = "Policy::default_replay_window")] - pub(crate) replay_window: Retention, - /// How long nodes keep the old epoch's document. - #[serde(default = "Policy::default_archive_retention")] - pub(crate) archive_retention: Retention, -} - -impl Policy { - pub(crate) fn default_replay_window() -> Retention { - Retention { - epochs: Some(1), - seconds: Some(7 * DAY_SECS), - } - } - - pub(crate) fn default_archive_retention() -> Retention { - Retention { - epochs: Some(2), - seconds: Some(30 * DAY_SECS), - } - } -} - -#[derive(Clone, Debug, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub(crate) struct Proposal { - pub(crate) coordinator: String, - pub(crate) from_epoch: EpochRecord, - /// Hex change hashes, sorted. The proposal record itself is not part of the base. - pub(crate) base_heads: Vec<String>, - pub(crate) policy: Policy, - pub(crate) settle_ms: u64, - pub(crate) proposed_at: u64, - pub(crate) settle_until: u64, - /// Peers that must stay active during the settle period. - pub(crate) peers_considered: Vec<String>, -} - -impl Proposal { - pub(crate) fn base_change_hashes(&self) -> Option<Vec<ChangeHash>> { - self.base_heads - .iter() - .map(|hex| decode_hex32(hex).map(ChangeHash)) - .collect() - } -} - -#[derive(Clone, Debug, Serialize, Deserialize)] -#[serde(tag = "outcome", rename_all = "camelCase")] -pub(crate) enum Decision { - #[serde(rename_all = "camelCase")] - Committed { - to_epoch: EpochRecord, - snapshot_bytes: u64, - decided_at: u64, - }, - #[serde(rename_all = "camelCase")] - Aborted { reason: String, decided_at: u64 }, -} - -#[derive(Clone, Debug, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub(crate) struct AbortRequest { - pub(crate) by: String, - pub(crate) at: u64, -} - -#[derive(Clone, Debug, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub(crate) struct Record { - pub(crate) proposal: Proposal, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub(crate) decision: Option<Decision>, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub(crate) abort_requested: Option<AbortRequest>, -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] -#[serde(rename_all = "camelCase")] -pub(crate) enum Phase { - /// Settling; the coordinator has not decided yet. - Proposed, - /// No decision long after the settle period: the coordinator is gone. - Expired, - Committed, - Aborted, -} - -impl Phase { - /// Still able to become (or already) the next epoch. - pub(crate) fn is_open(self) -> bool { - matches!(self, Phase::Proposed | Phase::Committed) - } - - pub(crate) fn is_final(self) -> bool { - !matches!(self, Phase::Proposed) - } -} - -impl Record { - /// The epoch this compaction committed to, if committed and readable. - pub(crate) fn committed_epoch(&self) -> Option<Epoch> { - match &self.decision { - Some(Decision::Committed { to_epoch, .. }) => to_epoch.to_epoch(), - _ => None, - } - } - - /// Everything needed to (re)build this compaction's snapshot; None if the proposal - /// is unreadable. - pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> { - 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<String> { - vec![META_ROOT.to_owned(), COMPACTIONS.to_owned()] -} - -pub(crate) fn record_path(id: &str, part: &str) -> Vec<String> { - let mut path = records_dir(); - path.push(id.to_owned()); - if !part.is_empty() { - path.push(part.to_owned()); - } - path -} - -/// The record with this ID, if it exists and is readable. -pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> { - let value = match store::read_path(doc, &record_path(id, "")) { - Ok(Some(value)) => value, - Ok(None) => return None, - Err(error) => { - warn!(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<T: Serialize>( - doc: &mut AutoCommit, - id: &str, - part: &str, - value: &T, -) -> Result<(), StoreError> { - let json = serde_json::to_value(value).map_err(anyhow::Error::from)?; - store::write_path(doc, &record_path(id, part), &json) -} - -pub(crate) fn write_proposal( - doc: &mut AutoCommit, - id: &str, - proposal: &Proposal, -) -> Result<(), StoreError> { - write_part(doc, id, "proposal", proposal) -} - -pub(crate) fn write_decision( - doc: &mut AutoCommit, - id: &str, - decision: &Decision, -) -> Result<(), StoreError> { - write_part(doc, id, "decision", decision) -} - -pub(crate) fn write_abort_request( - doc: &mut AutoCommit, - id: &str, - request: &AbortRequest, -) -> Result<(), StoreError> { - write_part(doc, id, "abortRequested", request) -} - -fn decode_hex32(hex: &str) -> Option<[u8; 32]> { - if hex.len() != 64 || !hex.is_ascii() { - return None; - } - let mut out = [0u8; 32]; - for (i, byte) in out.iter_mut().enumerate() { - *byte = u8::from_str_radix(&hex[2 * i..2 * i + 2], 16).ok()?; - } - Some(out) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn retention_covers_until_every_limit_passed() { - let both = Retention { - epochs: Some(1), - seconds: Some(10), - }; - assert!(both.covers(1, 60_000)); // epochs still within - assert!(both.covers(5, 5_000)); // time still within - assert!(!both.covers(2, 10_001)); - let none = Retention { - epochs: None, - seconds: None, - }; - assert!(!none.covers(0, 0)); - } - - fn proposal(settle_until: u64) -> Proposal { - Proposal { - coordinator: "node-a".into(), - from_epoch: Epoch::GENESIS.into(), - base_heads: vec!["ab".repeat(32)], - policy: Policy { - replay_window: Policy::default_replay_window(), - archive_retention: Policy::default_archive_retention(), - }, - settle_ms: 1_000, - proposed_at: settle_until.saturating_sub(1_000), - settle_until, - peers_considered: vec!["node-b".into()], - } - } - - #[test] - fn round_trip_through_the_document() { - let mut doc = AutoCommit::new(); - let p = proposal(10_000); - write_proposal(&mut doc, "c-1", &p).unwrap(); - write_abort_request( - &mut doc, - "c-1", - &AbortRequest { - by: "x".into(), - at: 5, - }, - ) - .unwrap(); - let decision = Decision::Committed { - to_epoch: EpochRecord { - generation: 1, - id: "cd".repeat(32), - }, - snapshot_bytes: 123, - decided_at: 11_000, - }; - write_decision(&mut doc, "c-1", &decision).unwrap(); - doc.commit(); - - let all = load_all(&doc); - assert_eq!(all.len(), 1); - let (id, record) = &all[0]; - assert_eq!(id, "c-1"); - assert_eq!(record.proposal.policy, p.policy); - assert_eq!(record.proposal.base_change_hashes().unwrap().len(), 1); - assert!(record.abort_requested.is_some()); - assert_eq!(record.phase(0, 0), Phase::Committed); - } - - #[test] - fn phases() { - let mut record = Record { - proposal: proposal(10_000), - decision: None, - abort_requested: None, - }; - assert_eq!(record.phase(10_000, 500), Phase::Proposed); - assert_eq!(record.phase(10_500, 500), Phase::Proposed); - assert_eq!(record.phase(10_501, 500), Phase::Expired); - record.decision = Some(Decision::Aborted { - reason: "test".into(), - decided_at: 1, - }); - assert_eq!(record.phase(0, 0), Phase::Aborted); - } - - #[test] - fn unreadable_records_are_skipped() { - let mut doc = AutoCommit::new(); - store::write_path( - &mut doc, - &record_path("broken", ""), - &serde_json::json!({ "proposal": "not an object" }), - ) - .unwrap(); - write_proposal(&mut doc, "fine", &proposal(1)).unwrap(); - doc.commit(); - let all = load_all(&doc); - assert_eq!(all.len(), 1); - assert_eq!(all[0].0, "fine"); - } - - #[test] - fn epoch_record_round_trip() { - let epoch = Epoch { - generation: 3, - id: [0x5a; 32], - }; - assert_eq!(EpochRecord::from(epoch).to_epoch(), Some(epoch)); - assert_eq!( - EpochRecord { - generation: 1, - id: "zz".into() - } - .to_epoch(), - None - ); - } -} diff --git a/src/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/<generation>`: -//! previous epoch, base heads, per-actor clock and retention policy. Later steps read it -//! at the snapshot change itself, so edits made afterwards through the store API can't -//! change what the protocol sees. - -// `Snapshot::clock` is kept for returning nodes (step 7); late nodes read the clock from -// the metadata instead. -#![cfg_attr(not(test), allow(dead_code))] - -use std::collections::BTreeMap; - -use automerge::{transaction::CommitOptions, ActorId, AutoCommit, ChangeHash}; -use serde_json::{json, Map as JsonMap, Value as JsonValue}; -use sha2::{Digest, Sha256}; - -use super::records::{EpochRecord, Policy}; -use crate::document::replica::Epoch; -use crate::document::store::{self, StoreError}; - -pub(crate) const META_ROOT: &str = "_continuum"; -pub(crate) const 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<String, u64>, -} - -#[derive(Debug, thiserror::Error)] -pub(crate) enum SnapshotError { - #[error("base head {0} is not in the local document")] - MissingBase(ChangeHash), - #[error("snapshot produced {0} heads instead of exactly one")] - UnexpectedHeads(usize), - #[error(transparent)] - Doc(#[from] automerge::AutomergeError), - #[error(transparent)] - Store(#[from] StoreError), - #[error("snapshot task failed: {0}")] - Task(String), -} - -/// Owned version of [`SnapshotParams`], for moving into a blocking task. -#[derive(Clone, Debug)] -pub(crate) struct SnapshotRequest { - pub(crate) compaction_id: String, - pub(crate) previous: Epoch, - pub(crate) base_heads: Vec<ChangeHash>, - pub(crate) policy: JsonValue, - pub(crate) proposed_at: u64, -} - -impl SnapshotRequest { - pub(crate) fn params(&self) -> SnapshotParams<'_> { - SnapshotParams { - 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<String, u64>, - 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<String, u64>, - policy: Policy, - #[serde(default)] - proposed_at: u64, -} - -fn parse_metadata(doc: &AutoCommit, generation: u64) -> Result<EpochMetadata, String> { - let value = store::read_path(doc, &epoch_metadata_path(generation)) - .map_err(|error| error.to_string())? - .ok_or_else(|| format!("no epoch metadata for generation {generation}"))?; - let record: MetadataRecord = serde_json::from_value(value) - .map_err(|error| format!("invalid epoch metadata: {error}"))?; - if record.generation != generation { - return Err(format!( - "epoch metadata names generation {} instead of {generation}", - record.generation - )); - } - let previous = record - .previous - .to_epoch() - .ok_or_else(|| "invalid previous epoch in metadata".to_owned())?; - - Ok(EpochMetadata { - 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<EpochMetadata, String> { - let root = ChangeHash(epoch.id); - let Some(change) = doc.get_change_by_hash(&root) else { - return Err(format!( - "document does not contain the snapshot change of {epoch}" - )); - }; - if !change.deps().is_empty() { - return Err(format!("snapshot change of {epoch} has dependencies")); - } - - let at_snapshot = doc.fork_at(&[root]).map_err(|error| error.to_string())?; - parse_metadata(&at_snapshot, epoch.generation) -} - -/// Metadata of an earlier generation, as the current document holds it. Snapshots -/// carry the metadata of all earlier epochs forward (only 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<EpochMetadata, String> { - parse_metadata(doc, generation) -} - -/// Path of the metadata record for an epoch generation. -pub(crate) fn epoch_metadata_path(generation: u64) -> Vec<String> { - vec![ - META_ROOT.to_owned(), - EPOCHS.to_owned(), - generation.to_string(), - ] -} - -/// Builds the snapshot of `source` at `params.base_heads`. -pub(crate) fn build( - source: &mut AutoCommit, - params: &SnapshotParams<'_>, -) -> Result<Snapshot, SnapshotError> { - // fork_at with unknown heads would quietly build something else. - for head in params.base_heads { - if source.get_change_by_hash(head).is_none() { - return Err(SnapshotError::MissingBase(*head)); - } - } - - let mut base = source.fork_at(params.base_heads)?; - let clock = clock_of(&mut base); - let generation = params.previous.generation + 1; - - let mut snapshot = - AutoCommit::new().with_actor(snapshot_actor(params.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<String> = 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<String, u64> { - let mut clock = BTreeMap::new(); - for change in doc.get_changes(&[]) { - let seq = clock.entry(change.actor_id().to_string()).or_insert(0); - *seq = (*seq).max(change.seq()); - } - clock -} - -/// The same 16-byte actor on every node, unique per 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<String, JsonValue> = map - .into_iter() - .map(|(key, value)| (key, canonical(value))) - .collect(); - JsonValue::Object(sorted.into_iter().collect::<JsonMap<_, _>>()) - } - JsonValue::Array(items) => JsonValue::Array(items.into_iter().map(canonical).collect()), - other => other, - } -} - -#[cfg(test)] -mod tests { - use super::*; - use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, Value, ROOT}; - - fn doc(actor: &[u8]) -> AutoCommit { - AutoCommit::new().with_actor(ActorId::from(actor)) - } - - fn params<'a>(heads: &'a [ChangeHash], policy: &'a JsonValue) -> SnapshotParams<'a> { - SnapshotParams { - 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<H: Host>(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<H: Host>(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<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplayStats, String> { - let (Some(expected), Some(request)) = (record.committed_epoch(), record.snapshot_request(id)) - else { - return Err("unreadable 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<ChangeHash>, ReplayStats), String> { - let mut copy = copy; - let snapshot = match prebuilt { - Some(snapshot) => snapshot, - None => snapshot::build(&mut copy, &request.params()) - .map_err(|error| format!("snapshot build failed: {error}"))?, - }; - if snapshot.epoch != expected { - return Err(format!( - "built snapshot {} does not match the committed epoch {expected}", - snapshot.epoch - )); - } - - // A fresh actor: this node's writes in the new epoch. - let mut doc = snapshot.doc.with_actor(ActorId::random()); - - // Keep this 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<H: Host>( - host: &H, - target: Epoch, - new_doc: AutoCommit, - copy_heads: Vec<ChangeHash>, -) -> Result<AutoCommit, String> { - host.switch_epoch(target, move |live| { - // Only one worker switches, but be sure the copy still describes the live document. - if !copy_heads - .iter() - .all(|head| live.get_change_by_hash(head).is_some()) - { - return Err("the document was replaced while preparing the switch".to_owned()); - } - let mut new_doc = new_doc; - let live_heads = live.get_heads(); - let late = delta::replay(live, ©_heads, &live_heads, &mut new_doc); - if late.replayed + late.failed > 0 { - info!( - replayed = late.replayed, - failed = late.failed, - "Replayed late changes" - ); - } - Ok(new_doc) - }) - .await -} |