rename compaction to remake

10 files changed, 0 insertions, 3497 deletions

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