add compaction

4 files changed, 595 insertions, 0 deletions

diff --git a/src/compaction/mod.rs b/src/compaction/mod.rs
new file mode 100644
index 0000000..703764b
--- /dev/null
+++ b/src/compaction/mod.rs
@@ -0,0 +1,6 @@
+//! Compaction: replacing the document with a snapshot in a new epoch.
+//!
+//! Step 3 provides the deterministic snapshot build. The protocol (proposal, settle,
+//! commit, switch) and the admin API follow in step 4.
+
+pub(crate) mod snapshot;
diff --git a/src/compaction/snapshot.rs b/src/compaction/snapshot.rs
new file mode 100644
index 0000000..9dd54c0
--- /dev/null
+++ b/src/compaction/snapshot.rs
@@ -0,0 +1,459 @@
+//! Deterministic snapshot of the document at agreed base heads.
+//!
+//! Every node builds the snapshot of a committed compaction on its own. For that to
+//! work, the snapshot must be byte-identical everywhere: same operations, same order,
+//! same actor, same timestamp. Its change hash then identifies the new epoch, and a
+//! node whose build produces a different hash knows it must fetch the snapshot instead.
+//!
+//! Determinism rests on:
+//! * the content: the merged view at the base heads (`store::materialize_into`), which
+//!   depends only on the base, not on how a node's history got there;
+//! * a fixed actor derived from the compaction ID and generation;
+//! * commit time 0 and no message;
+//! * canonical (sorted) JSON for the metadata;
+//! * the same Automerge version on every node (encodings may differ across versions).
+//!
+//! The snapshot carries its own metadata under `_continuum/epochs/<generation>`:
+//! previous epoch, base heads, per-actor clock and retention policy. Later steps read it
+//! at the snapshot change itself, so edits made afterwards through the store API can't
+//! change what the protocol sees.
+
+// Wired into the compaction protocol in step 4.
+#![cfg_attr(not(test), allow(dead_code))]
+
+use std::collections::BTreeMap;
+
+use automerge::{transaction::CommitOptions, ActorId, AutoCommit, ChangeHash};
+use serde_json::{json, Map as JsonMap, Value as JsonValue};
+use sha2::{Digest, Sha256};
+
+use crate::replica::Epoch;
+use crate::store::{self, StoreError};
+
+pub(crate) const META_ROOT: &str = "_continuum";
+pub(crate) const COMPACTIONS: &str = "compactions";
+pub(crate) const EPOCHS: &str = "epochs";
+
+/// Everything that determines a snapshot besides the base content.
+pub(crate) struct SnapshotParams<'a> {
+    /// ID of the committed compaction; seeds the snapshot actor.
+    pub(crate) compaction_id: &'a str,
+    /// The epoch being compacted. The snapshot starts generation `previous + 1`.
+    pub(crate) previous: Epoch,
+    /// Heads of the base. Every node must hold all of these changes.
+    pub(crate) base_heads: &'a [ChangeHash],
+    /// Retention policy from the compaction request; stored verbatim (canonicalized).
+    pub(crate) policy: &'a JsonValue,
+}
+
+pub(crate) struct Snapshot {
+    /// The new epoch: next generation, ID = hash of the snapshot change.
+    pub(crate) epoch: Epoch,
+    /// A document consisting of exactly that one change.
+    pub(crate) doc: AutoCommit,
+    /// Highest sequence number per actor (hex) contained in the base. A change
+    /// (actor, seq) is part of the snapshot iff seq <= clock[actor].
+    pub(crate) clock: BTreeMap<String, u64>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum SnapshotError {
+    #[error("base head {0} is not in the local document")]
+    MissingBase(ChangeHash),
+    #[error("snapshot produced {0} heads instead of exactly one")]
+    UnexpectedHeads(usize),
+    #[error(transparent)]
+    Doc(#[from] automerge::AutomergeError),
+    #[error(transparent)]
+    Store(#[from] StoreError),
+}
+
+/// Path of the metadata record for an epoch generation.
+pub(crate) fn epoch_metadata_path(generation: u64) -> Vec<String> {
+    vec![
+        META_ROOT.to_owned(),
+        EPOCHS.to_owned(),
+        generation.to_string(),
+    ]
+}
+
+/// Builds the snapshot of `source` at `params.base_heads`.
+pub(crate) fn build(
+    source: &mut AutoCommit,
+    params: &SnapshotParams<'_>,
+) -> Result<Snapshot, SnapshotError> {
+    // fork_at with unknown heads would quietly build something else.
+    for head in params.base_heads {
+        if source.get_change_by_hash(head).is_none() {
+            return Err(SnapshotError::MissingBase(*head));
+        }
+    }
+
+    let mut base = source.fork_at(params.base_heads)?;
+    let clock = clock_of(&mut base);
+    let generation = params.previous.generation + 1;
+
+    let mut snapshot =
+        AutoCommit::new().with_actor(snapshot_actor(params.compaction_id, generation));
+
+    // Compaction records describe the old epoch's protocol and stay behind.
+    store::materialize_into(&base, &mut snapshot, &is_compaction_record)?;
+
+    let mut base_heads: Vec<String> = params.base_heads.iter().map(|h| h.to_string()).collect();
+    base_heads.sort();
+
+    let metadata = canonical(json!({
+        "generation": generation,
+        "compactionId": params.compaction_id,
+        "previous": {
+            "generation": params.previous.generation,
+            "id": params.previous.id_hex(),
+        },
+        "baseHeads": base_heads,
+        "clock": clock,
+        "policy": params.policy,
+    }));
+    store::write_path(&mut snapshot, &epoch_metadata_path(generation), &metadata)?;
+
+    let _ = snapshot.commit_with(CommitOptions::default().with_time(0));
+
+    let heads = snapshot.get_heads();
+    let [id] = heads.as_slice() else {
+        return Err(SnapshotError::UnexpectedHeads(heads.len()));
+    };
+
+    Ok(Snapshot {
+        epoch: Epoch {
+            generation,
+            id: id.0,
+        },
+        doc: snapshot,
+        clock,
+    })
+}
+
+/// Highest sequence number per actor in `doc`'s history.
+fn clock_of(doc: &mut AutoCommit) -> BTreeMap<String, u64> {
+    let mut clock = BTreeMap::new();
+    for change in doc.get_changes(&[]) {
+        let seq = clock.entry(change.actor_id().to_string()).or_insert(0);
+        *seq = (*seq).max(change.seq());
+    }
+    clock
+}
+
+/// The same 16-byte actor on every node, unique per compaction and generation.
+fn snapshot_actor(compaction_id: &str, generation: u64) -> ActorId {
+    let digest = Sha256::new()
+        .chain_update(b"continuum.snapshot\0")
+        .chain_update(compaction_id.as_bytes())
+        .chain_update(b"\0")
+        .chain_update(generation.to_be_bytes())
+        .finalize();
+    ActorId::from(&digest[..16])
+}
+
+fn is_compaction_record(path: &[String]) -> bool {
+    matches!(path, [root, records] if root == META_ROOT && records == COMPACTIONS)
+}
+
+/// 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 {
+    match value {
+        JsonValue::Object(map) => {
+            let sorted: BTreeMap<String, JsonValue> = map
+                .into_iter()
+                .map(|(key, value)| (key, canonical(value)))
+                .collect();
+            JsonValue::Object(sorted.into_iter().collect::<JsonMap<_, _>>())
+        }
+        JsonValue::Array(items) => JsonValue::Array(items.into_iter().map(canonical).collect()),
+        other => other,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, Value, ROOT};
+
+    fn doc(actor: &[u8]) -> AutoCommit {
+        AutoCommit::new().with_actor(ActorId::from(actor))
+    }
+
+    fn params<'a>(heads: &'a [ChangeHash], policy: &'a JsonValue) -> SnapshotParams<'a> {
+        SnapshotParams {
+            compaction_id: "c-1",
+            previous: Epoch::GENESIS,
+            base_heads: heads,
+            policy,
+        }
+    }
+
+    fn snapshot_of(source: &mut AutoCommit) -> Snapshot {
+        let heads = source.get_heads();
+        let policy = json!({});
+        build(source, &params(&heads, &policy)).expect("snapshot")
+    }
+
+    /// The user-visible view, without our own metadata.
+    fn user_view(doc: &AutoCommit) -> JsonValue {
+        let mut view = store::read_path(doc, &[]).unwrap().unwrap();
+        view.as_object_mut().unwrap().remove(META_ROOT);
+        view
+    }
+
+    #[test]
+    fn same_view_different_history_same_snapshot() {
+        // One history: overwrites, increments, one-shot text.
+        let mut a = doc(b"actor-a");
+        a.put(ROOT, "x", 1_i64).unwrap();
+        a.commit();
+        a.put(ROOT, "x", 2_i64).unwrap();
+        a.put(ROOT, "c", ScalarValue::counter(0)).unwrap();
+        a.increment(ROOT, "c", 5).unwrap();
+        let text = a.put_object(ROOT, "t", ObjType::Text).unwrap();
+        a.splice_text(&text, 0, 0, "hello").unwrap();
+        let m = a.put_object(ROOT, "m", ObjType::Map).unwrap();
+        a.put(&m, "k", "v").unwrap();
+        a.commit();
+
+        // Another actor, other order, other steps, same result.
+        let mut b = doc(b"actor-b");
+        let m = b.put_object(ROOT, "m", ObjType::Map).unwrap();
+        b.put(&m, "k", "v").unwrap();
+        let text = b.put_object(ROOT, "t", ObjType::Text).unwrap();
+        b.splice_text(&text, 0, 0, "hel").unwrap();
+        b.commit();
+        b.splice_text(&text, 3, 0, "lo").unwrap();
+        b.put(ROOT, "c", ScalarValue::counter(5)).unwrap();
+        b.put(ROOT, "x", 2_i64).unwrap();
+        b.commit();
+
+        assert_eq!(user_view(&a), user_view(&b));
+
+        let snap_a = snapshot_of(&mut a);
+        let snap_b = snapshot_of(&mut b);
+
+        // Different histories mean different clocks, and the clock is part of the
+        // metadata. Equal views alone therefore give equal content but not equal IDs;
+        // equal bases do give equal IDs (see `rebuild_is_deterministic`).
+        assert_eq!(user_view(&snap_a.doc), user_view(&snap_b.doc));
+        assert_eq!(user_view(&snap_a.doc), user_view(&a));
+    }
+
+    #[test]
+    fn rebuild_is_deterministic() {
+        let mut a = doc(b"actor-a");
+        a.put(ROOT, "x", 1_i64).unwrap();
+        let list = a.put_object(ROOT, "l", ObjType::List).unwrap();
+        a.insert(&list, 0, "first").unwrap();
+        let inner = a.insert_object(&list, 1, ObjType::Map).unwrap();
+        a.put(&inner, "k", true).unwrap();
+        a.commit();
+
+        // A second node holding the same changes, received in a different way.
+        let mut b = doc(b"actor-b");
+        b.merge(&mut a.clone()).unwrap();
+
+        let first = snapshot_of(&mut a);
+        let again = snapshot_of(&mut a);
+        let elsewhere = snapshot_of(&mut b);
+
+        assert_eq!(first.epoch, again.epoch);
+        assert_eq!(first.epoch, elsewhere.epoch);
+        assert_eq!(first.epoch.generation, 1);
+    }
+
+    #[test]
+    fn conflicts_collapse_into_the_merged_view() {
+        let mut a = doc(b"actor-a");
+        let cfg = a.put_object(ROOT, "cfg", ObjType::Map).unwrap();
+        a.put(&cfg, "x", 1_i64).unwrap();
+        a.commit();
+
+        let mut b = doc(b"actor-b");
+        let cfg = b.put_object(ROOT, "cfg", ObjType::Map).unwrap();
+        b.put(&cfg, "y", 2_i64).unwrap();
+        b.commit();
+
+        a.merge(&mut b).unwrap();
+        let (_, conflicts) = store::read_path_with_conflicts(&a, &[]).unwrap().unwrap();
+        assert!(
+            !conflicts.is_empty(),
+            "test setup should produce a conflict"
+        );
+
+        let snap = snapshot_of(&mut a);
+        assert_eq!(user_view(&snap.doc), json!({ "cfg": { "x": 1, "y": 2 } }));
+
+        let (_, conflicts) = store::read_path_with_conflicts(&snap.doc, &[])
+            .unwrap()
+            .unwrap();
+        assert!(
+            conflicts.is_empty(),
+            "snapshot still has conflicts: {conflicts:?}"
+        );
+    }
+
+    #[test]
+    fn base_heads_bound_content_and_clock() {
+        let mut a = doc(b"actor-a");
+        for i in 0..3_i64 {
+            a.put(ROOT, format!("a{i}"), i).unwrap();
+            a.commit();
+        }
+
+        let mut b = doc(b"actor-b");
+        for i in 0..2_i64 {
+            b.put(ROOT, format!("b{i}"), i).unwrap();
+            b.commit();
+        }
+        a.merge(&mut b).unwrap();
+
+        let base = a.get_heads();
+
+        // After the base: must not appear in the snapshot or its clock.
+        a.put(ROOT, "late", true).unwrap();
+        a.commit();
+
+        let policy = json!({});
+        let snap = build(&mut a, &params(&base, &policy)).unwrap();
+
+        let view = user_view(&snap.doc);
+        assert!(view.get("late").is_none());
+        assert_eq!(view["a2"], json!(2));
+        assert_eq!(view["b1"], json!(1));
+
+        let actor_a = ActorId::from(b"actor-a".as_slice()).to_string();
+        let actor_b = ActorId::from(b"actor-b".as_slice()).to_string();
+        assert_eq!(snap.clock.get(&actor_a), Some(&3));
+        assert_eq!(snap.clock.get(&actor_b), Some(&2));
+    }
+
+    #[test]
+    fn metadata_written_and_compaction_records_dropped() {
+        let mut a = doc(b"actor-a");
+        store::write_path(
+            &mut a,
+            &[META_ROOT.into(), COMPACTIONS.into(), "c-1".into()],
+            &json!({ "state": "committed" }),
+        )
+        .unwrap();
+        store::write_path(
+            &mut a,
+            &[META_ROOT.into(), "forgotten".into(), "pk".into()],
+            &json!({ "at": 1 }),
+        )
+        .unwrap();
+        a.commit();
+
+        let heads = a.get_heads();
+        let policy = json!({ "replayWindow": { "epochs": 1 } });
+        let snap = build(&mut a, &params(&heads, &policy)).unwrap();
+
+        let meta = store::read_path(&snap.doc, &[META_ROOT.into()])
+            .unwrap()
+            .unwrap();
+        assert!(meta.get(COMPACTIONS).is_none());
+        assert_eq!(meta["forgotten"]["pk"]["at"], json!(1));
+
+        let epoch = &meta[EPOCHS]["1"];
+        assert_eq!(epoch["generation"], json!(1));
+        assert_eq!(epoch["compactionId"], json!("c-1"));
+        assert_eq!(epoch["previous"]["generation"], json!(0));
+        assert_eq!(epoch["policy"], policy);
+        assert_eq!(epoch["baseHeads"].as_array().unwrap().len(), heads.len());
+    }
+
+    #[test]
+    fn automerge_types_survive() {
+        let mut a = doc(b"actor-a");
+        a.put(ROOT, "c", ScalarValue::counter(3)).unwrap();
+        a.increment(ROOT, "c", 4).unwrap();
+        let text = a.put_object(ROOT, "t", ObjType::Text).unwrap();
+        a.splice_text(&text, 0, 0, "abc").unwrap();
+        a.put(ROOT, "bytes", ScalarValue::Bytes(vec![0, 1, 255]))
+            .unwrap();
+        a.put(ROOT, "ts", ScalarValue::Timestamp(1_700_000_000_000))
+            .unwrap();
+        a.put(ROOT, "neg_zero", -0.0_f64).unwrap();
+        a.commit();
+
+        let snap = snapshot_of(&mut a);
+        let d = &snap.doc;
+
+        match d.get(ROOT, "c").unwrap() {
+            Some((Value::Scalar(s), _)) => match s.as_ref() {
+                ScalarValue::Counter(c) => assert_eq!(i64::from(c.clone()), 7),
+                other => panic!("counter became {other:?}"),
+            },
+            other => panic!("unexpected {other:?}"),
+        }
+
+        let (_, text_id) = d.get(ROOT, "t").unwrap().unwrap();
+        assert_eq!(d.object_type(&text_id).unwrap(), ObjType::Text);
+        assert_eq!(d.text(&text_id).unwrap(), "abc");
+
+        match d.get(ROOT, "bytes").unwrap() {
+            Some((Value::Scalar(s), _)) => {
+                assert_eq!(s.as_ref(), &ScalarValue::Bytes(vec![0, 1, 255]))
+            }
+            other => panic!("unexpected {other:?}"),
+        }
+        match d.get(ROOT, "ts").unwrap() {
+            Some((Value::Scalar(s), _)) => {
+                assert_eq!(s.as_ref(), &ScalarValue::Timestamp(1_700_000_000_000))
+            }
+            other => panic!("unexpected {other:?}"),
+        }
+        match d.get(ROOT, "neg_zero").unwrap() {
+            Some((Value::Scalar(s), _)) => match s.as_ref() {
+                ScalarValue::F64(f) => assert_eq!(f.to_bits(), (-0.0_f64).to_bits()),
+                other => panic!("float became {other:?}"),
+            },
+            other => panic!("unexpected {other:?}"),
+        }
+    }
+
+    #[test]
+    fn snapshot_is_a_single_change_with_the_epoch_id() {
+        let mut a = doc(b"actor-a");
+        a.put(ROOT, "x", 1_i64).unwrap();
+        a.commit();
+
+        let mut snap = snapshot_of(&mut a);
+        let heads = snap.doc.get_heads();
+        assert_eq!(heads.len(), 1);
+        assert_eq!(heads[0].0, snap.epoch.id);
+        assert_eq!(snap.doc.get_changes(&[]).len(), 1);
+    }
+
+    #[test]
+    fn missing_base_is_rejected() {
+        let mut a = doc(b"actor-a");
+        a.put(ROOT, "x", 1_i64).unwrap();
+        a.commit();
+
+        let mut other = doc(b"actor-b");
+        other.put(ROOT, "y", 1_i64).unwrap();
+        other.commit();
+        let foreign = other.get_heads();
+
+        let policy = json!({});
+        assert!(matches!(
+            build(&mut a, &params(&foreign, &policy)),
+            Err(SnapshotError::MissingBase(_))
+        ));
+    }
+
+    #[test]
+    fn empty_document() {
+        let mut a = doc(b"actor-a");
+        let snap = snapshot_of(&mut a);
+        assert_eq!(snap.epoch.generation, 1);
+        assert!(snap.clock.is_empty());
+        assert_eq!(user_view(&snap.doc), json!({}));
+    }
+}
diff --git a/src/main.rs b/src/main.rs
index 850256f..8e128f0 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -21,6 +21,7 @@ use url::Url;
 mod api;
 mod auth;
 mod bootstrap;
+mod compaction;
 mod firewall;
 mod mycelium;
 mod peer;
diff --git a/src/store.rs b/src/store.rs
index cb35272..513728e 100644
--- a/src/store.rs
+++ b/src/store.rs
@@ -301,6 +301,135 @@ pub fn delete_path(doc: &mut AutoCommit, path: &[String], prune: bool) -> Result
     Ok(())
 }
 
+// --- materialization ---------------------------------------------------------
+
+/// Copies the merged view of `src` into `dst` as fresh objects.
+///
+/// * Conflicts collapse exactly as reads resolve them: concurrent maps merge, everything
+///   else keeps the newest candidate.
+/// * Automerge types survive: text stays text, counters stay counters (normalized to
+///   their current value), bytes, timestamps and floats are copied bit-exact.
+/// * Map keys are visited in sorted order and list elements in order, so two documents
+///   with the same view produce the same operations in the same order.
+///
+/// `skip` sees the path of every map entry and list element (list indices as decimal
+/// strings); returning true leaves that subtree out.
+#[cfg_attr(not(test), allow(dead_code))] // used by compaction from step 4 on
+pub(crate) fn materialize_into(
+    src: &AutoCommit,
+    dst: &mut AutoCommit,
+    skip: &dyn Fn(&[String]) -> bool,
+) -> Result<(), StoreError> {
+    let mut path = Vec::new();
+    copy_maps(
+        src,
+        &[automerge::ROOT],
+        dst,
+        &automerge::ROOT,
+        &mut path,
+        skip,
+    )
+}
+
+fn copy_maps(
+    src: &AutoCommit,
+    maps: &[ObjId],
+    dst: &mut AutoCommit,
+    target: &ObjId,
+    path: &mut Vec<String>,
+    skip: &dyn Fn(&[String]) -> bool,
+) -> Result<(), StoreError> {
+    // merged_keys is a BTreeSet: sorted, so the operation order is deterministic.
+    for key in merged_keys(src, maps) {
+        path.push(key.clone());
+        if !skip(path) {
+            if let Some(resolved) = resolve(src, maps, &key, None)? {
+                copy_resolved(src, &resolved, dst, target, Slot::Key(&key), path, skip)?;
+            }
+        }
+        path.pop();
+    }
+    Ok(())
+}
+
+fn copy_resolved(
+    src: &AutoCommit,
+    resolved: &Resolved,
+    dst: &mut AutoCommit,
+    parent: &ObjId,
+    slot: Slot,
+    path: &mut Vec<String>,
+    skip: &dyn Fn(&[String]) -> bool,
+) -> Result<(), StoreError> {
+    match resolved {
+        Resolved::Maps(maps) => {
+            let id = put_object(dst, parent, slot, ObjType::Map)?;
+            copy_maps(src, maps, dst, &id, path, skip)
+        }
+        Resolved::Other(Value::Scalar(scalar), _) => {
+            let scalar = match scalar.as_ref() {
+                // A read counter carries its increment history; copy only its value, so
+                // equal counters produce equal operations.
+                ScalarValue::Counter(counter) => ScalarValue::counter(i64::from(counter.clone())),
+                other => other.clone(),
+            };
+            put_scalar(dst, parent, slot, scalar)
+        }
+        Resolved::Other(Value::Object(obj_type), id) => {
+            copy_object(src, *obj_type, id, dst, parent, slot, path, skip)
+        }
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+fn copy_object(
+    src: &AutoCommit,
+    obj_type: ObjType,
+    id: &ObjId,
+    dst: &mut AutoCommit,
+    parent: &ObjId,
+    slot: Slot,
+    path: &mut Vec<String>,
+    skip: &dyn Fn(&[String]) -> bool,
+) -> Result<(), StoreError> {
+    match obj_type {
+        ObjType::Map => copy_resolved(
+            src,
+            &Resolved::Maps(vec![id.clone()]),
+            dst,
+            parent,
+            slot,
+            path,
+            skip,
+        ),
+        ObjType::List => {
+            let list = put_object(dst, parent, slot, ObjType::List)?;
+            // Skipped elements leave no gap in the copy.
+            let mut position = 0;
+            for (index, item) in src.list_range(id, ..).enumerate() {
+                path.push(index.to_string());
+                if !skip(path) {
+                    let child_id = item.id();
+                    let child = match Value::from(item.value).into_owned() {
+                        Value::Object(ObjType::Map) => Resolved::Maps(vec![child_id]),
+                        value => Resolved::Other(value, child_id),
+                    };
+                    copy_resolved(src, &child, dst, &list, Slot::Index(position), path, skip)?;
+                    position += 1;
+                }
+                path.pop();
+            }
+            Ok(())
+        }
+        ObjType::Text => {
+            let text = put_object(dst, parent, slot, ObjType::Text)?;
+            dst.splice_text(&text, 0, 0, &src.text(id)?)?;
+            Ok(())
+        }
+        ObjType::Table => Err(anyhow!("Automerge tables are not supported").into()),
+    }
+}
+
 /// Where a value is placed inside its parent object.
 #[derive(Clone, Copy)]
 enum Slot<'a> {