replace snapshots by quarantine & replay

13 files changed, 789 insertions, 600 deletions

diff --git a/src/compaction/api.rs b/src/compaction/api.rs
index e55d018..7f423da 100644
--- a/src/compaction/api.rs
+++ b/src/compaction/api.rs
@@ -85,7 +85,7 @@ enum StartError {
     Store(#[from] StoreError),
 }
 
-fn error_response(
+pub(super) fn error_response(
     status: StatusCode,
     message: impl Into<String>,
     details: Option<JsonValue>,
@@ -467,6 +467,7 @@ pub(crate) async fn epochs<H: Host>(State(host): State<H>) -> Response {
         "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
index 4694fbc..c71bb57 100644
--- a/src/compaction/catchup.rs
+++ b/src/compaction/catchup.rs
@@ -1,59 +1,67 @@
-//! Catching up with a newer epoch from a peer.
+//! 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, or
-//! one whose own build failed) can't learn the new epoch through ordinary sync. Instead,
-//! once an active peer advertises a newer generation, the node downloads that peer's
-//! document (`SnapshotRequest`/`SnapshotReply`), checks that it starts with the advertised
-//! snapshot change, and switches to it:
+//! 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
 //!
-//! * a node that wrote nothing the snapshot lacks (e.g. a brand-new node) adopts the
-//!   document as is, whatever the gap;
-//! * a node exactly one epoch behind replays its own changes the snapshot lacks, found
-//!   with the snapshot's per-actor clock;
-//! * a node further behind with such changes stays where it is and reports it; that is
-//!   the returning-node case (replay window or quarantine, step 7).
+//! 1. stops serving requests and keeps its document aside;
+//! 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.
 //!
-//! The document travels in chunks of `CHUNK_BYTES`, one request per chunk, so a large
-//! document neither needs one huge transport message nor one long timeout. The serving
-//! node saves its document once per download and serves every chunk from that copy; a
-//! SHA-256 digest of the whole copy is in every reply and checked at the end.
-//!
-//! Peers refuse explicitly (non-members, busy, other epoch), so the requester moves on
-//! to the next peer right away instead of timing out.
+//! 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::sync::Arc;
+use std::collections::BTreeMap;
+use std::convert::Infallible;
 use std::time::Duration;
 
-use automerge::{ActorId, AutoCommit};
-use sha2::{Digest, Sha256};
+use automerge::{AutoCommit, ChangeHash};
 use tokio::time::Instant;
-use tracing::{debug, error, info, warn};
+use tracing::{error, info, warn};
+use uuid::Uuid;
 
 use super::delta::{self, ReplayStats};
-use super::local::{CatchUp, LocalStatus, Served};
+use super::local::{CatchUp, LocalStatus, Quarantined, Rejoin};
 use super::records::unix_millis;
-use super::snapshot;
-use super::switch::swap_in;
+use super::snapshot::{self, EpochMetadata};
 use super::Host;
 use crate::replica::Epoch;
-use crate::transport::{ParsedMessage, SnapshotReply, SnapshotRequest};
 
-/// Pause before the next attempt (next peer) after a failed download.
-const RETRY: Duration = Duration::from_secs(10);
-/// Payload per reply.
-pub(crate) const CHUNK_BYTES: usize = 1024 * 1024;
-/// How long to wait for one chunk, and how often to ask for it.
-const CHUNK_TIMEOUT: Duration = Duration::from_secs(60);
-const CHUNK_ATTEMPTS: u32 = 3;
-/// A served copy older than this is replaced when a new download starts.
-const SERVED_FRESH_FOR: Duration = Duration::from_secs(60);
+/// 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.
+/// 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) {
-    let local = host.current_epoch();
+    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;
+    }
 
-    // Any active peer will do: it proved liveness recently, and whether it accepts us
-    // as a member is its decision (it refuses explicitly otherwise).
+    let local = host.current_epoch();
     let mut ahead: Vec<(String, Epoch)> = host
         .peer_summaries()
         .await
@@ -98,14 +106,10 @@ pub(super) async fn check<H: Host>(host: &H) {
     }
     let attempts = previous.map_or(0, |catch_up| catch_up.attempts) + 1;
 
-    // Rotate through the peers on the target epoch (sorted, so the rotation is stable).
-    let sources: Vec<&String> = ahead
-        .iter()
-        .filter(|(_, epoch)| *epoch == target)
-        .map(|(peer, _)| peer)
-        .collect();
-    let source = sources[(attempts as usize - 1) % sources.len()].clone();
+    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,
@@ -113,336 +117,376 @@ pub(super) async fn check<H: Host>(host: &H) {
         error: None,
     }));
 
-    info!(local = %local, target = %target, peer = %source, attempt = attempts, "Fetching newer epoch from peer");
-
-    let result = match download(host, &source, local, target).await {
-        Ok(document) => adopt(host, local, target, document).await,
-        Err(error) => Err(error),
+    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();
 
-    match result {
-        Ok(()) => host.compaction().set_catch_up(None),
-        Err(error) => {
-            warn!(target = %target, peer = %source, error = %error, "Catching up failed; will retry");
-            host.compaction().set_catch_up(Some(CatchUp {
-                target,
-                attempts,
-                last_attempt: Instant::now(),
-                error: Some(error),
-            }));
-        }
-    }
+    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,
+    },
 }
 
-/// Downloads `peer`'s document of `target`, chunk by chunk.
-async fn download<H: Host>(
-    host: &H,
-    peer: &str,
-    local: Epoch,
-    target: Epoch,
-) -> Result<Vec<u8>, String> {
-    let mut document: Vec<u8> = Vec::new();
-    let mut expected: Option<([u8; 32], u64)> = None;
-    let mut last_progress = 0;
+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;
 
-    loop {
-        let offset = document.len() as u64;
-        let reply = request_chunk(host, peer, local, target, offset).await?;
+    // 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;
 
-        if !reply.refused.is_empty() {
-            return Err(format!("peer refused: {}", reply.refused));
+    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;
         }
-        if reply.epoch != target {
-            return Err(format!("peer is on {} instead of {target}", reply.epoch));
+        Ok(Prepared::Failed(old, reason)) => {
+            let rejoin = Rejoin {
+                previous,
+                target,
+                old,
+                started,
+                was_ready,
+            };
+            return abandon(host, rejoin, reason).await;
         }
-        match expected {
-            None => {
-                info!(peer = %peer, document_bytes = reply.total, "Downloading document");
-                expected = Some((reply.digest, reply.total));
-            }
-            Some((digest, total)) if digest != reply.digest || total != reply.total => {
-                return Err("the peer's document changed during the download".to_owned());
+        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;
             }
-            Some(_) => {}
-        }
-        let total = reply.total;
-        if reply.data.is_empty() && offset < total {
-            return Err(format!("empty chunk at offset {offset} of {total}"));
-        }
-        document.extend_from_slice(&reply.data);
-
-        let percent = (document.len() as u64 * 100)
-            .checked_div(total)
-            .unwrap_or(100);
-        if percent >= last_progress + 10 {
-            last_progress = percent;
-            debug!(peer = %peer, percent, "Download progress");
+            return;
         }
+    };
 
-        if document.len() as u64 >= total {
-            break;
-        }
-    }
+    let mut stats = ReplayStats::default();
+    let mut quarantine_id = None;
+    let mut keep_archive = true;
+    let mut old = Some(old);
 
-    let (digest, total) = expected.expect("set by the first reply");
-    if document.len() as u64 != total || sha256(&document) != digest {
-        return Err("downloaded document does not match its digest".to_owned());
-    }
-    Ok(document)
-}
-
-async fn request_chunk<H: Host>(
-    host: &H,
-    peer: &str,
-    local: Epoch,
-    target: Epoch,
-    offset: u64,
-) -> Result<SnapshotReply, String> {
-    for attempt in 1..=CHUNK_ATTEMPTS {
-        let reply = host.compaction().await_reply(peer, target, offset);
-        let request = ParsedMessage::SnapshotRequest(SnapshotRequest {
-            epoch: local,
-            wanted: target,
-            offset,
-        });
-        if let Err(error) = host.send_message(peer, request).await {
-            host.compaction().stop_awaiting();
-            return Err(format!("sending the request failed: {error:#}"));
+    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);
         }
-
-        match tokio::time::timeout(CHUNK_TIMEOUT, reply).await {
-            Ok(Ok(reply)) => return Ok(reply),
-            Ok(Err(_)) => return Err("request superseded".to_owned()),
-            Err(_) => {
-                host.compaction().stop_awaiting();
-                debug!(peer = %peer, offset, attempt, "No reply for chunk; asking again");
-            }
+        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
         }
     }
-    Err(format!(
-        "no reply for the chunk at offset {offset} after {CHUNK_ATTEMPTS} attempts of {}s",
-        CHUNK_TIMEOUT.as_secs()
-    ))
-}
-
-/// Verifies the fetched document, carries over this node's own missing changes, and
-/// switches to it.
-async fn adopt<H: Host>(
-    host: &H,
-    local: Epoch,
-    target: Epoch,
-    document: Vec<u8>,
-) -> Result<(), String> {
-    // Copy under the read lock only; everything else happens without a lock.
-    let copy = host.read_doc(|doc| doc.clone()).await;
-    let bytes = document.len();
-
-    let prepared = tokio::task::spawn_blocking(move || -> Result<Prepared, String> {
-        let mut fetched =
-            AutoCommit::load(&document).map_err(|error| format!("invalid document: {error}"))?;
-        let metadata = snapshot::verify_document(&mut fetched, target)?;
-
-        let mut copy = copy;
-        let own = copy.get_actor().clone();
-        let copy_heads = copy.get_heads();
-        let had_history = !copy_heads.is_empty();
-        let missing = delta::missing_from_snapshot(&mut copy, &metadata.clock, &own);
 
-        let mut doc = fetched.with_actor(ActorId::random());
-        let stats = if missing.own == 0 {
-            ReplayStats::default()
-        } else if metadata.previous == local {
-            delta::replay(&mut copy, &missing.before, &copy_heads, &mut doc)
-        } else {
-            return Err(format!(
-                "{} local changes are not in {target}, and this node is more than one epoch behind (on {local}, {target} follows {}); not replaying them automatically",
-                missing.own, metadata.previous
-            ));
-        };
-
-        Ok(Prepared {
-            doc,
-            copy_heads,
-            stats,
-            own_missing: missing.own,
-            had_history,
-            metadata,
-        })
-    })
-    .await
-    .map_err(|error| format!("catch-up task failed: {error}"))??;
-
-    let Prepared {
-        doc,
-        copy_heads,
-        stats,
-        own_missing,
-        had_history,
-        metadata,
-    } = prepared;
-
-    let old = swap_in(host, target, doc, copy_heads).await?;
-
-    if had_history {
+    if let (true, true, Some(document)) = (had_history, keep_archive, old) {
         host.compaction().archive(
-            local,
-            old,
+            previous,
+            document,
             metadata.policy.archive_retention.clone(),
             target.generation,
         );
     }
 
-    // Visible in the compaction's report, if this node has its record.
     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;
+    }
 
     info!(
-        from = %local,
+        from = %previous,
         to = %target,
-        document_bytes = bytes,
         own_changes = own_missing,
         replayed = stats.replayed,
         failed = stats.failed,
-        "Caught up with newer epoch"
+        quarantine = ?quarantine_id,
+        secs = started.elapsed().as_secs(),
+        "Rejoined newer epoch"
     );
-    Ok(())
 }
 
-struct Prepared {
-    doc: AutoCommit,
-    copy_heads: Vec<automerge::ChangeHash>,
-    stats: ReplayStats,
-    own_missing: usize,
-    had_history: bool,
-    metadata: snapshot::EpochMetadata,
-}
+/// 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),
+    };
 
-/// Answers a peer's chunk request from a saved copy of this node's document, if this
-/// node is on the wanted epoch and the peer is a member of our cluster. Refuses
-/// explicitly otherwise.
-pub(crate) async fn serve_snapshot<H: Host>(host: H, peer: String, request: SnapshotRequest) {
-    let current = host.current_epoch();
-    let wanted = request.wanted;
+    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,
+        };
+    }
 
-    let reply = match served_copy(&host, &peer, current, wanted, request.offset).await {
-        Ok(served) => {
-            let start = (request.offset as usize).min(served.bytes.len());
-            let end = (start + CHUNK_BYTES).min(served.bytes.len());
-            if request.offset == 0 {
-                info!(peer = %peer, wanted = %wanted, document_bytes = served.bytes.len(), "Serving document");
-            }
-            SnapshotReply {
-                epoch: served.epoch,
-                wanted,
-                digest: served.digest,
-                total: served.bytes.len() as u64,
-                offset: request.offset,
-                data: served.bytes[start..end].to_vec(),
-                refused: String::new(),
-            }
-        }
-        Err(reason) => {
-            info!(peer = %peer, wanted = %wanted, reason = %reason, "Refusing document request");
-            SnapshotReply {
-                epoch: current,
-                wanted,
-                digest: [0; 32],
-                total: 0,
-                offset: request.offset,
-                data: Vec::new(),
-                refused: reason,
+    // 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
+                    ),
+                }
             }
         }
     };
 
-    if let Err(error) = host
-        .send_message(&peer, ParsedMessage::SnapshotReply(reply))
-        .await
-    {
-        warn!(peer = %peer, error = %format!("{error:#}"), "Sending document chunk failed");
+    Prepared::Ready {
+        old,
+        metadata,
+        plan,
+        own_missing: missing.own,
+        had_history: true,
     }
 }
 
-/// The saved document to serve chunks from: reused while a download is in progress,
-/// saved anew when a download starts and the copy is stale.
-async fn served_copy<H: Host>(
-    host: &H,
-    peer: &str,
-    current: Epoch,
-    wanted: Epoch,
-    offset: u64,
-) -> Result<Arc<Served>, String> {
-    if !host.is_cluster_member(peer).await {
-        return Err("not a member of this cluster (no completed handshake)".to_owned());
+/// 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 {}
     }
-    if current != wanted {
-        return Err(format!("this node is on {current}"));
+    if was_ready {
+        host.resume().await;
     }
 
-    if let Some(served) = host.compaction().served(wanted) {
-        if offset > 0 || served.saved_at.elapsed() < SERVED_FRESH_FOR {
-            return Ok(served);
-        }
-    } else if offset > 0 {
-        return Err("download expired; start again".to_owned());
-    }
+    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};
 
-    if !host.compaction().try_start_serving() {
-        return Err("busy saving the document; try again".to_owned());
+    /// 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)
     }
 
-    // Clone under the read lock, save without one. The history check guards against
-    // a switch between reading the epoch and cloning.
-    let mut doc = host.read_doc(|doc| doc.clone()).await;
-    let saved = tokio::task::spawn_blocking(move || {
-        if doc
-            .get_change_by_hash(&automerge::ChangeHash(wanted.id))
-            .is_some()
-        {
-            Some(doc.save())
-        } else {
-            None
+    #[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"),
         }
-    })
-    .await;
-    host.compaction().done_serving();
+    }
 
-    let bytes = match saved {
-        Ok(Some(bytes)) => bytes,
-        Ok(None) => return Err("this node switched epochs".to_owned()),
-        Err(error) => {
-            error!(error = %error, "Saving the document for a peer failed");
-            return Err("saving the document failed".to_owned());
+    #[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"),
         }
-    };
-
-    let served = Arc::new(Served {
-        epoch: wanted,
-        digest: sha256(&bytes),
-        bytes,
-        saved_at: Instant::now(),
-    });
-    host.compaction().set_served(Arc::clone(&served));
-    Ok(served)
-}
+    }
 
-fn sha256(bytes: &[u8]) -> [u8; 32] {
-    let mut digest = [0; 32];
-    digest.copy_from_slice(&Sha256::digest(bytes));
-    digest
-}
+    #[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,
+                ..
+            }
+        ));
+    }
 
-/// Hands a peer's chunk to the waiting download.
-pub(crate) fn receive_snapshot_reply<H: Host>(host: &H, peer: &str, reply: SnapshotReply) {
-    let offset = reply.offset;
-    if !host.compaction().deliver_reply(peer, reply) {
-        debug!(peer = %peer, offset, "Unexpected document chunk; dropped");
+    #[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/delta.rs b/src/compaction/delta.rs
index 533a7be..ba04799 100644
--- a/src/compaction/delta.rs
+++ b/src/compaction/delta.rs
@@ -39,15 +39,25 @@ pub(crate) fn replay(
     after: &[ChangeHash],
     dst: &mut AutoCommit,
 ) -> ReplayStats {
-    let mut stats = ReplayStats::default();
     if before == after {
-        return stats;
+        return ReplayStats::default();
     }
 
-    for path in changed_paths(src, before, after) {
-        replay_path(src, dst, &path, &mut stats);
-    }
+    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
 }
diff --git a/src/compaction/local.rs b/src/compaction/local.rs
index adf8764..c798a4c 100644
--- a/src/compaction/local.rs
+++ b/src/compaction/local.rs
@@ -1,30 +1,25 @@
 //! Per-node, in-memory compaction state: snapshots built ahead of the switch, switch
-//! progress for reports, and archived documents of previous epochs.
+//! 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 and rebuilds snapshots when needed.
+//! archive or quarantine, and rebuilds snapshots when needed.
 
 use std::collections::HashMap;
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Arc, Mutex, MutexGuard};
-use std::time::Duration;
+use std::sync::{Mutex, MutexGuard};
 
 use automerge::AutoCommit;
 use serde::Serialize;
-use tokio::sync::oneshot;
 use tokio::time::Instant;
 
 use super::records::{unix_millis, Retention};
 use super::snapshot::Snapshot;
 use crate::replica::Epoch;
-use crate::transport::SnapshotReply;
 
 #[derive(Default)]
 pub(crate) struct Compaction {
     // A std mutex: every access is short and never spans an await.
     inner: Mutex<Inner>,
-    /// Set while this node saves its document for a peer; one at a time.
-    serving: AtomicBool,
 }
 
 #[derive(Default)]
@@ -35,25 +30,14 @@ struct Inner {
     first_seen: HashMap<String, Instant>,
     status: HashMap<String, LocalStatus>,
     archive: Vec<Archived>,
-    /// The chunk this node is waiting for: (peer, wanted epoch, offset).
-    awaiting: Option<(String, Epoch, u64, oneshot::Sender<SnapshotReply>)>,
+    /// Last attempt to catch up with a newer epoch.
     catch_up: Option<CatchUp>,
-    /// The saved document chunks are served from (one download source at a time).
-    served: Option<Arc<Served>>,
+    /// A rejoin in progress: the node syncs an empty document of a newer epoch.
+    rejoin: Option<Rejoin>,
+    quarantine: Vec<Quarantined>,
 }
 
-/// A saved copy of this node's document, served to downloading peers.
-pub(crate) struct Served {
-    pub(crate) epoch: Epoch,
-    pub(crate) bytes: Vec<u8>,
-    pub(crate) digest: [u8; 32],
-    pub(crate) saved_at: Instant,
-}
-
-/// A served copy is dropped this long after it was saved.
-const SERVED_KEPT_FOR: Duration = Duration::from_secs(15 * 60);
-
-/// Fetching a newer epoch from a peer.
+/// Catching up with a newer epoch: attempts and the last failure.
 #[derive(Clone, Debug)]
 pub(crate) struct CatchUp {
     pub(crate) target: Epoch,
@@ -62,11 +46,38 @@ pub(crate) struct CatchUp {
     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>,
 }
@@ -81,6 +92,9 @@ pub(crate) struct LocalStatus {
     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>,
@@ -88,7 +102,7 @@ pub(crate) struct LocalStatus {
 
 struct Archived {
     epoch: Epoch,
-    // Kept for later steps (serving or auditing old epochs); not read yet.
+    // Kept for auditing and recovery by hand; not read by the protocol.
     #[allow(dead_code)]
     doc: AutoCommit,
     archived_at: u64,
@@ -154,88 +168,79 @@ impl Compaction {
         prune(&mut inner.archive, current_generation);
     }
 
-    /// Drops archives past their retention and a served copy nobody needs any more.
     pub(crate) fn prune_archive(&self, current_generation: u64) {
-        let mut inner = self.lock();
-        prune(&mut inner.archive, current_generation);
-        if inner
-            .served
-            .as_ref()
-            .is_some_and(|served| served.saved_at.elapsed() > SERVED_KEPT_FOR)
-        {
-            inner.served = None;
-        }
+        prune(&mut self.lock().archive, current_generation);
     }
 
-    /// Registers interest in one chunk from `peer`, replacing any earlier request.
-    pub(crate) fn await_reply(
-        &self,
-        peer: &str,
-        wanted: Epoch,
-        offset: u64,
-    ) -> oneshot::Receiver<SnapshotReply> {
-        let (sender, receiver) = oneshot::channel();
-        self.lock().awaiting = Some((peer.to_owned(), wanted, offset, sender));
-        receiver
+    pub(crate) fn catch_up(&self) -> Option<CatchUp> {
+        self.lock().catch_up.clone()
     }
 
-    /// Hands a reply to the waiting download. False unless it answers exactly the
-    /// pending request (late replies to earlier attempts are dropped).
-    pub(crate) fn deliver_reply(&self, peer: &str, reply: SnapshotReply) -> bool {
-        let mut inner = self.lock();
-        match inner.awaiting.take() {
-            Some((expected, wanted, offset, sender))
-                if expected == peer && wanted == reply.wanted && offset == reply.offset =>
-            {
-                sender.send(reply).is_ok()
-            }
-            other => {
-                inner.awaiting = other;
-                false
-            }
-        }
+    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(),
+        })
     }
 
-    /// The copy served for `epoch`, if any.
-    pub(crate) fn served(&self, epoch: Epoch) -> Option<Arc<Served>> {
+    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()
-            .served
+            .rejoin
             .as_ref()
-            .filter(|served| served.epoch == epoch)
-            .cloned()
+            .map(|rejoin| (rejoin.target, rejoin.started))
     }
 
-    pub(crate) fn set_served(&self, served: Arc<Served>) {
-        self.lock().served = Some(served);
+    pub(crate) fn start_rejoin(&self, rejoin: Rejoin) {
+        self.lock().rejoin = Some(rejoin);
     }
 
-    pub(crate) fn stop_awaiting(&self) {
-        self.lock().awaiting = None;
+    pub(crate) fn take_rejoin(&self) -> Option<Rejoin> {
+        self.lock().rejoin.take()
     }
 
-    /// Claims the right to serve one snapshot; release with [`Compaction::done_serving`].
-    pub(crate) fn try_start_serving(&self) -> bool {
-        !self.serving.swap(true, Ordering::AcqRel)
+    pub(crate) fn add_quarantine(&self, quarantined: Quarantined) {
+        self.lock().quarantine.push(quarantined);
     }
 
-    pub(crate) fn done_serving(&self) {
-        self.serving.store(false, Ordering::Release);
+    /// 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 catch_up(&self) -> Option<CatchUp> {
-        self.lock().catch_up.clone()
+    pub(crate) fn list_quarantine<T>(&self, f: impl Fn(&Quarantined) -> T) -> Vec<T> {
+        self.lock().quarantine.iter().map(f).collect()
     }
 
-    pub(crate) fn set_catch_up(&self, catch_up: Option<CatchUp>) {
-        self.lock().catch_up = catch_up;
+    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))
     }
 
-    pub(crate) fn catch_up_report(&self) -> Option<CatchUpReport> {
-        self.lock().catch_up.as_ref().map(|catch_up| CatchUpReport {
-            target: catch_up.target.to_string(),
-            attempts: catch_up.attempts,
-            error: catch_up.error.clone(),
-        })
+    /// 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> {
@@ -255,13 +260,10 @@ impl Compaction {
 fn prune(archive: &mut Vec<Archived>, current_generation: u64) {
     let now = unix_millis();
     archive.retain(|archived| {
-        let within_epochs = archived.retention.epochs.is_some_and(|epochs| {
-            current_generation.saturating_sub(archived.epoch.generation) <= epochs
-        });
-        let within_time = archived.retention.seconds.is_some_and(|seconds| {
-            now.saturating_sub(archived.archived_at) <= seconds.saturating_mul(1000)
-        });
-        within_epochs || within_time
+        archived.retention.covers(
+            current_generation.saturating_sub(archived.epoch.generation),
+            now.saturating_sub(archived.archived_at),
+        )
     });
 }
 
diff --git a/src/compaction/mod.rs b/src/compaction/mod.rs
index e97d9c8..4685b86 100644
--- a/src/compaction/mod.rs
+++ b/src/compaction/mod.rs
@@ -4,7 +4,9 @@
 //! * `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`: fetching a newer epoch from a peer (late, new or failed nodes).
+//! * `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.
@@ -17,11 +19,11 @@ mod catchup;
 mod coordinator;
 mod delta;
 mod local;
+mod quarantine;
 mod records;
 pub(crate) mod snapshot;
 mod switch;
 
-pub(crate) use catchup::{receive_snapshot_reply, serve_snapshot};
 pub(crate) use local::Compaction;
 pub(crate) use switch::run_worker;
 
@@ -38,7 +40,6 @@ use tokio::sync::broadcast;
 
 use crate::replica::Epoch;
 use crate::sync::ChangeSet;
-use crate::transport::ParsedMessage;
 
 /// What compaction needs from the node. Implemented by `AppState`.
 ///
@@ -61,15 +62,18 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
     /// Known (not forgotten) peers.
     fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send;
 
-    /// True if `peer` completed the handshake as a member of our cluster.
-    fn is_cluster_member(&self, peer: &str) -> impl Future<Output = bool> + 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;
 
-    /// Sends a message to a peer over the transport.
-    fn send_message(
-        &self,
-        peer: &str,
-        message: ParsedMessage,
-    ) -> impl Future<Output = anyhow::Result<()>> + Send;
+    /// Serves requests again after [`Host::suspend`].
+    fn resume(&self) -> impl Future<Output = ()> + Send;
+
+    /// 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>>;
@@ -130,4 +134,17 @@ pub(crate) fn router<H: Host>() -> Router<H> {
             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
new file mode 100644
index 0000000..da39ec0
--- /dev/null
+++ b/src/compaction/quarantine.rs
@@ -0,0 +1,160 @@
+//! `/admin/quarantine`: this node's own changes that were not replayed when it rejoined
+//! a newer epoch (outside the replay window, or impossible to tell apart).
+//!
+//! * `GET /admin/quarantine` lists the entries.
+//! * `GET /admin/quarantine/{id}` shows the writes: every changed path with the value
+//!   the old document's merged view holds there (`deleted` if none).
+//! * `GET /admin/quarantine/{id}/document` downloads the old document
+//!   (`AutoCommit::save()`), for inspection with other tools.
+//! * `POST /admin/quarantine/{id}/replay` writes them into the current epoch anyway (as
+//!   fresh writes, so they win over what is there) and removes the entry.
+//! * `DELETE /admin/quarantine/{id}` discards the entry.
+//!
+//! Quarantine lives in memory on the node that holds it.
+
+use std::convert::Infallible;
+
+use axum::{
+    extract::{Path, State},
+    http::{header, StatusCode},
+    response::{IntoResponse, Response},
+    Json,
+};
+use serde::Serialize;
+use serde_json::{json, Value as JsonValue};
+use tracing::info;
+
+use super::api::error_response;
+use super::delta;
+use super::local::Quarantined;
+use super::Host;
+use crate::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
index fbb9fb9..ccd52c3 100644
--- a/src/compaction/records.rs
+++ b/src/compaction/records.rs
@@ -71,6 +71,17 @@ pub(crate) struct Retention {
     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 {
@@ -192,6 +203,7 @@ impl Record {
             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,
         })
     }
 
@@ -297,6 +309,22 @@ fn decode_hex32(hex: &str) -> Option<[u8; 32]> {
 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(),
diff --git a/src/compaction/snapshot.rs b/src/compaction/snapshot.rs
index 86caba9..0a89e23 100644
--- a/src/compaction/snapshot.rs
+++ b/src/compaction/snapshot.rs
@@ -46,6 +46,8 @@ pub(crate) struct SnapshotParams<'a> {
     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 {
@@ -79,6 +81,7 @@ pub(crate) struct SnapshotRequest {
     pub(crate) previous: Epoch,
     pub(crate) base_heads: Vec<ChangeHash>,
     pub(crate) policy: JsonValue,
+    pub(crate) proposed_at: u64,
 }
 
 impl SnapshotRequest {
@@ -88,6 +91,7 @@ impl SnapshotRequest {
             previous: self.previous,
             base_heads: &self.base_heads,
             policy: &self.policy,
+            proposed_at: self.proposed_at,
         }
     }
 }
@@ -107,13 +111,16 @@ pub(crate) async fn build_detached(
     .map_err(|error| SnapshotError::Task(error.to_string()))?
 }
 
-/// The protocol's view of an epoch, read at its snapshot change.
+/// 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)]
@@ -124,6 +131,34 @@ struct MetadataRecord {
     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
@@ -144,28 +179,17 @@ pub(crate) fn verify_document(doc: &mut AutoCommit, epoch: Epoch) -> Result<Epoc
     }
 
     let at_snapshot = doc.fork_at(&[root]).map_err(|error| error.to_string())?;
-    let value = store::read_path(&at_snapshot, &epoch_metadata_path(epoch.generation))
-        .map_err(|error| error.to_string())?
-        .ok_or_else(|| format!("no epoch metadata for generation {}", epoch.generation))?;
-    let record: MetadataRecord = serde_json::from_value(value)
-        .map_err(|error| format!("invalid epoch metadata: {error}"))?;
-    if record.generation != epoch.generation {
-        return Err(format!(
-            "epoch metadata names generation {} instead of {}",
-            record.generation, epoch.generation
-        ));
-    }
-    let previous = record
-        .previous
-        .to_epoch()
-        .ok_or_else(|| "invalid previous epoch in metadata".to_owned())?;
+    parse_metadata(&at_snapshot, epoch.generation)
+}
 
-    Ok(EpochMetadata {
-        compaction_id: record.compaction_id,
-        previous,
-        clock: record.clock,
-        policy: record.policy,
-    })
+/// 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.
@@ -212,6 +236,7 @@ pub(crate) fn build(
         "baseHeads": base_heads,
         "clock": clock,
         "policy": params.policy,
+        "proposedAt": params.proposed_at,
     }));
     store::write_path(&mut snapshot, &epoch_metadata_path(generation), &metadata)?;
 
@@ -288,6 +313,7 @@ mod tests {
             previous: Epoch::GENESIS,
             base_heads: heads,
             policy,
+            proposed_at: 0,
         }
     }
 
diff --git a/src/compaction/switch.rs b/src/compaction/switch.rs
index 4093e7d..24bcee0 100644
--- a/src/compaction/switch.rs
+++ b/src/compaction/switch.rs
@@ -16,7 +16,7 @@
 //! during steps 1-2.
 //!
 //! A node whose build doesn't match the committed epoch, or that never saw the commit,
-//! fetches the new epoch from a peer instead (`catchup`).
+//! rejoins the new epoch from peers instead (`catchup`).
 
 use std::time::Duration;
 
@@ -70,6 +70,12 @@ 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);
 
@@ -116,6 +122,7 @@ async fn check<H: Host>(host: &H) {
                     switched_at: Some(unix_millis()),
                     replayed_paths: Some(stats.replayed),
                     failed_paths: Some(stats.failed),
+                    quarantine: None,
                     error: None,
                 },
             );
diff --git a/src/mycelium.rs b/src/mycelium.rs
index 96b9e6d..e623659 100644
--- a/src/mycelium.rs
+++ b/src/mycelium.rs
@@ -19,9 +19,7 @@ use url::Url;
 
 use crate::transport::EncodedMessage;
 use crate::transport::Keepalive;
-use crate::transport::{
-    Epoch, HelloReply, HelloRequest, SnapshotReply, SnapshotRequest, SyncEnvelope,
-};
+use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope};
 use crate::transport::{MessageHandler, ParsedMessage, Transport};
 
 const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;
@@ -32,8 +30,6 @@ const TOPIC_HELLO_REPLY: &str = "continuum.v1.hello-reply";
 const TOPIC_SYNC: &str = "continuum.v1.sync";
 const TOPIC_PING: &str = "continuum.v1.ping";
 const TOPIC_PONG: &str = "continuum.v1.pong";
-const TOPIC_SNAPSHOT_REQUEST: &str = "continuum.v1.snapshot-request";
-const TOPIC_SNAPSHOT_REPLY: &str = "continuum.v1.snapshot-reply";
 
 const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25);
 
@@ -715,12 +711,6 @@ fn decode_payload(topic: &str, payload: &str) -> Result<ParsedMessage, Transport
         }
         TOPIC_PING => Ok(ParsedMessage::Ping(bytes_to_keepalive(&bytes)?)),
         TOPIC_PONG => Ok(ParsedMessage::Pong(bytes_to_keepalive(&bytes)?)),
-        TOPIC_SNAPSHOT_REQUEST => Ok(ParsedMessage::SnapshotRequest(bytes_to_snapshot_request(
-            &bytes,
-        )?)),
-        TOPIC_SNAPSHOT_REPLY => Ok(ParsedMessage::SnapshotReply(bytes_to_snapshot_reply(
-            &bytes,
-        )?)),
         t => Err(TransportError::UnknownTopic(t.to_string())),
     }
 }
@@ -748,10 +738,6 @@ fn format_message(
         ParsedMessage::Sync(env) => (TOPIC_SYNC, sync_envelope_to_bytes(env)?),
         ParsedMessage::Ping(k) => (TOPIC_PING, keepalive_to_bytes(k)?),
         ParsedMessage::Pong(k) => (TOPIC_PONG, keepalive_to_bytes(k)?),
-        ParsedMessage::SnapshotRequest(req) => {
-            (TOPIC_SNAPSHOT_REQUEST, snapshot_request_to_bytes(req)?)
-        }
-        ParsedMessage::SnapshotReply(res) => (TOPIC_SNAPSHOT_REPLY, snapshot_reply_to_bytes(res)?),
     };
     Ok(OutboundMessage {
         dst: Destination {
@@ -929,72 +915,6 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr
     })
 }
 
-// --- Snapshot transfer ---
-#[derive(Serialize, Deserialize)]
-struct SnapshotRequestShadow {
-    epoch: EpochShadow,
-    wanted: EpochShadow,
-    offset: u64,
-}
-
-#[derive(Serialize, Deserialize)]
-struct SnapshotReplyShadow {
-    epoch: EpochShadow,
-    wanted: EpochShadow,
-    digest: ByteBuf,
-    total: u64,
-    offset: u64,
-    data: ByteBuf,
-    refused: String,
-}
-
-fn snapshot_request_to_bytes(msg: &SnapshotRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
-    rmp_serde::to_vec(&SnapshotRequestShadow {
-        epoch: epoch_to_shadow(&msg.epoch),
-        wanted: epoch_to_shadow(&msg.wanted),
-        offset: msg.offset,
-    })
-}
-
-fn bytes_to_snapshot_request(bytes: &[u8]) -> Result<SnapshotRequest, TransportError> {
-    let shadow: SnapshotRequestShadow = rmp_serde::from_slice(bytes)?;
-    Ok(SnapshotRequest {
-        epoch: epoch_from_shadow(shadow.epoch)?,
-        wanted: epoch_from_shadow(shadow.wanted)?,
-        offset: shadow.offset,
-    })
-}
-
-fn snapshot_reply_to_bytes(msg: &SnapshotReply) -> Result<Vec<u8>, rmp_serde::encode::Error> {
-    rmp_serde::to_vec(&SnapshotReplyShadow {
-        epoch: epoch_to_shadow(&msg.epoch),
-        wanted: epoch_to_shadow(&msg.wanted),
-        digest: ByteBuf::from(msg.digest.to_vec()),
-        total: msg.total,
-        offset: msg.offset,
-        data: ByteBuf::from(msg.data.clone()),
-        refused: msg.refused.clone(),
-    })
-}
-
-fn bytes_to_snapshot_reply(bytes: &[u8]) -> Result<SnapshotReply, TransportError> {
-    let shadow: SnapshotReplyShadow = rmp_serde::from_slice(bytes)?;
-    let digest = shadow
-        .digest
-        .as_slice()
-        .try_into()
-        .map_err(|_| TransportError::InvalidPayload("digest must be 32 bytes".into()))?;
-    Ok(SnapshotReply {
-        epoch: epoch_from_shadow(shadow.epoch)?,
-        wanted: epoch_from_shadow(shadow.wanted)?,
-        digest,
-        total: shadow.total,
-        offset: shadow.offset,
-        data: shadow.data.into_vec(),
-        refused: shadow.refused,
-    })
-}
-
 #[derive(Error, Debug)]
 pub enum TransportError {
     #[error("Mycelium delivered a message without a sender public key")]
diff --git a/src/peer.rs b/src/peer.rs
index 50460b0..54ba0a4 100644
--- a/src/peer.rs
+++ b/src/peer.rs
@@ -537,13 +537,22 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool
                         .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
                 }
             }
-            other => {
+            // Expected after a compaction: either side switches or catches up.
+            EpochRelation::PeerAhead | EpochRelation::PeerBehind => {
+                info!(
+                    peer = %peer,
+                    local_epoch = %local,
+                    peer_epoch = %remote,
+                    relation = ?relation,
+                    "Peer is on another epoch; not syncing with it until both are on the same"
+                );
+            }
+            EpochRelation::Diverged => {
                 warn!(
                     peer = %peer,
                     local_epoch = %local,
                     peer_epoch = %remote,
-                    relation = ?other,
-                    "Peer is on a different epoch; not syncing with it"
+                    "Peer is on a different epoch of the same generation; not syncing with it"
                 );
             }
         }
@@ -663,17 +672,6 @@ pub(crate) async fn handle_message(
             let local = local_keepalive(state).await;
             heal_if_diverged(state, &peer, &local, &remote).await;
         }
-        transport::ParsedMessage::SnapshotRequest(request) => {
-            // Saving a large document takes a while; don't stall the inbox.
-            tokio::spawn(crate::compaction::serve_snapshot(
-                state.clone(),
-                peer.clone(),
-                request,
-            ));
-        }
-        transport::ParsedMessage::SnapshotReply(reply) => {
-            crate::compaction::receive_snapshot_reply(state, &peer, reply);
-        }
     }
 
     Ok(())
diff --git a/src/state.rs b/src/state.rs
index f9da5e0..81663f3 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -606,25 +606,45 @@ impl Host for AppState {
         }
     }
 
-    fn is_cluster_member(&self, peer: &str) -> impl Future<Output = bool> + Send {
+    fn suspend(&self) -> impl Future<Output = bool> + Send {
         let state = self.clone();
-        let peer = peer.to_owned();
         async move {
-            let Some(ours) = state.cluster_id().await else {
+            let mut lifecycle = state.lifecycle.write().await;
+            let NodeLifecycle::Ready { cluster_id } = &*lifecycle else {
                 return false;
             };
-            crate::peer::peer_cluster(&state, &peer).await.as_deref() == Some(ours.as_str())
+            let cluster_id = cluster_id.clone();
+            debug!(cluster_id = %cluster_id, "Node suspended (rejoining)");
+            *lifecycle = NodeLifecycle::Joining {
+                discovered_cluster_id: Some(cluster_id),
+            };
+            true
         }
     }
 
-    fn send_message(
-        &self,
-        peer: &str,
-        message: transport::ParsedMessage,
-    ) -> impl Future<Output = anyhow::Result<()>> + Send {
+    fn resume(&self) -> impl Future<Output = ()> + Send {
         let state = self.clone();
-        let peer = peer.to_owned();
-        async move { state.transport.send_message(&peer, &message).await }
+        async move {
+            let mut lifecycle = state.lifecycle.write().await;
+            if let NodeLifecycle::Joining {
+                discovered_cluster_id: Some(cluster_id),
+            } = &*lifecycle
+            {
+                let cluster_id = cluster_id.clone();
+                debug!(cluster_id = %cluster_id, "Node resumed");
+                *lifecycle = NodeLifecycle::Ready { cluster_id };
+            }
+        }
+    }
+
+    fn initial_sync_done(&self) -> bool {
+        self.initial_sync_done
+            .load(std::sync::atomic::Ordering::Acquire)
+    }
+
+    fn reset_initial_sync(&self) {
+        self.initial_sync_done
+            .store(false, std::sync::atomic::Ordering::Release);
     }
 
     fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>> {
diff --git a/src/transport.rs b/src/transport.rs
index 51faf16..9cfb026 100644
--- a/src/transport.rs
+++ b/src/transport.rs
@@ -23,9 +23,6 @@ pub enum ParsedMessage {
     Sync(SyncEnvelope),
     Ping(Keepalive),
     Pong(Keepalive),
-    /// Asks a peer for its document of a newer epoch (late or new nodes).
-    SnapshotRequest(SnapshotRequest),
-    SnapshotReply(SnapshotReply),
 }
 
 impl ParsedMessage {
@@ -36,8 +33,6 @@ impl ParsedMessage {
             ParsedMessage::HelloReply(reply) => reply.epoch,
             ParsedMessage::Sync(envelope) => envelope.epoch,
             ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch,
-            ParsedMessage::SnapshotRequest(request) => request.epoch,
-            ParsedMessage::SnapshotReply(reply) => reply.epoch,
         }
     }
 }
@@ -62,45 +57,6 @@ pub struct SyncEnvelope {
     pub message: Message,
 }
 
-#[derive(Debug, Clone)]
-pub struct SnapshotRequest {
-    /// The requester's epoch.
-    pub epoch: Epoch,
-    /// The epoch whose document is wanted.
-    pub wanted: Epoch,
-    /// Byte offset of the wanted chunk in the saved document.
-    pub offset: u64,
-}
-
-/// One chunk of the sender's saved document, or a refusal.
-#[derive(Clone)]
-pub struct SnapshotReply {
-    /// The sender's epoch; equals `wanted` unless refused.
-    pub epoch: Epoch,
-    pub wanted: Epoch,
-    /// SHA-256 of the whole saved document; identifies the copy chunks come from.
-    pub digest: [u8; 32],
-    /// Length of the whole saved document.
-    pub total: u64,
-    pub offset: u64,
-    pub data: Vec<u8>,
-    /// Non-empty if the sender won't serve the document, with the reason.
-    pub refused: String,
-}
-
-impl std::fmt::Debug for SnapshotReply {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("SnapshotReply")
-            .field("epoch", &self.epoch)
-            .field("wanted", &self.wanted)
-            .field("total", &self.total)
-            .field("offset", &self.offset)
-            .field("data_bytes", &self.data.len())
-            .field("refused", &self.refused)
-            .finish()
-    }
-}
-
 /// A message serialized for the wire. Cloning is cheap (shared buffer).
 #[derive(Clone, Debug)]
 pub struct EncodedMessage {