remake: replays keep their writers

A remake's replay writes each value as the user who wrote it,
splitting where several users wrote under a path, so replays are
judged as the original writes.

Auth plan M6 (remakes, 2.4).

6 files changed, 203 insertions, 28 deletions

diff --git a/docs/remakes.md b/docs/remakes.md
index 7b21b9e..cad429f 100644
--- a/docs/remakes.md
+++ b/docs/remakes.md
@@ -305,7 +305,9 @@ The swarm stays writable throughout.
 * **Last writer wins**: a write is carried over only if nothing newer touched its path,
   a path above it or one below it, in the new epoch; newer by the times the writers'
   clocks gave their changes. A carried-over write keeps the time it was made, so a
-  later carry-over judges by that. A write that loses is left out, logged (`Writes not
+  later carry-over judges by that, and the user it was made for (`GET /admin/changes`);
+  the carrying node commits it. Another member's own entry (its alive stamp, its
+  links) is carried over by that member only. A write that loses is left out, logged (`Writes not
   replayed: newer ones touched their paths`), or with `--superseded-writes quarantine`
   kept in the quarantine. Clocks that disagree (beyond `--clock-skew-warn`) can make
   the older write win.
diff --git a/src/document/replica.rs b/src/document/replica.rs
index ead42f6..a409969 100644
--- a/src/document/replica.rs
+++ b/src/document/replica.rs
@@ -385,7 +385,7 @@ pub(crate) const USER_TRAILER: &str = "User";
 
 impl Writer<'_> {
     /// The change's message: its trailers (`trailers`), if it has any.
-    fn message(&self) -> Option<String> {
+    pub(crate) fn message(&self) -> Option<String> {
         self.user.map(|user| format!("{USER_TRAILER}: {user}"))
     }
 }
@@ -481,7 +481,7 @@ fn run<T, E>(
 /// Makes `node` the author of `doc`'s changes from now on. Setting an author gives the
 /// document a new actor; this node keeps its own (`with_actor`), which it tracks
 /// (`RemakeState::own_actors`, the document file), as only it writes with it.
-fn authored(doc: &mut Document, node: &str) {
+pub(crate) fn authored(doc: &mut Document, node: &str) {
     if doc
         .get_author()
         .is_some_and(|author| author.as_bytes() == node.as_bytes())
diff --git a/src/remake/bench.rs b/src/remake/bench.rs
index fa9f358..96f76d8 100644
--- a/src/remake/bench.rs
+++ b/src/remake/bench.rs
@@ -33,6 +33,7 @@ fn replay(keys: usize, late: usize, newer: usize) {
             &base,
             &after,
             &mut target,
+            "node",
             &Translator::default(),
             Superseded::Drop,
         ));
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 1ee9508..71f077a 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -404,6 +404,7 @@ async fn finish<H: Host>(host: &H) {
     // The writes, under the lock, with the switch back to serving requests.
     let quarantined_whole = matches!(plan, Plan::Quarantine { .. });
     let superseded = host.superseded();
+    let node = host.node_id();
     let ((stats, unplaced), old) = host
         .finish_rejoin(old, move |live, old| {
             let mut stats = ReplayStats::default();
@@ -420,6 +421,7 @@ async fn finish<H: Host>(host: &H) {
                         &before,
                         &after,
                         live,
+                        &node,
                         &translator,
                         superseded,
                     ));
@@ -436,6 +438,7 @@ async fn finish<H: Host>(host: &H) {
                         &began,
                         &heads,
                         live,
+                        &node,
                         &translator,
                         superseded,
                     ));
diff --git a/src/remake/delta.rs b/src/remake/delta.rs
index 6864c4d..06508a3 100644
--- a/src/remake/delta.rs
+++ b/src/remake/delta.rs
@@ -21,6 +21,11 @@
 //! The protocol's own metadata (`_continuum/remakes`, `_continuum/epochs`) is never
 //! replayed: the snapshot owns it.
 //!
+//! Replayed changes are the replaying node's, and name the user each write was made
+//! for (`replica::Writer`), as rules by path judge them (`access::rules`): they are
+//! grouped by user as by time. Another member's entry written without a user (its
+//! alive stamp, its links) is left to that member to replay: only it may write 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
@@ -78,13 +83,15 @@ pub(crate) enum Superseded {
 /// `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.
+/// replay. The replayed paths are committed by `node`, grouped by the time of the writes
+/// they replay and the user each was made for, each change stamped with the one and
+/// naming the other.
 pub(crate) fn replay_newest(
     src: &Document,
     before: &[ChangeHash],
     after: &[ChangeHash],
     dst: &mut Document,
+    node: &str,
     translator: &Translator,
     superseded: Superseded,
 ) -> ReplayStats {
@@ -92,15 +99,19 @@ pub(crate) fn replay_newest(
     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
+    // Split where several users wrote below a path: each part replayed for its own.
+    let paths = by_user(src, dst, &written, changed_paths(src, before, after));
+    let ours: Vec<(i64, Option<String>)> = paths
         .iter()
-        .map(|path| newest(&written, path).unwrap_or(0))
+        .map(|path| {
+            let at = newest(&written, path).map_or(0, |(at, _)| at);
+            (at, writer(&written, path))
+        })
         .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 oldest = ours.iter().map(|(at, _)| *at).min().unwrap_or(0);
     let newer: Vec<ChangeHash> = dst
         .get_changes(&[])
         .iter()
@@ -109,10 +120,13 @@ pub(crate) fn replay_newest(
         .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) {
+    let mut by_time: BTreeMap<(i64, Option<String>), Vec<Vec<String>>> = BTreeMap::new();
+    for (path, (at, user)) in paths.into_iter().zip(ours) {
+        if user.is_none() && others_entry(&path, node) {
+            continue;
+        }
         match newest(&theirs, &path) {
-            Some(newer) if newer > at => {
+            Some((newer, _)) if newer > at => {
                 stats.superseded += 1;
                 if superseded == Superseded::Quarantine {
                     stats.quarantined.push((
@@ -121,13 +135,22 @@ pub(crate) fn replay_newest(
                     ));
                 }
             }
-            _ => by_time.entry(at).or_default().push(path),
+            _ => by_time.entry((at, user)).or_default().push(path),
         }
     }
-    for (at, paths) in by_time {
+    crate::document::replica::authored(dst, node);
+    for ((at, user), 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));
+        let writer = crate::document::replica::Writer {
+            node,
+            user: user.as_deref(),
+        };
+        let mut options = automerge::transaction::CommitOptions::default().with_time(at);
+        if let Some(message) = writer.message() {
+            options = options.with_message(message);
+        }
+        transaction.commit_with(options);
     }
     if stats.superseded > 0 {
         info!(
@@ -163,32 +186,99 @@ fn between(doc: &Document, before: &[ChangeHash], after: &[ChangeHash]) -> Vec<C
     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>>)> {
+/// A change's time, the user it names (`replica::user_of`), and the store paths it
+/// touched.
+type Touched = (i64, Option<String>, Vec<Vec<String>>);
+
+/// For each of `changes` (in `doc`), its time, its user 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<Touched> {
     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))
+            let user = change
+                .message()
+                .and_then(|message| crate::document::replica::user_of(message))
+                .map(str::to_owned);
+            Some((change.timestamp(), user, 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> {
+/// The time and user of the newest of `changes` that touched `path`, a path above it or
+/// one below it (of two at once, the user named last in order).
+fn newest(changes: &[Touched], path: &[String]) -> Option<(i64, Option<String>)> {
     changes
         .iter()
-        .filter(|(_, paths)| {
+        .filter(|(_, _, paths)| {
             paths
                 .iter()
                 .any(|other| other.starts_with(path) || path.starts_with(other))
         })
-        .map(|(at, _)| *at)
+        .map(|(at, user, _)| (*at, user.clone()))
+        .max()
+}
+
+/// The user of the newest of `changes` that wrote at `path` or below it: whose write it
+/// is (of two at once, the one named last in order).
+fn writer(changes: &[Touched], path: &[String]) -> Option<String> {
+    changes
+        .iter()
+        .filter(|(_, _, paths)| paths.iter().any(|other| other.starts_with(path)))
+        .map(|(at, user, _)| (*at, user.clone()))
         .max()
+        .and_then(|(_, user)| user)
+}
+
+/// `paths`, each a map that writes for more than one user changed below split into its
+/// entries (in `src` or `dst`), until each part is one user's: a replayed change names
+/// one user, and rules by path judge it by that user (`access::rules`).
+fn by_user(
+    src: &Document,
+    dst: &Document,
+    changes: &[Touched],
+    paths: Vec<Vec<String>>,
+) -> Vec<Vec<String>> {
+    let mut out = Vec::new();
+    let mut todo = paths;
+    while let Some(path) = todo.pop() {
+        let users: HashSet<&Option<String>> = changes
+            .iter()
+            .filter(|(_, _, paths)| paths.iter().any(|other| other.starts_with(&path)))
+            .map(|(_, user, _)| user)
+            .collect();
+        let keys = store::keys_at(src, &path).ok().flatten();
+        match keys {
+            Some(mut keys) if users.len() > 1 => {
+                keys.extend(
+                    store::keys_at(dst, &path)
+                        .ok()
+                        .flatten()
+                        .unwrap_or_default(),
+                );
+                keys.sort();
+                keys.dedup();
+                for key in keys {
+                    let mut child = path.clone();
+                    child.push(key);
+                    todo.push(child);
+                }
+            }
+            _ => out.push(path),
+        }
+    }
+    out.sort();
+    out
+}
+
+/// Whether `path` is in another member's entry than `node`'s (`_continuum/members/<pk>`):
+/// only that member writes it of its own (`access::rules`).
+fn others_entry(path: &[String], node: &str) -> bool {
+    matches!(path, [root, members, owner, ..]
+        if root == meta::ROOT && members == meta::MEMBERS && owner != node)
 }
 
 /// Replays into `dst` everything that changed in `src` between `before` and `after`,
@@ -577,6 +667,72 @@ mod tests {
         transaction.commit_with(automerge::transaction::CommitOptions::default().with_time(at));
     }
 
+    /// Replayed changes are the replaying node's and name each write's user; another
+    /// member's entry, written without a user, is left to that member.
+    #[test]
+    fn replays_keep_their_writers() {
+        let path = |text: &str| -> Vec<String> { text.split('/').map(str::to_owned).collect() };
+        let mut old = Document::new();
+        change(&mut old, |doc| {
+            store::write_path(doc, &path("_continuum/members/node"), &json!({})).unwrap();
+            store::write_path(doc, &path("_continuum/members/other"), &json!({})).unwrap();
+        });
+        let base = old.get_heads();
+        let mut new = old.clone().with_actor(automerge::ActorId::random());
+        let mut commit = |at: i64, user: Option<&str>, write: &str, value: serde_json::Value| {
+            let mut transaction = old.transaction();
+            store::write_path(&mut transaction, &path(write), &value).unwrap();
+            let mut options = automerge::transaction::CommitOptions::default().with_time(at);
+            if let Some(user) = user {
+                options = options.with_message(format!("User: {user}"));
+            }
+            transaction.commit_with(options);
+        };
+        commit(10, Some("alice"), "notes/x", json!(1));
+        commit(10, Some("bob"), "notes/y", json!(2));
+        commit(11, None, "_continuum/members/other/alive", json!(11));
+        commit(11, None, "_continuum/members/node/alive", json!(11));
+        let after = old.get_heads();
+        let new_heads = new.get_heads();
+        let stats = replay_newest(
+            &old,
+            &base,
+            &after,
+            &mut new,
+            "node",
+            &Translator::default(),
+            Superseded::Drop,
+        );
+        assert_eq!(stats.replayed, 3);
+        assert!(
+            store::read_path(&new, &path("_continuum/members/other/alive"))
+                .unwrap()
+                .is_none()
+        );
+        assert_eq!(
+            store::read_path(&new, &path("_continuum/members/node/alive")).unwrap(),
+            Some(json!(11))
+        );
+        // One change per user and time, each by the node, naming its user.
+        let replayed: Vec<(Option<String>, i64)> = new
+            .get_changes(&new_heads)
+            .iter()
+            .map(|change| (change.message().map(str::to_owned), change.timestamp()))
+            .collect();
+        assert_eq!(
+            replayed,
+            vec![
+                (Some("User: alice".to_owned()), 10),
+                (Some("User: bob".to_owned()), 10),
+                (None, 11),
+            ]
+        );
+        assert_eq!(
+            new.get_author().map(|author| author.as_bytes().to_vec()),
+            Some(b"node".to_vec())
+        );
+    }
+
     /// 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(
@@ -606,6 +762,7 @@ mod tests {
             &base,
             &after,
             &mut new,
+            "node",
             &Translator::default(),
             superseded,
         );
@@ -694,6 +851,7 @@ mod tests {
             &base,
             &after,
             &mut new,
+            "node",
             &Translator::default(),
             Superseded::Drop,
         );
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index 4c72b39..dd08b38 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -216,6 +216,7 @@ async fn switch<H: Host>(
     let actor = ActorId::random();
     host.remake().add_own_actor(actor.clone());
     let new_actor = actor.clone();
+    let node = host.node_id();
 
     // 2. Build (or reuse), verify and replay without holding any lock.
     let (new_doc, copy_heads, mut stats) = tokio::task::spawn_blocking(
@@ -233,8 +234,9 @@ async fn switch<H: Host>(
                 ));
             }
 
-            // A fresh actor: this node's writes in the new epoch.
+            // A fresh actor: this node's writes in the new epoch, made by it.
             let mut doc = snapshot.doc.with_actor(actor);
+            crate::document::replica::authored(&mut doc, &node);
 
             let heads = copy.get_heads();
 
@@ -248,8 +250,15 @@ async fn switch<H: Host>(
             }
 
             // 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);
+            let stats = delta::replay_newest(
+                &copy,
+                &base,
+                &heads,
+                &mut doc,
+                &node,
+                &translator,
+                superseded,
+            );
             Ok((doc, heads, stats))
         },
     )
@@ -302,6 +311,7 @@ pub(super) async fn swap_in<H: Host>(
     translator: Translator,
 ) -> Result<(Document, ReplayStats), String> {
     let superseded = host.superseded();
+    let node = host.node_id();
     // 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);
@@ -318,6 +328,7 @@ pub(super) async fn swap_in<H: Host>(
                 &copy_heads,
                 &live_heads,
                 &mut new_doc,
+                &node,
                 &translator,
                 superseded,
             );