turn compaction into a kind of remake

1 files changed, 72 insertions, 7 deletions

diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs
index 6c14ec4..ac2b3ec 100644
--- a/src/remake/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -28,6 +28,7 @@ use automerge::{ActorId, AutoCommit, ChangeHash, transaction::CommitOptions};
 use serde_json::{Map as JsonMap, Value as JsonValue, json};
 use sha2::{Digest, Sha256};
 
+use super::kind::{DEFAULT_KIND, RemakeKind, Translator, by_name};
 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" })