add epoch switch

12 files changed, 1812 insertions, 11 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(&copy, &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, &copy_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)
+}
diff --git a/src/main.rs b/src/main.rs
index 8e128f0..1127cd8 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -157,6 +157,7 @@ async fn main() -> Result<()> {
         initial_sync_done: Arc::new(AtomicBool::new(false)),
         change_events,
         timing: Timing::from_keepalive(Duration::from_secs(args.keepalive_secs)),
+        compaction: Arc::new(compaction::Compaction::default()),
     };
 
     info!("Application state initialized");
@@ -195,6 +196,9 @@ async fn main() -> Result<()> {
     let publisher_state = state.clone();
     tokio::spawn(async move { sync::run_change_publisher(publisher_state).await });
 
+    // Switches to committed epochs. Runs only once the node is Ready.
+    tokio::spawn(compaction::run_worker(state.clone()));
+
     info!(
         cluster_id = ?state.cluster_id().await,
         ready = state.is_ready().await,
diff --git a/src/state.rs b/src/state.rs
index 8e595af..a8b1917 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -12,7 +12,7 @@ use tokio::sync::{broadcast, mpsc, watch, Mutex, RwLock};
 use tokio::time::Instant;
 use tracing::{debug, error, warn};
 
-use crate::compaction::{Host, PeerSummary};
+use crate::compaction::{Compaction, Host, PeerSummary};
 use crate::firewall::Firewall;
 use crate::replica::{Epoch, Replica};
 use crate::sync::{ChangeSet, SyncEvent};
@@ -279,6 +279,9 @@ pub(crate) struct AppState {
     pub(crate) change_events: broadcast::Sender<Arc<ChangeSet>>,
 
     pub(crate) timing: Timing,
+
+    /// Per-node compaction state (prebuilt snapshots, switch status, archive).
+    pub(crate) compaction: Arc<Compaction>,
 }
 
 impl AppState {
@@ -495,6 +498,31 @@ impl AppState {
     }
 }
 
+impl AppState {
+    /// After the document was replaced: every peer's sync state refers to the old
+    /// document, so start over with all of them. Peers still on the old epoch are
+    /// skipped by the sync worker until they switch too.
+    async fn after_epoch_switch(&self) {
+        let peers: Vec<(String, Arc<Mutex<PeerState>>)> = self
+            .peer_states
+            .read()
+            .await
+            .iter()
+            .map(|(id, peer)| (id.clone(), Arc::clone(peer)))
+            .collect();
+
+        for (_, peer) in &peers {
+            peer.lock().await.reset_sync();
+        }
+
+        self.publish_change();
+
+        for (id, _) in peers {
+            let _ = self.sync_events.send(SyncEvent::PeerNeedsSync(id));
+        }
+    }
+}
+
 pub(crate) fn unix_millis() -> u64 {
     std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
@@ -504,6 +532,10 @@ pub(crate) fn unix_millis() -> u64 {
 
 /// The node as seen by the compaction module.
 impl Host for AppState {
+    fn compaction(&self) -> &Compaction {
+        &self.compaction
+    }
+
     fn node_id(&self) -> String {
         self.local_id().to_owned()
     }
@@ -572,4 +604,38 @@ impl Host for AppState {
         let state = self.clone();
         async move { state.mutate_doc(f).await }
     }
+
+    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,
+    {
+        let state = self.clone();
+        async move {
+            let swapped = {
+                let state = state.clone();
+                tokio::task::spawn_blocking(move || -> Result<AutoCommit, E> {
+                    let mut replica = state.doc.blocking_write();
+                    let new_doc = f(&mut replica.doc)?;
+                    let old_doc = std::mem::replace(&mut replica.doc, new_doc);
+                    replica.epoch = epoch;
+                    // Updated under the same lock, so the cached epoch never disagrees
+                    // with the document for anyone holding the lock.
+                    state.epoch_tx.send_replace(epoch);
+                    Ok(old_doc)
+                })
+                .await
+                .expect("epoch switch task panicked")
+            };
+
+            if swapped.is_ok() {
+                state.after_epoch_switch().await;
+            }
+            swapped
+        }
+    }
 }
diff --git a/src/store.rs b/src/store.rs
index 1f1398e..1815635 100644
--- a/src/store.rs
+++ b/src/store.rs
@@ -250,9 +250,19 @@ pub fn write_path(
     value: &JsonValue,
 ) -> Result<(), StoreError> {
     let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?;
+    let maps = ensure_maps(doc, parents)?;
 
+    // Replace semantics: clear the key in every merged parent, or an old map
+    // would merge into the new value.
+    delete_in_all(doc, &maps, leaf)?;
+    json_to_automerge(doc, &maps[0], leaf, value)?;
+    Ok(())
+}
+
+/// The merged maps at `path`, creating missing maps on the way.
+fn ensure_maps(doc: &mut AutoCommit, path: &[String]) -> Result<Vec<ObjId>, StoreError> {
     let mut maps = vec![automerge::ROOT];
-    for segment in parents {
+    for segment in path {
         maps = match resolve(doc, &maps, segment, None)? {
             Some(Resolved::Maps(next)) => next,
             Some(Resolved::Other(..)) => return Err(StoreError::Conflict),
@@ -260,12 +270,53 @@ pub fn write_path(
             None => vec![doc.put_object(&maps[0], segment.as_str(), ObjType::Map)?],
         };
     }
+    Ok(maps)
+}
 
-    // Replace semantics: clear the key in every merged parent, or an old map
-    // would merge into the new value.
-    delete_in_all(doc, &maps, leaf)?;
-    json_to_automerge(doc, &maps[0], leaf, value)?;
-    Ok(())
+/// The resolved value at `path` (the empty path is the root map).
+fn resolve_at(doc: &AutoCommit, path: &[String]) -> Result<Option<Resolved>, StoreError> {
+    let mut current = Resolved::Maps(vec![automerge::ROOT]);
+    for segment in path {
+        let Resolved::Maps(maps) = &current else {
+            return Ok(None);
+        };
+        match resolve(doc, maps, segment, None)? {
+            Some(next) => current = next,
+            None => return Ok(None),
+        }
+    }
+    Ok(Some(current))
+}
+
+/// Makes `dst` hold at `path` what `src`'s merged view holds there: a type-preserving
+/// copy (see `materialize_into`) if it exists, a delete otherwise. Replaces the whole
+/// subtree at `path`; the path must not be empty.
+pub(crate) fn copy_path(
+    src: &AutoCommit,
+    dst: &mut AutoCommit,
+    path: &[String],
+) -> Result<(), StoreError> {
+    let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?;
+
+    let Some(resolved) = resolve_at(src, path)? else {
+        return match delete_path(dst, path, false) {
+            Ok(()) | Err(StoreError::NotFound) => Ok(()),
+            Err(error) => Err(error),
+        };
+    };
+
+    let maps = ensure_maps(dst, parents)?;
+    delete_in_all(dst, &maps, leaf)?;
+    let mut trace = path.to_vec();
+    copy_resolved(
+        src,
+        &resolved,
+        dst,
+        &maps[0],
+        Slot::Key(leaf),
+        &mut trace,
+        &|_| false,
+    )
 }
 
 /// Deletes the value at `path`, optionally pruning ancestors that become empty.
diff --git a/src/sync.rs b/src/sync.rs
index d9f629e..07ed1a0 100644
--- a/src/sync.rs
+++ b/src/sync.rs
@@ -458,6 +458,12 @@ pub(crate) async fn run_change_publisher(state: AppState) {
                     return (heads, Vec::new(), Vec::new());
                 }
 
+                // After an epoch switch the previous heads belong to the old document:
+                // report the root as changed, which every stream is related to.
+                if !before.iter().all(|h| doc.get_change_by_hash(h).is_some()) {
+                    return (heads, vec![Vec::new()], Vec::new());
+                }
+
                 let mut paths = Vec::new();
                 let mut conflicts = Vec::new();
                 for patch in doc.diff(&before, &heads) {