turn compaction into a kind of remake

9 files changed, 589 insertions, 72 deletions

diff --git a/src/remake/api.rs b/src/remake/api.rs
index 9b490f4..9339ab3 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -16,7 +16,8 @@
 //!   is in progress, without having to know its ID.
 //!
 //! Request body for POST (all optional):
-//! `{ "replayWindow": {"epochs": 1, "seconds": 604800},
+//! `{ "kind": "compact", "spec": null,
+//!    "replayWindow": {"epochs": 1, "seconds": 604800},
 //!    "archiveRetention": {"epochs": 2, "seconds": 2592000},
 //!    "settleSecs": 480 }`
 
@@ -40,7 +41,7 @@ use super::local::LocalStatus;
 use super::records::{
     self, unix_millis, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention,
 };
-use super::Host;
+use super::{kind, Host};
 use crate::document::store::StoreError;
 
 #[derive(Debug, Default, Deserialize)]
@@ -61,6 +62,10 @@ pub(crate) struct StartQuery {
 #[derive(Debug, Default, Deserialize)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
 struct StartRequest {
+    /// The remake kind; `compact` if absent.
+    kind: Option<String>,
+    /// Kind-specific part of the request.
+    spec: Option<JsonValue>,
     replay_window: Option<Retention>,
     archive_retention: Option<Retention>,
     settle_secs: Option<u64>,
@@ -154,11 +159,30 @@ async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> {
     }
 }
 
+/// `POST /admin/remakes`.
 pub(crate) async fn start<H: Host>(
     State(host): State<H>,
     Query(query): Query<StartQuery>,
     body: Bytes,
 ) -> Response {
+    start_kind(host, query, body, None).await
+}
+
+/// `POST /admin/compaction`: the original endpoint, for compacting remakes only.
+pub(crate) async fn start_compaction<H: Host>(
+    State(host): State<H>,
+    Query(query): Query<StartQuery>,
+    body: Bytes,
+) -> Response {
+    start_kind(host, query, body, Some(kind::DEFAULT_KIND)).await
+}
+
+async fn start_kind<H: Host>(
+    host: H,
+    query: StartQuery,
+    body: Bytes,
+    only: Option<&'static str>,
+) -> Response {
     if !host.is_node_ready().await {
         return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None);
     }
@@ -178,6 +202,32 @@ pub(crate) async fn start<H: Host>(
         }
     };
 
+    let kind_name = request.kind.as_deref().unwrap_or(kind::DEFAULT_KIND);
+    if only.is_some_and(|only| only != kind_name) {
+        return error_response(
+            StatusCode::BAD_REQUEST,
+            format!("this endpoint only starts {} remakes", kind::DEFAULT_KIND),
+            None,
+        );
+    }
+    let Some(kind) = kind::by_name(kind_name) else {
+        return error_response(
+            StatusCode::BAD_REQUEST,
+            format!("unknown remake kind {kind_name:?}"),
+            None,
+        );
+    };
+    let prepared = match kind.propose(request.spec.as_ref().unwrap_or(&JsonValue::Null)) {
+        Ok(prepared) => prepared,
+        Err(reason) => {
+            return error_response(
+                StatusCode::BAD_REQUEST,
+                format!("invalid {kind_name} spec: {reason}"),
+                None,
+            )
+        }
+    };
+
     let dormant_after = host.dormant_after();
     let settle = request
         .settle_secs
@@ -239,6 +289,9 @@ pub(crate) async fn start<H: Host>(
         proposed_at: now,
         settle_until: now + settle_ms,
         peers_considered,
+        kind: kind.name().to_owned(),
+        content: prepared.content,
+        translation: prepared.translation,
     };
 
     // From here on, nothing depends on the client: writing the proposal and starting the
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 22aa0f8..c8878cc 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -30,10 +30,11 @@ 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::kind::Translator;
+use super::local::{CatchUp, LocalStatus, Rejoin};
+use super::quarantine;
 use super::records::unix_millis;
 use super::snapshot::{self, EpochMetadata};
 use super::Host;
@@ -151,7 +152,8 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
 /// What to do with this node's own changes the new epoch lacks.
 enum Plan {
     Nothing,
-    Replay(Vec<Vec<String>>),
+    /// Replay these paths, carried through every remake since the node's epoch.
+    Replay(Vec<Vec<String>>, Translator),
     Quarantine {
         paths: Vec<Vec<String>>,
         reason: String,
@@ -234,31 +236,33 @@ async fn finish<H: Host>(host: &H) {
 
     match plan {
         Plan::Nothing => {}
-        Plan::Replay(paths) => {
+        Plan::Replay(paths, translator) => {
             let document = old.take().expect("old document");
             let (replayed, document) = host
                 .commit_doc(move |doc| {
-                    let stats = delta::replay_paths(&document, doc, &paths);
+                    let stats = delta::replay_paths(&document, doc, &paths, &translator);
                     Ok::<_, Infallible>((stats, document))
                 })
                 .await
                 .unwrap_or_else(|never| match never {});
             stats = replayed;
+            // Writes the remakes since had no place for.
+            quarantine_id = quarantine::add(
+                host,
+                previous,
+                target,
+                &document,
+                std::mem::take(&mut stats.quarantined),
+            );
             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.remake().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);
+            let document = old.take().expect("old document");
+            let entries = paths
+                .into_iter()
+                .map(|path| (path, reason.clone()))
+                .collect();
+            quarantine_id = quarantine::add(host, previous, target, &document, entries);
             keep_archive = false; // the quarantine holds the document
         }
     }
@@ -346,18 +350,38 @@ fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: E
     let paths = delta::changed_paths(&mut old, &missing.before, &old_heads);
     let gap = target.generation - previous.generation;
 
-    let plan = match &next {
-        None => Plan::Quarantine {
+    // Writes made in our epoch go through every remake since, in order.
+    let translator = (previous.generation + 1..=target.generation).try_fold(
+        Translator::default(),
+        |mut translator, generation| {
+            let epoch = snapshot::read_epoch_metadata(&copy, generation)?;
+            let kind = epoch.kind().ok_or_else(|| {
+                format!(
+                    "epoch generation {generation} was made by unknown remake kind {:?}",
+                    epoch.kind
+                )
+            })?;
+            translator.push(kind, epoch.translation);
+            Ok::<_, String>(translator)
+        },
+    );
+
+    let plan = match (&next, translator) {
+        (Some(_), Err(reason)) => Plan::Quarantine {
+            paths,
+            reason: format!("cannot translate {} local changes: {reason}", missing.own),
+        },
+        (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) => {
+        (Some(next), Ok(translator)) => {
             let away_ms = unix_millis().saturating_sub(next.proposed_at);
             if next.policy.replay_window.covers(gap, away_ms) {
-                Plan::Replay(paths)
+                Plan::Replay(paths, translator)
             } else {
                 Plan::Quarantine {
                     paths,
@@ -440,6 +464,9 @@ mod tests {
                 base_heads: &base,
                 policy,
                 proposed_at: 0,
+                kind: "compact",
+                content: &JsonValue::Null,
+                translation: &JsonValue::Null,
             },
         )
         .unwrap();
@@ -459,7 +486,7 @@ mod tests {
         let (old, cluster, target) = scenario(&json!({}));
         match prepare(cluster, old, Epoch::GENESIS, target) {
             Prepared::Ready {
-                plan: Plan::Replay(paths),
+                plan: Plan::Replay(paths, _),
                 own_missing,
                 ..
             } => {
diff --git a/src/remake/delta.rs b/src/remake/delta.rs
index b07c862..c161e2c 100644
--- a/src/remake/delta.rs
+++ b/src/remake/delta.rs
@@ -6,6 +6,10 @@
 //! there now (or a delete if nothing). Replayed values are fresh operations, so they win
 //! over what the snapshot holds at the same path.
 //!
+//! Where a path goes is the remake kind's decision ([`Translator`]): the same path
+//! (compaction), another path, nowhere (dropped), or quarantine. Protocol data under
+//! `_continuum` always keeps its path.
+//!
 //! Granularity:
 //! * Map entries are replayed per key, so changes to sibling keys made on different
 //!   nodes don't overwrite each other.
@@ -22,13 +26,28 @@ use std::collections::{BTreeMap, HashSet};
 use automerge::{ActorId, AutoCommit, ChangeHash, PatchAction, Prop};
 use tracing::warn;
 
+use super::kind::{Translation, Translator};
 use super::snapshot::{EPOCHS, META_ROOT, REMAKES};
 use crate::document::store::{self, StoreError};
 
-#[derive(Debug, Default, Clone, Copy)]
+#[derive(Debug, Default, Clone)]
 pub(crate) struct ReplayStats {
     pub(crate) replayed: usize,
     pub(crate) failed: usize,
+    /// Paths the remake declared obsolete.
+    pub(crate) dropped: usize,
+    /// Paths the remake had no place for, with the reason.
+    pub(crate) quarantined: Vec<(Vec<String>, String)>,
+}
+
+impl ReplayStats {
+    /// Adds another replay's numbers (e.g. late changes replayed under the lock).
+    pub(crate) fn absorb(&mut self, other: ReplayStats) {
+        self.replayed += other.replayed;
+        self.failed += other.failed;
+        self.dropped += other.dropped;
+        self.quarantined.extend(other.quarantined);
+    }
 }
 
 /// Replays into `dst` everything that changed in `src` between `before` and `after`,
@@ -38,13 +57,14 @@ pub(crate) fn replay(
     before: &[ChangeHash],
     after: &[ChangeHash],
     dst: &mut AutoCommit,
+    translator: &Translator,
 ) -> ReplayStats {
     if before == after {
         return ReplayStats::default();
     }
 
     let paths = changed_paths(src, before, after);
-    replay_paths(src, dst, &paths)
+    replay_paths(src, dst, &paths, translator)
 }
 
 /// Replays the state of `paths` (from [`changed_paths`]) from `src` into `dst`, as one
@@ -53,16 +73,23 @@ pub(crate) fn replay_paths(
     src: &AutoCommit,
     dst: &mut AutoCommit,
     paths: &[Vec<String>],
+    translator: &Translator,
 ) -> ReplayStats {
     let mut stats = ReplayStats::default();
     for path in paths {
-        replay_path(src, dst, path, &mut stats);
+        replay_path(src, dst, path, translator, &mut stats);
     }
     let _ = dst.commit();
     stats
 }
 
-fn replay_path(src: &AutoCommit, dst: &mut AutoCommit, path: &[String], stats: &mut ReplayStats) {
+fn replay_path(
+    src: &AutoCommit,
+    dst: &mut AutoCommit,
+    path: &[String],
+    translator: &Translator,
+    stats: &mut ReplayStats,
+) {
     if is_protocol_path(path) {
         return;
     }
@@ -77,12 +104,24 @@ fn replay_path(src: &AutoCommit, dst: &mut AutoCommit, path: &[String], stats: &
         for key in keys {
             let mut child = path.to_vec();
             child.push(key);
-            replay_path(src, dst, &child, stats);
+            replay_path(src, dst, &child, translator, stats);
         }
         return;
     }
 
-    match store::copy_path(src, dst, path) {
+    let copied = match translator.translate(path) {
+        Translation::Same => store::copy_path(src, dst, path),
+        Translation::To(to) => store::copy_path_to(src, path, dst, &to),
+        Translation::Drop => {
+            stats.dropped += 1;
+            return;
+        }
+        Translation::Quarantine(reason) => {
+            stats.quarantined.push((path.to_vec(), reason));
+            return;
+        }
+    };
+    match copied {
         Ok(()) => stats.replayed += 1,
         Err(error) => {
             log_failure(path, &error);
@@ -295,7 +334,7 @@ mod tests {
         old.commit();
         let after = old.get_heads();
 
-        let stats = replay(&mut old, &base, &after, &mut new);
+        let stats = replay(&mut old, &base, &after, &mut new, &Translator::default());
         assert_eq!(stats.failed, 0);
         assert_eq!(
             view(&new),
@@ -378,7 +417,7 @@ mod tests {
         old.commit();
         let after = old.get_heads();
 
-        replay(&mut old, &base, &after, &mut new);
+        replay(&mut old, &base, &after, &mut new, &Translator::default());
         let meta = &view(&new)[META_ROOT];
         assert_eq!(meta[EPOCHS]["1"]["generation"], json!(1));
         assert_eq!(meta["forgotten"]["pk"], json!(1));
diff --git a/src/remake/kind.rs b/src/remake/kind.rs
new file mode 100644
index 0000000..329fd09
--- /dev/null
+++ b/src/remake/kind.rs
@@ -0,0 +1,230 @@
+//! Kinds of remake: what makes one remake differ from another.
+//!
+//! Everything else about a remake (proposal, settle period, decision, switch, rejoin,
+//! replay window, quarantine, archive) is the same for every kind. A kind decides two
+//! things only:
+//!
+//! * the user content of the next epoch's initial document ([`RemakeKind::build`]),
+//!   which must be deterministic: every node builds it on its own and must produce the
+//!   same bytes, since its hash names the epoch;
+//! * where a write made in the old epoch after the base goes in the new one
+//!   ([`RemakeKind::translate`]).
+//!
+//! Protocol data under `_continuum` is not a kind's business: the generic part carries
+//! it forward and replays it unchanged.
+
+use automerge::AutoCommit;
+use serde_json::Value as JsonValue;
+
+use super::snapshot::META_ROOT;
+use crate::document::store::{self, StoreError};
+
+/// Name of the kind used when a request doesn't name one.
+pub(crate) const DEFAULT_KIND: &str = "compact";
+
+/// What the coordinator stores in the proposal for every node.
+#[derive(Debug, Default)]
+pub(crate) struct Prepared {
+    /// Input to [`RemakeKind::build`]; replicated with the proposal (old epoch only).
+    pub(crate) content: JsonValue,
+    /// Input to [`RemakeKind::translate`]; also kept in the new epoch's metadata, so
+    /// nodes that rejoin later can translate their writes.
+    pub(crate) translation: JsonValue,
+}
+
+/// Where a path written in the old epoch after the base goes.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum Translation {
+    /// To the same path (compaction).
+    Same,
+    /// To another path.
+    To(Vec<String>),
+    /// Nowhere automatically: quarantined for an operator, with the reason.
+    Quarantine(String),
+    /// Nowhere: the remake declared the path obsolete.
+    Drop,
+}
+
+pub(crate) trait RemakeKind: Send + Sync + 'static {
+    /// Stable name, stored in proposals and epoch metadata.
+    fn name(&self) -> &'static str;
+
+    /// Coordinator only: validates the kind-specific part of the request.
+    fn propose(&self, spec: &JsonValue) -> Result<Prepared, String>;
+
+    /// Every node: writes the user content of the next epoch into `target`, a fresh
+    /// document. `old_at_base` is the old epoch at the base. Must be deterministic and
+    /// must not write under `_continuum` (anything there is replaced).
+    fn build(
+        &self,
+        old_at_base: &AutoCommit,
+        content: &JsonValue,
+        target: &mut AutoCommit,
+    ) -> Result<(), StoreError>;
+
+    /// Every node: where a user path written after the base goes.
+    fn translate(&self, path: &[String], translation: &JsonValue) -> Translation;
+}
+
+/// The kind with this name.
+pub(crate) fn by_name(name: &str) -> Option<&'static dyn RemakeKind> {
+    match name {
+        "compact" => Some(&Compact),
+        _ => None,
+    }
+}
+
+/// Paths owned by the protocol rather than by any kind.
+pub(crate) fn is_reserved(path: &[String]) -> bool {
+    path.first().is_some_and(|root| root == META_ROOT)
+}
+
+/// Compaction: the new document has the old one's content at the base, without its
+/// history. Writes keep their paths.
+pub(crate) struct Compact;
+
+impl RemakeKind for Compact {
+    fn name(&self) -> &'static str {
+        "compact"
+    }
+
+    fn propose(&self, spec: &JsonValue) -> Result<Prepared, String> {
+        match spec {
+            JsonValue::Null => Ok(Prepared::default()),
+            JsonValue::Object(fields) if fields.is_empty() => Ok(Prepared::default()),
+            _ => Err("compaction takes no spec".to_owned()),
+        }
+    }
+
+    fn build(
+        &self,
+        old_at_base: &AutoCommit,
+        _content: &JsonValue,
+        target: &mut AutoCommit,
+    ) -> Result<(), StoreError> {
+        store::materialize_into(old_at_base, target, &is_reserved)
+    }
+
+    fn translate(&self, _path: &[String], _translation: &JsonValue) -> Translation {
+        Translation::Same
+    }
+}
+
+/// The translations of several consecutive remakes, applied in order: a node that
+/// rejoins several epochs later carries its writes through every remake in between.
+#[derive(Default, Clone)]
+pub(crate) struct Translator {
+    steps: Vec<(&'static dyn RemakeKind, JsonValue)>,
+}
+
+impl Translator {
+    /// One remake.
+    pub(crate) fn single(kind: &'static dyn RemakeKind, translation: JsonValue) -> Self {
+        Self {
+            steps: vec![(kind, translation)],
+        }
+    }
+
+    pub(crate) fn push(&mut self, kind: &'static dyn RemakeKind, translation: JsonValue) {
+        self.steps.push((kind, translation));
+    }
+
+    /// Where `path` ends up after all steps. Reserved paths always stay put.
+    pub(crate) fn translate(&self, path: &[String]) -> Translation {
+        if is_reserved(path) {
+            return Translation::Same;
+        }
+        let mut current = path.to_vec();
+        let mut moved = false;
+        for (kind, translation) in &self.steps {
+            match kind.translate(&current, translation) {
+                Translation::Same => {}
+                Translation::To(next) => {
+                    current = next;
+                    moved = true;
+                }
+                stop @ (Translation::Quarantine(_) | Translation::Drop) => return stop,
+            }
+        }
+        if moved {
+            Translation::To(current)
+        } else {
+            Translation::Same
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A test kind that moves `a/...` to `b/...`, drops `x`, quarantines the rest.
+    struct Rename;
+
+    impl RemakeKind for Rename {
+        fn name(&self) -> &'static str {
+            "rename-test"
+        }
+        fn propose(&self, _spec: &JsonValue) -> Result<Prepared, String> {
+            Ok(Prepared::default())
+        }
+        fn build(
+            &self,
+            _old: &AutoCommit,
+            _content: &JsonValue,
+            _target: &mut AutoCommit,
+        ) -> Result<(), StoreError> {
+            Ok(())
+        }
+        fn translate(&self, path: &[String], _translation: &JsonValue) -> Translation {
+            match path.split_first() {
+                Some((first, rest)) if first == "a" => {
+                    let mut to = vec!["b".to_owned()];
+                    to.extend_from_slice(rest);
+                    Translation::To(to)
+                }
+                Some((first, _)) if first == "b" => Translation::Same,
+                Some((first, _)) if first == "x" => Translation::Drop,
+                _ => Translation::Quarantine("no rule".to_owned()),
+            }
+        }
+    }
+
+    fn path(parts: &[&str]) -> Vec<String> {
+        parts.iter().map(|part| (*part).to_owned()).collect()
+    }
+
+    #[test]
+    fn compaction_keeps_paths() {
+        assert_eq!(
+            Translator::single(&Compact, JsonValue::Null).translate(&path(&["a", "b"])),
+            Translation::Same
+        );
+        assert!(Compact.propose(&JsonValue::Null).is_ok());
+        assert!(Compact.propose(&serde_json::json!({ "x": 1 })).is_err());
+    }
+
+    #[test]
+    fn steps_compose() {
+        static RENAME: Rename = Rename;
+        let mut translator = Translator::default();
+        translator.push(&RENAME, JsonValue::Null);
+        translator.push(&Compact, JsonValue::Null);
+        translator.push(&RENAME, JsonValue::Null); // b stays b
+
+        assert_eq!(
+            translator.translate(&path(&["a", "k"])),
+            Translation::To(path(&["b", "k"]))
+        );
+        assert_eq!(translator.translate(&path(&["x"])), Translation::Drop);
+        assert!(matches!(
+            translator.translate(&path(&["other"])),
+            Translation::Quarantine(_)
+        ));
+        // Protocol data is never translated.
+        assert_eq!(
+            translator.translate(&path(&[META_ROOT, "forgotten", "pk"])),
+            Translation::Same
+        );
+    }
+}
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index d971fcf..e31eadf 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -1,6 +1,8 @@
 //! Remakes: coordinated moves of the whole cluster to a new document, a new epoch.
 //! Compaction is one: the new document has the old one's content without its history.
 //!
+//! * `kind`: what differs between kinds of remake (building the new document, and
+//!   where writes made after the base go); compaction is the only kind so far.
 //! * `snapshot`: deterministic build of the next epoch's initial document.
 //! * `records`: the replicated remake records under `_continuum/remakes/<id>`.
 //! * `coordinator`: proposal, settle period and decision on the node that was asked.
@@ -19,6 +21,7 @@ mod api;
 mod catchup;
 mod coordinator;
 mod delta;
+mod kind;
 mod local;
 mod quarantine;
 mod records;
@@ -128,7 +131,7 @@ pub(crate) fn router<H: Host>() -> Router<H> {
                 .delete(api::abort_current::<H>),
         )
         // The original endpoint, kept as a shortcut for a compacting remake.
-        .route("/admin/compaction", post(api::start::<H>))
+        .route("/admin/compaction", post(api::start_compaction::<H>))
         .route(
             "/admin/remakes/{id}",
             get(api::get::<H>).delete(api::abort::<H>),
diff --git a/src/remake/quarantine.rs b/src/remake/quarantine.rs
index 5643549..deed8cd 100644
--- a/src/remake/quarantine.rs
+++ b/src/remake/quarantine.rs
@@ -14,6 +14,7 @@
 
 use std::convert::Infallible;
 
+use automerge::AutoCommit;
 use axum::{
     extract::{Path, State},
     http::{header, StatusCode},
@@ -22,12 +23,16 @@ use axum::{
 };
 use serde::Serialize;
 use serde_json::{json, Value as JsonValue};
-use tracing::info;
+use tracing::{info, warn};
+use uuid::Uuid;
 
 use super::api::error_response;
 use super::delta;
+use super::kind::Translator;
 use super::local::Quarantined;
+use super::records::unix_millis;
 use super::Host;
+use crate::document::replica::Epoch;
 use crate::document::store;
 
 #[derive(Debug, Serialize)]
@@ -58,6 +63,41 @@ fn not_found() -> Response {
     error_response(StatusCode::NOT_FOUND, "no such quarantine entry", None)
 }
 
+/// Puts writes a remake had no place for into quarantine, with `document` (the old
+/// epoch's) to read them from. Returns the entry's ID, or None if there is nothing.
+pub(super) fn add<H: Host>(
+    host: &H,
+    epoch: Epoch,
+    target: Epoch,
+    document: &AutoCommit,
+    quarantined: Vec<(Vec<String>, String)>,
+) -> Option<String> {
+    if quarantined.is_empty() {
+        return None;
+    }
+    let mut reasons: Vec<&str> = quarantined
+        .iter()
+        .map(|(_, reason)| reason.as_str())
+        .collect();
+    reasons.sort_unstable();
+    reasons.dedup();
+    let reason = reasons.join("; ");
+    let paths: Vec<Vec<String>> = quarantined.into_iter().map(|(path, _)| path).collect();
+
+    let id = Uuid::new_v4().to_string();
+    warn!(quarantine = %id, epoch = %epoch, paths = paths.len(), reason = %reason, "Quarantined writes the remake has no place for");
+    host.remake().add_quarantine(Quarantined {
+        id: id.clone(),
+        epoch,
+        target,
+        created_at: unix_millis(),
+        reason,
+        paths,
+        document: document.clone(),
+    });
+    Some(id)
+}
+
 pub(crate) async fn list<H: Host>(State(host): State<H>) -> Response {
     Json(host.remake().list_quarantine(summary)).into_response()
 }
@@ -129,7 +169,9 @@ pub(crate) async fn replay<H: Host>(State(host): State<H>, Path(id): Path<String
 
     let (stats, entry) = host
         .commit_doc(move |doc| {
-            let stats = delta::replay_paths(&entry.document, doc, &entry.paths);
+            // As they are: the operator decided they belong at their old paths.
+            let stats =
+                delta::replay_paths(&entry.document, doc, &entry.paths, &Translator::default());
             Ok::<_, Infallible>((stats, entry))
         })
         .await
diff --git a/src/remake/records.rs b/src/remake/records.rs
index 219fc9b..5e7841f 100644
--- a/src/remake/records.rs
+++ b/src/remake/records.rs
@@ -17,8 +17,10 @@
 
 use automerge::{AutoCommit, ChangeHash};
 use serde::{Deserialize, Serialize};
+use serde_json::Value as JsonValue;
 use tracing::warn;
 
+use super::kind::DEFAULT_KIND;
 use super::snapshot::{SnapshotRequest, META_ROOT, REMAKES};
 use crate::document::replica::Epoch;
 use crate::document::store::{self, StoreError};
@@ -123,6 +125,19 @@ pub(crate) struct Proposal {
     pub(crate) settle_until: u64,
     /// Peers that must stay active during the settle period.
     pub(crate) peers_considered: Vec<String>,
+    /// The remake kind (`compact` for records written before kinds existed).
+    #[serde(default = "default_kind")]
+    pub(crate) kind: String,
+    /// The kind's build input.
+    #[serde(default, skip_serializing_if = "JsonValue::is_null")]
+    pub(crate) content: JsonValue,
+    /// The kind's translation input.
+    #[serde(default, skip_serializing_if = "JsonValue::is_null")]
+    pub(crate) translation: JsonValue,
+}
+
+fn default_kind() -> String {
+    DEFAULT_KIND.to_owned()
 }
 
 impl Proposal {
@@ -204,6 +219,9 @@ impl Record {
             base_heads: self.proposal.base_change_hashes()?,
             policy: serde_json::to_value(&self.proposal.policy).ok()?,
             proposed_at: self.proposal.proposed_at,
+            kind: self.proposal.kind.clone(),
+            content: self.proposal.content.clone(),
+            translation: self.proposal.translation.clone(),
         })
     }
 
@@ -338,6 +356,9 @@ mod tests {
             proposed_at: settle_until.saturating_sub(1_000),
             settle_until,
             peers_considered: vec!["node-b".into()],
+            kind: DEFAULT_KIND.to_owned(),
+            content: JsonValue::Null,
+            translation: JsonValue::Null,
         }
     }
 
diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs
index da23fdf..2f2dfd6 100644
--- a/src/remake/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -28,6 +28,7 @@ use automerge::{transaction::CommitOptions, ActorId, AutoCommit, ChangeHash};
 use serde_json::{json, Map as JsonMap, Value as JsonValue};
 use sha2::{Digest, Sha256};
 
+use super::kind::{by_name, RemakeKind, Translator, DEFAULT_KIND};
 use super::records::{EpochRecord, Policy};
 use crate::document::replica::Epoch;
 use crate::document::store::{self, StoreError};
@@ -48,6 +49,12 @@ pub(crate) struct SnapshotParams<'a> {
     pub(crate) policy: &'a JsonValue,
     /// When the remake was proposed (unix ms); recorded for replay windows.
     pub(crate) proposed_at: u64,
+    /// Name of the remake kind, which builds the user content.
+    pub(crate) kind: &'a str,
+    /// The kind's build input, from the proposal.
+    pub(crate) content: &'a JsonValue,
+    /// The kind's translation input; recorded in the epoch metadata.
+    pub(crate) translation: &'a JsonValue,
 }
 
 pub(crate) struct Snapshot {
@@ -72,6 +79,8 @@ pub(crate) enum SnapshotError {
     Store(#[from] StoreError),
     #[error("snapshot task failed: {0}")]
     Task(String),
+    #[error("unknown remake kind {0:?}")]
+    UnknownKind(String),
 }
 
 /// Owned version of [`SnapshotParams`], for moving into a blocking task.
@@ -82,6 +91,9 @@ pub(crate) struct SnapshotRequest {
     pub(crate) base_heads: Vec<ChangeHash>,
     pub(crate) policy: JsonValue,
     pub(crate) proposed_at: u64,
+    pub(crate) kind: String,
+    pub(crate) content: JsonValue,
+    pub(crate) translation: JsonValue,
 }
 
 impl SnapshotRequest {
@@ -92,8 +104,16 @@ impl SnapshotRequest {
             base_heads: &self.base_heads,
             policy: &self.policy,
             proposed_at: self.proposed_at,
+            kind: &self.kind,
+            content: &self.content,
+            translation: &self.translation,
         }
     }
+
+    /// The kind's translation for writes after the base.
+    pub(crate) fn translator(&self) -> Option<Translator> {
+        by_name(&self.kind).map(|kind| Translator::single(kind, self.translation.clone()))
+    }
 }
 
 /// Builds the snapshot from `copy`, a clone of the live document, on the blocking pool.
@@ -121,6 +141,20 @@ pub(crate) struct EpochMetadata {
     pub(crate) policy: Policy,
     /// When the remake was proposed (unix ms): the point the cluster moved on.
     pub(crate) proposed_at: u64,
+    /// The remake kind and its translation input, for writes made before the epoch.
+    pub(crate) kind: String,
+    pub(crate) translation: JsonValue,
+}
+
+impl EpochMetadata {
+    /// The translation of writes from `previous` into this epoch, if the kind is known.
+    pub(crate) fn kind(&self) -> Option<&'static dyn RemakeKind> {
+        by_name(&self.kind)
+    }
+}
+
+fn default_kind() -> String {
+    DEFAULT_KIND.to_owned()
 }
 
 #[derive(serde::Deserialize)]
@@ -133,6 +167,10 @@ struct MetadataRecord {
     policy: Policy,
     #[serde(default)]
     proposed_at: u64,
+    #[serde(default = "default_kind")]
+    kind: String,
+    #[serde(default)]
+    translation: JsonValue,
 }
 
 fn parse_metadata(doc: &AutoCommit, generation: u64) -> Result<EpochMetadata, String> {
@@ -158,6 +196,8 @@ fn parse_metadata(doc: &AutoCommit, generation: u64) -> Result<EpochMetadata, St
         clock: record.clock,
         policy: record.policy,
         proposed_at: record.proposed_at,
+        kind: record.kind,
+        translation: record.translation,
     })
 }
 
@@ -217,10 +257,25 @@ pub(crate) fn build(
     let clock = clock_of(&mut base);
     let generation = params.previous.generation + 1;
 
+    let kind =
+        by_name(params.kind).ok_or_else(|| SnapshotError::UnknownKind(params.kind.to_owned()))?;
     let mut snapshot = AutoCommit::new().with_actor(snapshot_actor(params.remake_id, generation));
 
-    // Remake records describe the old epoch's protocol and stay behind.
-    store::materialize_into(&base, &mut snapshot, &is_remake_record)?;
+    // The user content is the kind's.
+    kind.build(&base, params.content, &mut snapshot)?;
+
+    // Protocol data carries forward, whatever the kind: forget records, earlier epochs'
+    // metadata (rejoining nodes need it). Remake records describe the old epoch's
+    // protocol and stay behind.
+    match store::delete_path(&mut snapshot, &[META_ROOT.to_owned()], false) {
+        Ok(()) | Err(StoreError::NotFound) => {}
+        Err(error) => return Err(error.into()),
+    }
+    let mut protocol_keys = store::keys_at(&base, &[META_ROOT.to_owned()])?;
+    protocol_keys.sort();
+    for key in protocol_keys.into_iter().filter(|key| key != REMAKES) {
+        store::copy_path(&base, &mut snapshot, &[META_ROOT.to_owned(), key])?;
+    }
 
     let mut base_heads: Vec<String> = params.base_heads.iter().map(|h| h.to_string()).collect();
     base_heads.sort();
@@ -236,6 +291,8 @@ pub(crate) fn build(
         "clock": clock,
         "policy": params.policy,
         "proposedAt": params.proposed_at,
+        "kind": kind.name(),
+        "translation": params.translation,
     }));
     store::write_path(&mut snapshot, &epoch_metadata_path(generation), &metadata)?;
 
@@ -277,10 +334,6 @@ fn snapshot_actor(remake_id: &str, generation: u64) -> ActorId {
     ActorId::from(&digest[..16])
 }
 
-fn is_remake_record(path: &[String]) -> bool {
-    matches!(path, [root, records] if root == META_ROOT && records == REMAKES)
-}
-
 /// 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 {
@@ -313,9 +366,14 @@ mod tests {
             base_heads: heads,
             policy,
             proposed_at: 0,
+            kind: "compact",
+            content: &NULL,
+            translation: &NULL,
         }
     }
 
+    static NULL: JsonValue = JsonValue::Null;
+
     fn snapshot_of(source: &mut AutoCommit) -> Snapshot {
         let heads = source.get_heads();
         let policy = json!({});
@@ -487,6 +545,7 @@ mod tests {
         let epoch = &meta[EPOCHS]["1"];
         assert_eq!(epoch["generation"], json!(1));
         assert_eq!(epoch["remakeId"], json!("c-1"));
+        assert_eq!(epoch["kind"], json!("compact"));
         assert_eq!(epoch["previous"]["generation"], json!(0));
         assert_eq!(epoch["policy"], policy);
         assert_eq!(epoch["baseHeads"].as_array().unwrap().len(), heads.len());
@@ -628,7 +687,13 @@ mod tests {
         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);
+        delta::replay(
+            &mut late,
+            &missing.before,
+            &heads,
+            &mut adopted,
+            &Translator::default(),
+        );
         assert_eq!(
             user_view(&adopted),
             json!({ "shared": "base", "new": 1, "mine": "late" })
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index b4ab829..a5031fc 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -26,7 +26,9 @@ use tracing::{error, info, warn};
 
 use super::catchup;
 use super::delta::{self, ReplayStats};
+use super::kind::Translator;
 use super::local::LocalStatus;
+use super::quarantine;
 use super::records::{self, unix_millis, EpochRecord, Record};
 use super::snapshot::{self, META_ROOT, REMAKES};
 use super::Host;
@@ -115,14 +117,14 @@ async fn check<H: Host>(host: &H) {
     }
 
     match switch(host, &id, &record).await {
-        Ok(stats) => {
+        Ok((stats, quarantine)) => {
             host.remake().set_status(
                 &id,
                 LocalStatus {
                     switched_at: Some(unix_millis()),
                     replayed_paths: Some(stats.replayed),
                     failed_paths: Some(stats.failed),
-                    quarantine: None,
+                    quarantine,
                     error: None,
                 },
             );
@@ -140,11 +142,21 @@ async fn check<H: Host>(host: &H) {
     }
 }
 
-async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplayStats, String> {
+/// Returns the replay numbers and the quarantine entry for writes the remake had no
+/// place for, if any.
+async fn switch<H: Host>(
+    host: &H,
+    id: &str,
+    record: &Record,
+) -> Result<(ReplayStats, Option<String>), String> {
     let (Some(expected), Some(request)) = (record.committed_epoch(), record.snapshot_request(id))
     else {
         return Err("unreadable remake record".to_owned());
     };
+    let translator = request
+        .translator()
+        .ok_or_else(|| format!("unknown remake kind {:?}", request.kind))?;
+    let late_translator = translator.clone();
     let previous = request.previous;
     let base = request.base_heads.clone();
     let retention = record.proposal.policy.archive_retention.clone();
@@ -165,7 +177,7 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
         .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(
+    let (new_doc, copy_heads, mut stats) = tokio::task::spawn_blocking(
         move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> {
             let mut copy = copy;
             let snapshot = match prebuilt {
@@ -191,7 +203,7 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
             }
 
             let heads = copy.get_heads();
-            let stats = delta::replay(&mut copy, &base, &heads, &mut doc);
+            let stats = delta::replay(&mut copy, &base, &heads, &mut doc, &translator);
             Ok((doc, heads, stats))
         },
     )
@@ -199,9 +211,18 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
     .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?;
+    let (old, late) = swap_in(host, expected, new_doc, copy_heads, late_translator).await?;
+    stats.absorb(late);
 
-    // 4. Keep the old document as configured.
+    // 4. Writes the remake had no place for wait for an operator (the old document
+    //    holds them); the old document is kept as configured either way.
+    let quarantine = quarantine::add(
+        host,
+        previous,
+        expected,
+        &old,
+        std::mem::take(&mut stats.quarantined),
+    );
     host.remake()
         .archive(previous, old, retention, expected.generation);
 
@@ -210,9 +231,11 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
         epoch = %expected,
         replayed = stats.replayed,
         failed = stats.failed,
+        dropped = stats.dropped,
+        quarantine = ?quarantine,
         "Switched to the new epoch"
     );
-    Ok(stats)
+    Ok((stats, quarantine))
 }
 
 /// Under the write lock: replays into `new_doc` whatever the live document received
@@ -223,26 +246,40 @@ pub(super) async fn swap_in<H: Host>(
     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
+    translator: Translator,
+) -> Result<(AutoCommit, ReplayStats), String> {
+    // The closure runs on the blocking pool; its replay numbers come back through this.
+    let late_stats = std::sync::Arc::new(std::sync::Mutex::new(ReplayStats::default()));
+    let late_out = std::sync::Arc::clone(&late_stats);
+    let old = 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, &translator);
+            if late.replayed + late.failed > 0 {
+                info!(
+                    replayed = late.replayed,
+                    failed = late.failed,
+                    "Replayed late changes"
+                );
+            }
+            *late_out
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner()) = late;
+            Ok(new_doc)
+        })
+        .await?;
+    let late = std::mem::take(
+        &mut *late_stats
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()),
+    );
+    Ok((old, late))
 }