LWW everywhere, mostly

1 files changed, 283 insertions, 4 deletions

diff --git a/src/remake/delta.rs b/src/remake/delta.rs
index 06f2913..2c6c93d 100644
--- a/src/remake/delta.rs
+++ b/src/remake/delta.rs
@@ -20,11 +20,19 @@
 //!
 //! 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, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 
-use crate::document::{ActorId, ChangeHash, Document, Transactable};
-use tracing::warn;
+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};
@@ -36,6 +44,10 @@ pub(crate) struct ReplayStats {
     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)>,
 }
@@ -46,12 +58,144 @@ impl 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.
+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(doc, change.deps(), &[*hash], InLists::Whole)
+                .into_iter()
+                .map(|changed| changed.path)
+                .collect();
+            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. Both head sets must be in `src`.
+/// 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],
@@ -423,6 +567,141 @@ mod tests {
         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.