path: root/src/remake/delta.rs

//! 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.
//!
//! 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.
//! * Lists and text are replayed whole: concurrent edits of the same list or text on two
//!   nodes around a remake 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/remakes`, `_continuum/epochs`) is never
//! replayed: the snapshot owns it.
//!
//! Last writer wins, by the writers' clocks ([`replay_newest`]): a replayed value is a
//! fresh operation, which by Automerge's order beats everything the document it goes
//! into has, newer writes included. So a path is replayed only if no change in that
//! document touched it (or a path above or below it) later than the write being
//! replayed, as the changes' timestamps say; and the replayed changes carry the times
//! of the writes they replay, not of the replay, so later replays judge by those.

use std::collections::{BTreeMap, HashMap, HashSet};

use crate::document::{ActorId, ChangeHash, Document, ReadDoc, Transactable};
use serde::{Deserialize, Serialize};
use tracing::{info, warn};

use super::kind::{Translation, Translator};
use crate::document::meta::{self, EPOCHS, REMAKES, VOTES};
use crate::document::store::{self, InLists, StoreError};

#[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 not replayed because a newer write touched them ([`replay_newest`]), and
    /// left out (`Superseded::Drop`; with `Superseded::Quarantine`, they are among
    /// `quarantined`).
    pub(crate) superseded: 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.superseded += other.superseded;
        self.quarantined.extend(other.quarantined);
    }
}

/// What a replay does with a write that a newer one superseded (`--superseded-writes`):
/// leave it out, logged (last writer wins), or put it in the quarantine, where an
/// operator can still replay it.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) enum Superseded {
    #[default]
    Drop,
    Quarantine,
}

/// Replays into `dst` what changed in `src` between `before` and `after` (both in
/// `src`), last writer wins: a path whose newest write there (by its change's
/// timestamp) is older than a change in `dst` touching it, or a path above or below it,
/// stays as `dst` has it (`superseded` says what becomes of the write). Ties go to the
/// replay. The replayed paths are committed grouped by the time of the writes they
/// replay, each change stamped with it.
pub(crate) fn replay_newest(
    src: &Document,
    before: &[ChangeHash],
    after: &[ChangeHash],
    dst: &mut Document,
    translator: &Translator,
    superseded: Superseded,
) -> ReplayStats {
    let mut stats = ReplayStats::default();
    if before == after {
        return stats;
    }
    let paths = changed_paths(src, before, after);
    let written = touched(src, &between(src, before, after));
    let ours: Vec<i64> = paths
        .iter()
        .map(|path| newest(&written, path).unwrap_or(0))
        .collect();
    // Only changes newer than the oldest write replayed can supersede one; the
    // snapshot (time 0) never does.
    let oldest = ours.iter().copied().min().unwrap_or(0);
    let newer: Vec<ChangeHash> = dst
        .get_changes(&[])
        .iter()
        .filter(|change| change.timestamp() > oldest)
        .map(|change| change.hash())
        .collect();
    let theirs = touched(dst, &newer);

    let mut by_time: BTreeMap<i64, Vec<Vec<String>>> = BTreeMap::new();
    for (path, at) in paths.into_iter().zip(ours) {
        match newest(&theirs, &path) {
            Some(newer) if newer > at => {
                stats.superseded += 1;
                if superseded == Superseded::Quarantine {
                    stats.quarantined.push((
                        path,
                        format!("a newer write (at {newer}, this one at {at}) superseded it"),
                    ));
                }
            }
            _ => by_time.entry(at).or_default().push(path),
        }
    }
    for (at, paths) in by_time {
        let mut transaction = dst.transaction();
        stats.absorb(replay_paths(src, &mut transaction, &paths, translator));
        transaction.commit_with(automerge::transaction::CommitOptions::default().with_time(at));
    }
    if stats.superseded > 0 {
        info!(
            superseded = stats.superseded,
            quarantined = superseded == Superseded::Quarantine,
            "Writes not replayed: newer ones touched their paths"
        );
    }
    if superseded == Superseded::Quarantine {
        // Counted as quarantined, not as left out.
        stats.superseded = 0;
    }
    stats
}

/// The changes of `doc` after `before` up to `after`: those `after` builds on that
/// `before` doesn't.
fn between(doc: &Document, before: &[ChangeHash], after: &[ChangeHash]) -> Vec<ChangeHash> {
    let deps: HashMap<ChangeHash, Vec<ChangeHash>> = doc
        .get_changes(before)
        .iter()
        .map(|change| (change.hash(), change.deps().to_vec()))
        .collect();
    let mut found = HashSet::new();
    let mut todo: Vec<ChangeHash> = after.to_vec();
    while let Some(hash) = todo.pop() {
        if let Some(parents) = deps.get(&hash)
            && found.insert(hash)
        {
            todo.extend(parents.iter().copied());
        }
    }
    found.into_iter().collect()
}

/// For each of `changes` (in `doc`), its time and the store paths it touched, from its
/// operations (`store::changed_by`: a diff per change would cost what the maps it
/// touched hold, `remake::bench`).
fn touched(doc: &Document, changes: &[ChangeHash]) -> Vec<(i64, Vec<Vec<String>>)> {
    changes
        .iter()
        .filter_map(|hash| {
            let change = doc.get_change_by_hash(hash)?;
            let paths = store::changed_by(doc, hash, InLists::Whole)?;
            Some((change.timestamp(), paths))
        })
        .collect()
}

/// The time of the newest of `changes` that touched `path`, a path above it or one
/// below it.
fn newest(changes: &[(i64, Vec<Vec<String>>)], path: &[String]) -> Option<i64> {
    changes
        .iter()
        .filter(|(_, paths)| {
            paths
                .iter()
                .any(|other| other.starts_with(path) || path.starts_with(other))
        })
        .map(|(at, _)| *at)
        .max()
}

/// Replays into `dst` everything that changed in `src` between `before` and `after`,
/// as one commit, whatever `dst` has (no last writer wins): what a replay copies, for
/// tests. Both head sets must be in `src`.
#[cfg(test)]
pub(crate) fn replay(
    src: &Document,
    before: &[ChangeHash],
    after: &[ChangeHash],
    dst: &mut Document,
    translator: &Translator,
) -> ReplayStats {
    if before == after {
        return ReplayStats::default();
    }

    let paths = changed_paths(src, before, after);
    let mut transaction = dst.transaction();
    let stats = replay_paths(src, &mut transaction, &paths, translator);
    transaction.commit();
    stats
}

/// Replays the state of `paths` (from [`changed_paths`]) from `src` into `dst` (the
/// caller commits). Lets callers find the paths outside a lock and copy them inside it.
pub(crate) fn replay_paths(
    src: &Document,
    dst: &mut impl Transactable,
    paths: &[Vec<String>],
    translator: &Translator,
) -> ReplayStats {
    let mut stats = ReplayStats::default();
    for path in paths {
        replay_path(src, dst, path, translator, &mut stats);
    }
    stats
}

fn replay_path(
    src: &Document,
    dst: &mut impl Transactable,
    path: &[String],
    translator: &Translator,
    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, translator, stats);
        }
        return;
    }

    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);
            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 {
    let [root, part, ..] = path else {
        return false;
    };
    root == meta::ROOT && [REMAKES, VOTES, EPOCHS].contains(&part.as_str())
}

/// 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)
}

/// What a late node has that a snapshot lacks.
#[derive(Debug)]
pub(crate) struct Missing {
    /// Heads of everything except `own` changes the snapshot lacks (and changes built on
    /// them). Replaying from these heads to the document's heads replays exactly those.
    pub(crate) before: Vec<ChangeHash>,
    /// Changes by `own` that the snapshot lacks.
    pub(crate) own: usize,
}

/// Splits `doc`'s history against a snapshot `clock`: a change (actor, seq) is in the
/// snapshot iff seq <= clock[actor].
///
/// Only changes this node wrote itself (any of the actors in `own`: one per run and
/// epoch) are worth replaying when it rejoins: other nodes' changes after the base were
/// replayed by the nodes that switched, and replaying them again would overwrite newer
/// writes in the new epoch with stale ones.
pub(crate) fn missing_from_snapshot(
    doc: &Document,
    clock: &BTreeMap<String, u64>,
    own: &[ActorId],
) -> Missing {
    let own_hex: HashSet<String> = own.iter().map(ActorId::to_string).collect();
    let changes: Vec<(ChangeHash, Vec<ChangeHash>, bool)> = doc
        .get_changes(&[])
        .iter()
        .map(|change| {
            let actor = change.actor_id().to_string();
            let in_snapshot = clock.get(&actor).is_some_and(|&seq| change.seq() <= seq);
            let own_missing = own_hex.contains(&actor) && !in_snapshot;
            (change.hash(), change.deps().to_vec(), own_missing)
        })
        .collect();

    // Excluded: own missing changes and everything that (transitively) depends on them.
    // A fixpoint, so it doesn't rely on the order `get_changes` returns.
    let mut excluded: HashSet<ChangeHash> = changes
        .iter()
        .filter(|(_, _, own_missing)| *own_missing)
        .map(|(hash, _, _)| *hash)
        .collect();
    let own_count = excluded.len();
    loop {
        let before = excluded.len();
        for (hash, deps, _) in &changes {
            if !excluded.contains(hash) && deps.iter().any(|dep| excluded.contains(dep)) {
                excluded.insert(*hash);
            }
        }
        if excluded.len() == before {
            break;
        }
    }

    // Heads of the kept part: kept changes no kept change depends on.
    let kept: Vec<&(ChangeHash, Vec<ChangeHash>, bool)> = changes
        .iter()
        .filter(|(hash, _, _)| !excluded.contains(hash))
        .collect();
    let depended_on: HashSet<ChangeHash> = kept
        .iter()
        .flat_map(|(_, deps, _)| deps.iter().copied())
        .collect();
    let mut before: Vec<ChangeHash> = kept
        .iter()
        .map(|(hash, _, _)| *hash)
        .filter(|hash| !depended_on.contains(hash))
        .collect();
    before.sort();

    Missing {
        before,
        own: own_count,
    }
}

/// 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: &Document,
    before: &[ChangeHash],
    after: &[ChangeHash],
) -> Vec<Vec<String>> {
    let mut paths: Vec<Vec<String>> = store::changed(doc, before, after, InLists::Whole)
        .into_iter()
        .map(|changed| changed.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
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::document::testing::{change, try_change};
    use automerge::{ObjType, ROOT, ReadDoc, ScalarValue, transaction::Transactable};
    use serde_json::json;

    fn view(doc: &Document) -> serde_json::Value {
        store::read_path(doc, &[]).unwrap().unwrap()
    }

    /// A random store write as the API makes it (one change): a scalar or a small map
    /// at a path up to three keys deep, or a delete. Writes the store refuses (e.g. into
    /// a scalar) are rolled back.
    fn random_write(rng: &mut crate::testrand::Rng, doc: &mut Document) {
        const KEYS: [&str; 3] = ["a", "b", "c"];
        let depth = rng.between(1, 3);
        let path: Vec<String> = (0..depth).map(|_| rng.pick(&KEYS).to_string()).collect();
        let _ = if rng.chance(0.2) {
            try_change(doc, |doc| store::delete_path(doc, &path, false))
        } else {
            let value = match rng.below(4) {
                0 => json!(rng.below(100)),
                1 => json!(format!("s{}", rng.below(100))),
                2 => json!(rng.chance(0.5)),
                _ => {
                    let mut map = serde_json::Map::new();
                    map.insert(rng.pick(&KEYS).to_string(), json!(rng.below(100)));
                    serde_json::Value::Object(map)
                }
            };
            try_change(doc, |doc| store::write_path(doc, &path, &value))
        };
    }

    /// Compaction's promise: a snapshot of the base plus everything replayed since gives
    /// the old document's content, whatever several nodes wrote concurrently.
    #[test]
    fn replay_reproduces_the_document() {
        crate::testrand::for_seeds(200, |rng| {
            let actor = |i: u8| ActorId::from(&[i; 16][..]);
            let mut old = Document::new().with_actor(actor(0));
            for _ in 0..rng.between(0, 8) {
                random_write(rng, &mut old);
            }
            let base = old.get_heads();
            let mut new = Document::new().with_actor(actor(100));
            change(&mut new, |new| {
                store::materialize_into(&old, new, &|_| false).unwrap()
            });

            let mut nodes: Vec<Document> = (1..=rng.between(1, 3))
                .map(|i| old.fork().with_actor(actor(i as u8)))
                .collect();
            for _ in 0..rng.between(1, 20) {
                let node = rng.below(nodes.len());
                random_write(rng, &mut nodes[node]);
            }
            for node in &mut nodes {
                old.merge(node).unwrap();
            }
            let after = old.get_heads();

            let stats = replay(&old, &base, &after, &mut new, &Translator::default());
            assert_eq!(stats.failed, 0);
            assert_eq!(view(&new), view(&old));
        });
    }

    /// Every non-empty-object value of `json` with its path.
    fn leaves(
        json: &serde_json::Value,
        path: &mut Vec<String>,
        out: &mut Vec<(Vec<String>, String)>,
    ) {
        match json {
            serde_json::Value::Object(map) => {
                for (key, value) in map {
                    path.push(key.clone());
                    leaves(value, path, out);
                    path.pop();
                }
            }
            other => out.push((path.clone(), other.to_string())),
        }
    }

    /// Over random histories (writes and deletes, some on concurrent forks): the paths
    /// are sorted, none is a prefix of another, and together they cover every value that
    /// differs between before and after.
    #[test]
    fn changed_paths_cover_every_change_minimally() {
        crate::testrand::for_seeds(300, |rng| {
            let actor = |i: u8| ActorId::from(&[i; 16][..]);
            let mut doc = Document::new().with_actor(actor(0));
            for _ in 0..rng.between(0, 8) {
                random_write(rng, &mut doc);
            }
            let before = doc.get_heads();
            let mut forks: Vec<Document> = (1..=rng.between(1, 3))
                .map(|i| doc.fork().with_actor(actor(i as u8)))
                .collect();
            for _ in 0..rng.between(0, 12) {
                let fork = rng.below(forks.len());
                random_write(rng, &mut forks[fork]);
            }
            for fork in &mut forks {
                doc.merge(fork).unwrap();
            }
            let after = doc.get_heads();

            let paths = changed_paths(&doc, &before, &after);
            for pair in paths.windows(2) {
                assert!(pair[0] < pair[1], "not sorted: {paths:?}");
                assert!(!pair[1].starts_with(&pair[0]), "not minimal: {paths:?}");
            }

            let mut old = Vec::new();
            leaves(
                &view(&doc.fork_at(&before).unwrap()),
                &mut Vec::new(),
                &mut old,
            );
            let mut new = Vec::new();
            leaves(&view(&doc), &mut Vec::new(), &mut new);
            let differing = old
                .iter()
                .filter(|leaf| !new.contains(leaf))
                .chain(new.iter().filter(|leaf| !old.contains(leaf)));
            for (path, _) in differing {
                assert!(
                    paths.iter().any(|changed| path.starts_with(changed)),
                    "{path:?} changed, but {paths:?} don't cover it"
                );
            }
        });
    }

    #[test]
    fn minimal_paths() {
        let mut doc = Document::new();
        change(&mut doc, |doc| doc.put(ROOT, "untouched", 1_i64).unwrap());
        let before = doc.get_heads();

        change(&mut doc, |doc| {
            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();
        });
        let middle = doc.get_heads();
        change(&mut doc, |doc| {
            doc.increment(ROOT, "c", 2).unwrap();
            let list = doc.put_object(ROOT, "l", ObjType::List).unwrap();
            doc.insert(&list, 0, "e").unwrap();
        });
        let after = doc.get_heads();

        let all = changed_paths(&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(&doc, &middle, &after);
        assert_eq!(later, vec![vec!["c".to_owned()], vec!["l".to_owned()]]);
    }

    /// Writes to `doc` as one change made at `at` (Unix ms).
    fn write_at(
        doc: &mut Document,
        at: i64,
        f: impl FnOnce(&mut automerge::transaction::Transaction<'_>),
    ) {
        let mut transaction = doc.transaction();
        f(&mut transaction);
        transaction.commit_with(automerge::transaction::CommitOptions::default().with_time(at));
    }

    /// An old document with `x`, `m/a` and `m/b` written at 10 after the base, and a new
    /// one holding the base; `newer` writes to the new one first.
    fn superseding(
        newer: impl FnOnce(&mut Document),
        superseded: Superseded,
    ) -> (Document, ReplayStats) {
        let mut old = Document::new();
        write_at(&mut old, 1, |old| {
            let m = old.put_object(ROOT, "m", ObjType::Map).unwrap();
            old.put(&m, "a", "base").unwrap();
        });
        let base = old.get_heads();
        let mut new = Document::new();
        write_at(&mut new, 0, |new| {
            store::materialize_into(&old, new, &|_| false).unwrap();
        });
        newer(&mut new);
        write_at(&mut old, 10, |old| {
            old.put(ROOT, "x", "old").unwrap();
            let (_, m) = old.get(ROOT, "m").unwrap().unwrap();
            old.put(&m, "a", "old").unwrap();
            old.put(&m, "b", "old").unwrap();
        });
        let after = old.get_heads();
        let stats = replay_newest(
            &old,
            &base,
            &after,
            &mut new,
            &Translator::default(),
            superseded,
        );
        (new, stats)
    }

    #[test]
    fn replay_newest_keeps_newer_writes() {
        // Nothing newer: everything replayed, at the time it was written.
        let (new, stats) = superseding(|_| {}, Superseded::Drop);
        assert_eq!((stats.replayed, stats.superseded), (3, 0));
        assert_eq!(
            view(&new),
            json!({ "x": "old", "m": { "a": "old", "b": "old" } })
        );
        let times: Vec<i64> = new.get_changes(&[]).iter().map(|c| c.timestamp()).collect();
        assert_eq!(times, [0, 10]);

        // A newer write at the path wins; an older one doesn't.
        let (new, stats) = superseding(
            |new| write_at(new, 20, |new| new.put(ROOT, "x", "newer").unwrap()),
            Superseded::Drop,
        );
        assert_eq!((stats.replayed, stats.superseded), (2, 1));
        assert_eq!(view(&new)["x"], json!("newer"));
        assert_eq!(view(&new)["m"], json!({ "a": "old", "b": "old" }));
        let (new, _) = superseding(
            |new| write_at(new, 5, |new| new.put(ROOT, "x", "older").unwrap()),
            Superseded::Drop,
        );
        assert_eq!(view(&new)["x"], json!("old"));

        // Per key: a newer `m/a` wins over the replayed `m/a`, not over `m/b`.
        let (new, stats) = superseding(
            |new| {
                write_at(new, 20, |new| {
                    let (_, m) = new.get(ROOT, "m").unwrap().unwrap();
                    new.put(&m, "a", "newer").unwrap();
                })
            },
            Superseded::Drop,
        );
        assert_eq!(stats.superseded, 1);
        assert_eq!(view(&new)["m"], json!({ "a": "newer", "b": "old" }));
        // A newer write above the paths (`m` replaced) wins over both.
        let (new, stats) = superseding(
            |new| write_at(new, 20, |new| new.put(ROOT, "m", "a scalar now").unwrap()),
            Superseded::Drop,
        );
        assert_eq!(stats.superseded, 2);
        assert_eq!(view(&new)["m"], json!("a scalar now"));

        // Quarantined instead, with the time of each.
        let (_, stats) = superseding(
            |new| write_at(new, 20, |new| new.put(ROOT, "x", "newer").unwrap()),
            Superseded::Quarantine,
        );
        assert_eq!(stats.superseded, 0);
        assert_eq!(stats.quarantined.len(), 1);
        assert_eq!(stats.quarantined[0].0, ["x"]);
    }

    /// A newer write below a replayed path wins too: the replay would replace it.
    #[test]
    fn replay_newest_keeps_newer_writes_below() {
        let mut old = Document::new();
        write_at(&mut old, 1, |old| old.put(ROOT, "base", true).unwrap());
        let base = old.get_heads();
        let mut new = Document::new();
        write_at(&mut new, 0, |new| {
            store::materialize_into(&old, new, &|_| false).unwrap();
            new.put_object(ROOT, "m", ObjType::Map).unwrap();
        });
        write_at(&mut new, 20, |new| {
            let (_, m) = new.get(ROOT, "m").unwrap().unwrap();
            new.put(&m, "a", "newer").unwrap();
        });
        // The whole map, older.
        write_at(&mut old, 10, |old| {
            let m = old.put_object(ROOT, "m", ObjType::Map).unwrap();
            old.put(&m, "a", "old").unwrap();
        });
        let after = old.get_heads();
        let stats = replay_newest(
            &old,
            &base,
            &after,
            &mut new,
            &Translator::default(),
            Superseded::Drop,
        );
        assert_eq!(stats.superseded, 1);
        assert_eq!(view(&new)["m"], json!({ "a": "newer" }));
    }

    #[test]
    fn replays_state_per_key() {
        // Base shared by the old document and the snapshot.
        let mut old = Document::new();
        let m = change(&mut old, |old| {
            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();
            m
        });
        let base = old.get_heads();

        let mut new = Document::new();
        change(&mut new, |new| {
            store::materialize_into(&old, 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();
        });

        // The delta: one key changed, one key deleted.
        change(&mut old, |old| {
            old.put(&m, "change", "delta").unwrap();
            old.delete(ROOT, "gone").unwrap();
        });
        let after = old.get_heads();

        let stats = replay(&old, &base, &after, &mut new, &Translator::default());
        assert_eq!(stats.failed, 0);
        assert_eq!(
            view(&new),
            json!({ "m": { "keep": "new epoch", "change": "delta" } })
        );
    }

    #[test]
    fn missing_counts_only_own_changes() {
        let own = ActorId::from(&b"own-actor"[..]);
        let other = ActorId::from(&b"other-actor"[..]);

        let mut doc = Document::new().with_actor(own.clone());
        change(&mut doc, |doc| doc.put(ROOT, "base", 1_i64).unwrap());
        let base = doc.get_heads();

        // The snapshot contains the base only.
        let clock: BTreeMap<String, u64> = [(own.to_string(), 1)].into_iter().collect();

        // A foreign change after the base (replayed by its own node) ...
        let mut foreign = doc.fork().with_actor(other);
        change(&mut foreign, |foreign| {
            foreign.put(ROOT, "theirs", 1_i64).unwrap()
        });
        doc.merge(&mut foreign).unwrap();
        let with_foreign = doc.get_heads();

        // ... and one of ours.
        change(&mut doc, |doc| doc.put(ROOT, "mine", 1_i64).unwrap());
        let after = doc.get_heads();

        let missing = missing_from_snapshot(&doc, &clock, std::slice::from_ref(&own));
        assert_eq!(missing.own, 1);
        assert_eq!(missing.before, with_foreign);
        assert_ne!(missing.before, base);
        assert_eq!(
            changed_paths(&doc, &missing.before, &after),
            vec![vec!["mine".to_owned()]]
        );

        // Nothing of ours missing: nothing to replay.
        let full: BTreeMap<String, u64> = [(own.to_string(), 2)].into_iter().collect();
        let none = missing_from_snapshot(&doc, &full, std::slice::from_ref(&own));
        assert_eq!(none.own, 0);
        let mut heads = after.clone();
        heads.sort();
        assert_eq!(none.before, heads);
    }

    #[test]
    fn protocol_metadata_is_never_replayed() {
        let mut old = Document::new();
        change(&mut old, |old| old.put(ROOT, "x", 1_i64).unwrap());
        let base = old.get_heads();

        let mut new = Document::new();
        change(&mut new, |new| {
            store::write_path(
                new,
                &meta::path(&[EPOCHS, "1"]),
                &json!({ "generation": 1 }),
            )
            .unwrap()
        });

        // The old epoch replaces `_continuum` wholesale and writes a remake record.
        change(&mut old, |old| {
            store::write_path(old, &meta::path(&[]), &json!({ "forgotten": { "pk": 1 } })).unwrap();
            store::write_path(old, &meta::path(&[REMAKES, "c"]), &json!({ "x": 1 })).unwrap();
        });
        let after = old.get_heads();

        replay(&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));
        assert!(meta.get(REMAKES).is_none());
    }
}