LWW everywhere, mostly

4 files changed, 340 insertions, 23 deletions

diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 2da8540..319037d 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -49,7 +49,6 @@ use super::quarantine;
 use super::snapshot::{self, EpochMetadata};
 use crate::clock::UnixMillis;
 use crate::document::replica::Epoch;
-use crate::document::store::StoreError;
 
 /// Pause before trying the same target again after a failed rejoin.
 const RETRY: Duration = Duration::from_secs(30);
@@ -308,8 +307,14 @@ async fn fetch_snapshot<H: Host>(host: H, target: Epoch) {
 /// What to do with this node's own changes the new epoch lacks.
 enum Plan {
     Nothing,
-    /// Replay these paths, carried through every remake since the node's epoch.
-    Replay(Vec<Vec<String>>, Translator),
+    /// Replay these paths (what changed from `before` to `after` in the old
+    /// document), carried through every remake since the node's epoch.
+    Replay {
+        paths: Vec<Vec<String>>,
+        before: Vec<ChangeHash>,
+        after: Vec<ChangeHash>,
+        translator: Translator,
+    },
     Quarantine {
         paths: Vec<Vec<String>>,
         reason: String,
@@ -402,13 +407,18 @@ async fn finish<H: Host>(host: &H) {
 
     match plan {
         Plan::Nothing => {}
-        Plan::Replay(paths, translator) => {
+        Plan::Replay {
+            paths,
+            before,
+            after,
+            translator,
+        } => {
             let document = Arc::new(old.take().expect("old document"));
+            let superseded = host.superseded();
             let replayed = {
                 let document = Arc::clone(&document);
-                let paths = paths.clone();
-                host.commit_doc(move |doc| {
-                    Ok::<_, StoreError>(delta::replay_paths(&document, doc, &paths, &translator))
+                host.commit_changes(move |doc| {
+                    delta::replay_newest(&document, &before, &after, doc, &translator, superseded)
                 })
                 .await
             };
@@ -478,6 +488,7 @@ async fn finish<H: Host>(host: &H) {
         own_changes = own_missing,
         replayed = stats.replayed,
         failed = stats.failed,
+        superseded = stats.superseded,
         quarantine = ?quarantine_id,
         secs = started.elapsed().as_secs(),
         "Rejoined newer epoch"
@@ -578,7 +589,12 @@ fn prepare(
         (Some(next), Ok(translator)) => {
             let away_ms = now.ms_since(next.proposed_at);
             if next.policy.replay_window.covers(gap, away_ms) {
-                Plan::Replay(paths, translator)
+                Plan::Replay {
+                    paths,
+                    before: missing.before.clone(),
+                    after: old_heads.clone(),
+                    translator,
+                }
             } else {
                 Plan::Quarantine {
                     paths,
@@ -733,7 +749,7 @@ mod tests {
             UnixMillis(0),
         ) {
             Prepared::Ready {
-                plan: Plan::Replay(paths, _),
+                plan: Plan::Replay { paths, .. },
                 own_missing,
                 ..
             } => {
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.
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index 01e1e3e..0d3d073 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -37,6 +37,7 @@ mod vote;
 
 #[cfg(test)]
 pub(crate) use api::testing;
+pub(crate) use delta::Superseded;
 pub(crate) use kind::remake_genesis;
 pub(crate) use local::RemakeState;
 pub(crate) use switch::run_worker;
@@ -193,6 +194,17 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
         T: Send + 'static,
         E: From<StoreError> + Send + 'static;
 
+    /// What a replay does with writes a newer one superseded (`--superseded-writes`).
+    fn superseded(&self) -> Superseded;
+
+    /// Changes `f` commits itself into the live document (as many as it likes, each
+    /// with its own time), under the write lock, then published and replicated.
+    /// Refused while the document isn't writable, as [`Host::commit_doc`].
+    fn commit_changes<T, F>(&self, f: F) -> impl Future<Output = Result<T, StoreError>> + Send
+    where
+        F: FnOnce(&mut Document) -> T + Send + 'static,
+        T: Send + 'static;
+
     /// Lets writes in again, after a rejoin swapped in a document that took none
     /// ([`Host::switch_epoch`] with `writable` false).
     fn accept_writes(&self) -> impl Future<Output = ()> + Send;
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index 9b03789..7d6611d 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -193,6 +193,7 @@ async fn switch<H: Host>(
         .translator()
         .ok_or_else(|| format!("unknown remake kind {:?}", request.kind))?;
     let late_translator = translator.clone();
+    let superseded = host.superseded();
     let previous = request.previous;
     let base = request.base_heads.clone();
     let retention = record.proposal.policy.archive_retention.clone();
@@ -235,20 +236,20 @@ async fn switch<H: Host>(
             // A fresh actor: this node's writes in the new epoch.
             let mut doc = snapshot.doc.with_actor(actor);
 
-            // The carried record and the replayed writes go into one change.
             let heads = copy.get_heads();
-            let paths = delta::changed_paths(&copy, &base, &heads);
-            let mut transaction = doc.transaction();
 
             // Keep this remake's record visible in the new epoch, for its history.
-            if carry_record
-                && let Err(error) = store::copy_path(&copy, &mut transaction, &record_path)
-            {
-                warn!(error = %error, "Could not carry the remake record into the new epoch");
+            if carry_record {
+                let mut transaction = doc.transaction();
+                if let Err(error) = store::copy_path(&copy, &mut transaction, &record_path) {
+                    warn!(error = %error, "Could not carry the remake record into the new epoch");
+                }
+                transaction.commit();
             }
 
-            let stats = delta::replay_paths(&copy, &mut transaction, &paths, &translator);
-            transaction.commit();
+            // Each write at the time it was made, so later replays judge by it.
+            let stats =
+                delta::replay_newest(&copy, &base, &heads, &mut doc, &translator, superseded);
             Ok((doc, heads, stats))
         },
     )
@@ -283,6 +284,7 @@ async fn switch<H: Host>(
         replayed = stats.replayed,
         failed = stats.failed,
         dropped = stats.dropped,
+        superseded = stats.superseded,
         quarantine = ?quarantine,
         "Switched to the new epoch"
     );
@@ -299,6 +301,7 @@ pub(super) async fn swap_in<H: Host>(
     copy_heads: Vec<ChangeHash>,
     translator: Translator,
 ) -> Result<(Document, ReplayStats), String> {
+    let superseded = host.superseded();
     // The closure runs on the blocking pool; its replay numbers come back through this.
     let late_stats = std::sync::Arc::new(std::sync::Mutex::new(ReplayStats::default()));
     let late_out = std::sync::Arc::clone(&late_stats);
@@ -310,7 +313,14 @@ pub(super) async fn swap_in<H: Host>(
             }
             let mut new_doc = new_doc;
             let live_heads = live.get_heads();
-            let late = delta::replay(live, &copy_heads, &live_heads, &mut new_doc, &translator);
+            let late = delta::replay_newest(
+                live,
+                &copy_heads,
+                &live_heads,
+                &mut new_doc,
+                &translator,
+                superseded,
+            );
             if late.replayed + late.failed > 0 {
                 info!(
                     replayed = late.replayed,