add epoch switch

1 files changed, 257 insertions, 0 deletions

diff --git a/src/compaction/delta.rs b/src/compaction/delta.rs
new file mode 100644
index 0000000..3bc9456
--- /dev/null
+++ b/src/compaction/delta.rs
@@ -0,0 +1,257 @@
+//! Replaying a node's changes after the base into the new epoch.
+//!
+//! Changes from the old epoch can't be applied to the new document (their history is
+//! unrelated), so they are replayed as state: for every path that changed between two
+//! sets of heads, the new document gets whatever the old document's merged view holds
+//! there now (or a delete if nothing). Replayed values are fresh operations, so they win
+//! over what the snapshot holds at the same path.
+//!
+//! Granularity:
+//! * Map entries are replayed per key, so changes to sibling keys made on different
+//!   nodes don't overwrite each other.
+//! * Lists and text are replayed whole: concurrent edits of the same list or text on two
+//!   nodes around a compaction resolve as last-writer-wins.
+//! * Edits inside a map that lost a merge conflict produce no patch in Automerge and are
+//!   not replayed. Such maps only exist briefly between concurrent creations.
+//!
+//! The protocol's own metadata (`_continuum/compactions`, `_continuum/epochs`) is never
+//! replayed: the snapshot owns it.
+
+use automerge::{AutoCommit, ChangeHash, PatchAction, Prop};
+use tracing::warn;
+
+use super::snapshot::{COMPACTIONS, EPOCHS, META_ROOT};
+use crate::store::{self, StoreError};
+
+#[derive(Debug, Default, Clone, Copy)]
+pub(crate) struct ReplayStats {
+    pub(crate) replayed: usize,
+    pub(crate) failed: usize,
+}
+
+/// Replays into `dst` everything that changed in `src` between `before` and `after`,
+/// as one commit. Both head sets must be in `src`.
+pub(crate) fn replay(
+    src: &mut AutoCommit,
+    before: &[ChangeHash],
+    after: &[ChangeHash],
+    dst: &mut AutoCommit,
+) -> ReplayStats {
+    let mut stats = ReplayStats::default();
+    if before == after {
+        return stats;
+    }
+
+    for path in changed_paths(src, before, after) {
+        replay_path(src, dst, &path, &mut stats);
+    }
+
+    let _ = dst.commit();
+    stats
+}
+
+fn replay_path(src: &AutoCommit, dst: &mut AutoCommit, path: &[String], stats: &mut ReplayStats) {
+    if is_protocol_path(path) {
+        return;
+    }
+
+    // Replacing an ancestor of the protocol metadata as a whole would clobber it; replay
+    // its children one by one instead.
+    if contains_protocol_path(path) {
+        let mut keys = store::keys_at(src, path).unwrap_or_default();
+        keys.extend(store::keys_at(dst, path).unwrap_or_default());
+        keys.sort();
+        keys.dedup();
+        for key in keys {
+            let mut child = path.to_vec();
+            child.push(key);
+            replay_path(src, dst, &child, stats);
+        }
+        return;
+    }
+
+    match store::copy_path(src, dst, path) {
+        Ok(()) => stats.replayed += 1,
+        Err(error) => {
+            log_failure(path, &error);
+            stats.failed += 1;
+        }
+    }
+}
+
+fn log_failure(path: &[String], error: &StoreError) {
+    warn!(path = ?path, error = %error, "Could not replay change into the new epoch");
+}
+
+fn is_protocol_path(path: &[String]) -> bool {
+    matches!(path, [root, part, ..] if root == META_ROOT && (part == COMPACTIONS || part == EPOCHS))
+}
+
+/// True for the root and for `_continuum` itself.
+fn contains_protocol_path(path: &[String]) -> bool {
+    matches!(path, [] | [_]) && path.first().is_none_or(|root| root == META_ROOT)
+}
+
+/// The minimal set of store paths covering every change between `before` and `after`:
+/// sorted, and no path is a prefix of another.
+pub(crate) fn changed_paths(
+    doc: &mut AutoCommit,
+    before: &[ChangeHash],
+    after: &[ChangeHash],
+) -> Vec<Vec<String>> {
+    let mut paths: Vec<Vec<String>> = doc.diff(before, after).iter().map(patch_path).collect();
+    paths.sort();
+    paths.dedup();
+
+    // Sorted order puts a prefix right before its extensions, so comparing with the
+    // last kept path is enough.
+    let mut minimal: Vec<Vec<String>> = Vec::new();
+    for path in paths {
+        if minimal.last().is_some_and(|kept| path.starts_with(kept)) {
+            continue;
+        }
+        minimal.push(path);
+    }
+    minimal
+}
+
+fn patch_path(patch: &automerge::Patch) -> Vec<String> {
+    let mut path = Vec::new();
+    for (_, prop) in &patch.path {
+        match prop {
+            Prop::Map(key) => path.push(key.clone()),
+            // Inside a list: the list is replayed as a whole.
+            Prop::Seq(_) => return path,
+        }
+    }
+
+    match &patch.action {
+        PatchAction::PutMap { key, .. } | PatchAction::DeleteMap { key, .. } => {
+            path.push(key.clone())
+        }
+        PatchAction::Increment {
+            prop: Prop::Map(key),
+            ..
+        } => path.push(key.clone()),
+        // List and text edits: the containing object.
+        _ => {}
+    }
+    path
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, ROOT};
+    use serde_json::json;
+
+    fn view(doc: &AutoCommit) -> serde_json::Value {
+        store::read_path(doc, &[]).unwrap().unwrap()
+    }
+
+    #[test]
+    fn minimal_paths() {
+        let mut doc = AutoCommit::new();
+        doc.put(ROOT, "untouched", 1_i64).unwrap();
+        let before = doc.get_heads();
+
+        let m = doc.put_object(ROOT, "m", ObjType::Map).unwrap();
+        doc.put(&m, "a", 1_i64).unwrap();
+        doc.put(ROOT, "x", 1_i64).unwrap();
+        doc.put(ROOT, "c", ScalarValue::counter(0)).unwrap();
+        doc.commit();
+        let middle = doc.get_heads();
+        doc.increment(ROOT, "c", 2).unwrap();
+        let list = doc.put_object(ROOT, "l", ObjType::List).unwrap();
+        doc.insert(&list, 0, "e").unwrap();
+        doc.commit();
+        let after = doc.get_heads();
+
+        let all = changed_paths(&mut doc, &before, &after);
+        assert_eq!(
+            all,
+            vec![
+                vec!["c".to_owned()],
+                vec!["l".to_owned()],
+                vec!["m".to_owned()],
+                vec!["x".to_owned()],
+            ]
+        );
+
+        let later = changed_paths(&mut doc, &middle, &after);
+        assert_eq!(later, vec![vec!["c".to_owned()], vec!["l".to_owned()]]);
+    }
+
+    #[test]
+    fn replays_state_per_key() {
+        // Base shared by the old document and the snapshot.
+        let mut old = AutoCommit::new();
+        let m = old.put_object(ROOT, "m", ObjType::Map).unwrap();
+        old.put(&m, "keep", "base").unwrap();
+        old.put(&m, "change", "base").unwrap();
+        old.put(ROOT, "gone", true).unwrap();
+        old.commit();
+        let base = old.get_heads();
+
+        let mut new = AutoCommit::new();
+        store::materialize_into(&old, &mut new, &|_| false).unwrap();
+        // Something the new epoch already holds that the delta must not clobber.
+        let Some((_, m_new)) = new.get(ROOT, "m").unwrap() else {
+            panic!("m missing");
+        };
+        new.put(&m_new, "keep", "new epoch").unwrap();
+        new.commit();
+
+        // The delta: one key changed, one key deleted.
+        old.put(&m, "change", "delta").unwrap();
+        old.delete(ROOT, "gone").unwrap();
+        old.commit();
+        let after = old.get_heads();
+
+        let stats = replay(&mut old, &base, &after, &mut new);
+        assert_eq!(stats.failed, 0);
+        assert_eq!(
+            view(&new),
+            json!({ "m": { "keep": "new epoch", "change": "delta" } })
+        );
+    }
+
+    #[test]
+    fn protocol_metadata_is_never_replayed() {
+        let mut old = AutoCommit::new();
+        old.put(ROOT, "x", 1_i64).unwrap();
+        old.commit();
+        let base = old.get_heads();
+
+        let mut new = AutoCommit::new();
+        store::write_path(
+            &mut new,
+            &[META_ROOT.into(), EPOCHS.into(), "1".into()],
+            &json!({ "generation": 1 }),
+        )
+        .unwrap();
+        new.commit();
+
+        // The old epoch replaces `_continuum` wholesale and writes a compaction record.
+        store::write_path(
+            &mut old,
+            &[META_ROOT.into()],
+            &json!({ "forgotten": { "pk": 1 } }),
+        )
+        .unwrap();
+        store::write_path(
+            &mut old,
+            &[META_ROOT.into(), COMPACTIONS.into(), "c".into()],
+            &json!({ "x": 1 }),
+        )
+        .unwrap();
+        old.commit();
+        let after = old.get_heads();
+
+        replay(&mut old, &base, &after, &mut new);
+        let meta = &view(&new)[META_ROOT];
+        assert_eq!(meta[EPOCHS]["1"]["generation"], json!(1));
+        assert_eq!(meta["forgotten"]["pk"], json!(1));
+        assert!(meta.get(COMPACTIONS).is_none());
+    }
+}