turn compaction into a kind of remake

1 files changed, 49 insertions, 22 deletions

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,
                 ..
             } => {