turn compaction into a kind of remake

1 files changed, 47 insertions, 8 deletions

diff --git a/src/remake/delta.rs b/src/remake/delta.rs
index c0ba272..d318c00 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));