diff options
Diffstat (limited to 'src/compaction')
| -rw-r--r-- | src/compaction/api.rs | 369 | ||||
| -rw-r--r-- | src/compaction/coordinator.rs | 183 | ||||
| -rw-r--r-- | src/compaction/delta.rs | 257 | ||||
| -rw-r--r-- | src/compaction/local.rs | 198 | ||||
| -rw-r--r-- | src/compaction/mod.rs | 28 | ||||
| -rw-r--r-- | src/compaction/records.rs | 398 | ||||
| -rw-r--r-- | src/compaction/snapshot.rs | 39 | ||||
| -rw-r--r-- | src/compaction/switch.rs | 210 |
8 files changed, 1678 insertions, 4 deletions
diff --git a/src/compaction/api.rs b/src/compaction/api.rs new file mode 100644 index 0000000..b8c55db --- /dev/null +++ b/src/compaction/api.rs @@ -0,0 +1,369 @@ +//! `/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. +//! * `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 and the archived previous ones. +//! * `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. +//! +//! 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}; +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::store::StoreError; + +#[derive(Debug, Default, Deserialize)] +pub(crate) struct WaitQuery { + #[serde(default)] + wait: 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), +} + +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> { + let mut changes = host.subscribe_changes(); + loop { + let report = load_report(host, id).await?; + 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<WaitQuery>, + 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 { + return error_response( + StatusCode::BAD_REQUEST, + format!( + "settleSecs must be at least {} (the dormancy threshold), so a partition is detected before deciding", + 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, + }; + + let record_id = id.clone(); + let written = 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; + + 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" + ); + tokio::spawn(coordinator::run(host.clone(), id.clone())); + + 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() +} + +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), + } +} + +pub(crate) async fn abort<H: Host>( + State(host): State<H>, + Path(id): Path<String>, + Query(query): Query<WaitQuery>, +) -> 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 query.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 and the previous epochs this node still keeps. +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(), + })) + .into_response() +} diff --git a/src/compaction/coordinator.rs b/src/compaction/coordinator.rs new file mode 100644 index 0000000..1b41faf --- /dev/null +++ b/src/compaction/coordinator.rs @@ -0,0 +1,183 @@ +//! The coordinator: the node that received the proposal watches the cluster during the +//! settle period and then writes the decision. +//! +//! It aborts if, at any point before deciding: +//! * an operator requested an abort, +//! * a competing proposal for the same epoch appeared (both coordinators abort), +//! * a peer it considered went dormant (a partition may hide a competitor), +//! * the local epoch changed, or +//! * the decision would come later than the settle period plus the grace margin +//! (other nodes report the proposal as expired by then). +//! +//! Otherwise it builds the snapshot at the base heads and commits, recording the new +//! epoch so every node can verify its own build against it. +//! +//! The task lives only in memory. If the coordinator dies, nobody decides and the +//! proposal expires. + +use std::collections::HashMap; +use std::time::Duration; + +use tracing::{info, warn}; + +use super::records::{self, Decision, Record}; +use super::snapshot; +use super::Host; +use crate::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 new file mode 100644 index 0000000..3bc9456 --- /dev/null +++ b/src/compaction/delta.rs @@ -0,0 +1,257 @@ +//! 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 automerge::{AutoCommit, ChangeHash, PatchAction, Prop}; +use tracing::warn; + +use super::snapshot::{COMPACTIONS, EPOCHS, META_ROOT}; +use crate::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 { + let mut stats = ReplayStats::default(); + if before == after { + return stats; + } + + for path in changed_paths(src, before, after) { + 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) +} + +/// 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 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 new file mode 100644 index 0000000..6290a5e --- /dev/null +++ b/src/compaction/local.rs @@ -0,0 +1,198 @@ +//! Per-node, in-memory compaction state: snapshots built ahead of the switch, switch +//! progress for reports, and archived documents of previous epochs. +//! +//! Nothing here is replicated or durable. After a restart the node simply has no +//! archive 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::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>, +} + +/// 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>, + /// 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 later steps (serving or auditing old epochs); not read yet. + #[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 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| { + let within_epochs = archived.retention.epochs.is_some_and(|epochs| { + current_generation.saturating_sub(archived.epoch.generation) <= epochs + }); + let within_time = archived.retention.seconds.is_some_and(|seconds| { + now.saturating_sub(archived.archived_at) <= seconds.saturating_mul(1000) + }); + within_epochs || within_time + }); +} + +#[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 index 5608eae..6cb1a98 100644 --- a/src/compaction/mod.rs +++ b/src/compaction/mod.rs @@ -3,18 +3,24 @@ //! * `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). +//! * `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. //! -//! Step 4 decides compactions but does not switch epochs yet; a committed record only -//! names the new epoch. -//! //! The rest of the node is reached exclusively through [`Host`], so this module never //! depends on `AppState`, peers or sync internals. mod api; mod coordinator; +mod delta; +mod local; 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; @@ -34,6 +40,9 @@ use crate::sync::ChangeSet; /// /// 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; @@ -69,6 +78,18 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { 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)] @@ -87,4 +108,5 @@ pub(crate) fn router<H: Host>() -> Router<H> { "/admin/compactions/{id}", get(api::get::<H>).delete(api::abort::<H>), ) + .route("/admin/epochs", get(api::epochs::<H>)) } diff --git a/src/compaction/records.rs b/src/compaction/records.rs new file mode 100644 index 0000000..ce2b2a5 --- /dev/null +++ b/src/compaction/records.rs @@ -0,0 +1,398 @@ +//! 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::replica::Epoch; +use crate::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>, +} + +#[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. + pub(crate) replay_window: Retention, + /// How long nodes keep the old epoch's document. + 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()?, + }) + } + + 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::*; + + 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 index 83ff28a..1f57352 100644 --- a/src/compaction/snapshot.rs +++ b/src/compaction/snapshot.rs @@ -18,7 +18,7 @@ //! at the snapshot change itself, so edits made afterwards through the store API can't //! change what the protocol sees. -// `Snapshot::doc` and `Snapshot::clock` are consumed by the epoch switch (step 5). +// `Snapshot::clock` is used by returning nodes (step 7). #![cfg_attr(not(test), allow(dead_code))] use std::collections::BTreeMap; @@ -66,6 +66,43 @@ pub(crate) enum SnapshotError { 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, +} + +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, + } + } +} + +/// 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()))? } /// Path of the metadata record for an epoch generation. diff --git a/src/compaction/switch.rs b/src/compaction/switch.rs new file mode 100644 index 0000000..5eb1962 --- /dev/null +++ b/src/compaction/switch.rs @@ -0,0 +1,210 @@ +//! 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 stays on the old epoch and +//! reports the mismatch; fetching the snapshot from peers comes with step 6. + +use std::time::Duration; + +use automerge::{ActorId, AutoCommit}; +use tokio::sync::broadcast; +use tracing::{error, info, warn}; + +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::store; +use crate::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); + + 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() { + [] => return, + [(id, record)] => (id.clone(), record.clone()), + several => { + // The settle protocol prevents this; never guess which one is right. + let ids: Vec<&str> = several.iter().map(|(id, _)| id.as_str()).collect(); + error!(compactions = ?ids, "Several compactions committed for this epoch; not switching"); + return; + } + }; + + if host + .compaction() + .status(&id) + .is_some_and(|status| status.error.is_some()) + { + return; // already failed; reported in the compaction's status + } + + 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), + 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(); + 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<automerge::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 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 = host + .switch_epoch(expected, move |live| { + 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::<_, String>(new_doc) + }) + .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) +} |