remake: a replay taken back keeps the node its author

Review 2, N4: replay_newest names the node the document's author once,
before its first group. Taking a refused group back forks the document
at the heads before it, and Automerge's fork has no author; when the
refused group was the actor's first change, the next one was committed
as seq 1 without the footer naming the node, refused as "it names no
node", taken back too, and so on: the whole replay in the quarantine
with a misleading reason.

The fork now names the node its author again. Test: the first group
refused (a capability revoked in the new epoch), the next replayed.

1 files changed, 72 insertions, 1 deletions

diff --git a/src/remake/delta.rs b/src/remake/delta.rs
index 32f5ff7..dd5df4f 100644
--- a/src/remake/delta.rs
+++ b/src/remake/delta.rs
@@ -163,7 +163,12 @@ pub(crate) fn replay_newest(
             Err(why) => {
                 let actor = dst.get_actor().clone();
                 match dst.fork_at(&before) {
-                    Ok(kept) => *dst = kept.with_actor(actor),
+                    // The author too: a fork has none, and the next group may be the
+                    // actor's first change, whose footer names it.
+                    Ok(kept) => {
+                        *dst = kept.with_actor(actor);
+                        crate::document::replica::authored(dst, node);
+                    }
                     Err(error) => {
                         warn!(error = %error, "Could not take back a replay every node would refuse");
                         stats.absorb(group);
@@ -775,6 +780,72 @@ mod tests {
         assert!(store::read_path(&new, &path("notes/y")).unwrap().is_none());
     }
 
+    /// The first group refused: those after it still replayed, still the node's (a
+    /// fork, taking the refused one back, has no author of its own).
+    #[test]
+    fn replays_after_a_first_one_refused() {
+        let path = |text: &str| -> Vec<String> { text.split('/').map(str::to_owned).collect() };
+        let alice = crate::access::registry::entry(
+            "root",
+            1,
+            &[(path("notes"), &[crate::access::Right::Write][..])],
+        );
+        let mut old = Document::new();
+        change(&mut old, |doc| {
+            store::write_path(doc, &path("_continuum/swarm"), &json!("s")).unwrap();
+            store::write_path(
+                doc,
+                &path("_continuum/access/policy"),
+                &json!({"version": 1}),
+            )
+            .unwrap();
+            store::write_path(doc, &path("_continuum/members/node"), &json!({})).unwrap();
+            store::write_path(doc, &path("_continuum/users/alice"), &alice).unwrap();
+            store::write_path(doc, &path("_continuum/users/bob"), &alice).unwrap();
+        });
+        crate::document::replica::authored(&mut old, "node");
+        let base = old.get_heads();
+        // In the new epoch, bob may write nothing any more: by another actor, so the
+        // replaying one's first change is the first group.
+        let mut revoker = old.clone().with_actor(automerge::ActorId::random());
+        change(&mut revoker, |doc| {
+            store::write_path(
+                doc,
+                &path("_continuum/users/bob"),
+                &crate::access::registry::entry("root", 1, &[]),
+            )
+            .unwrap();
+        });
+        let mut new = old.clone().with_actor(automerge::ActorId::random());
+        new.merge(&mut revoker).unwrap();
+        let mut commit = |user: &str, write: &str, at: i64| {
+            let mut transaction = old.transaction();
+            store::write_path(&mut transaction, &path(write), &json!(1)).unwrap();
+            transaction.commit_with(
+                automerge::transaction::CommitOptions::default()
+                    .with_time(at)
+                    .with_message(format!("User: {user}")),
+            );
+        };
+        commit("bob", "notes/y", 5);
+        commit("alice", "notes/x", 10);
+        let after = old.get_heads();
+        let stats = replay_newest(
+            &old,
+            &base,
+            &after,
+            &mut new,
+            crate::document::replica::Epoch::GENESIS,
+            "node",
+            &Translator::default(),
+            Superseded::Drop,
+        );
+        assert_eq!(stats.quarantined.len(), 1, "{:?}", stats.quarantined);
+        assert_eq!(stats.quarantined[0].0, path("notes/y"));
+        assert_eq!(stats.replayed, 1);
+        assert!(store::read_path(&new, &path("notes/x")).unwrap().is_some());
+    }
+
     /// 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]