remake: replays judged before they go in

A switching or rejoining node replays late writes as fresh changes
naming each write's user, and nothing judged them before they left: a
replay every node would refuse (a revoked capability, a second
envelope at a signed path, readers changed) was refused everywhere,
and with it everything the node wrote after it.

Under a policy, replay_newest now judges each group it commits as
every node will (rules::judge_held, in the new epoch); one they would
refuse is taken back (forked at the heads before it, same actor) and
its paths quarantined with the reason. replay_newest takes the epoch.

Review finding 3 (remake replays).

5 files changed, 122 insertions, 2 deletions

diff --git a/docs/remakes.md b/docs/remakes.md
index 8e585c6..b70835e 100644
--- a/docs/remakes.md
+++ b/docs/remakes.md
@@ -311,6 +311,10 @@ The swarm stays writable throughout.
   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.
+* **Under a policy** ([`access.md`](access.md)), a carried-over write is judged as every
+  node will judge it, in the new epoch: one they would refuse (its user may no longer
+  write there, say) goes to the quarantine instead, so it can't cut the carrying node
+  off.
 * **Granularity**: map entries are carried over per key. Lists and text are carried
   over whole, so concurrent edits of the same list on two nodes around a remake end as
   last-writer-wins.
diff --git a/src/remake/bench.rs b/src/remake/bench.rs
index 96f76d8..4c12838 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,
+            crate::document::replica::Epoch::GENESIS,
             "node",
             &Translator::default(),
             Superseded::Drop,
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 71f077a..9830c33 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -421,6 +421,7 @@ async fn finish<H: Host>(host: &H) {
                         &before,
                         &after,
                         live,
+                        target,
                         &node,
                         &translator,
                         superseded,
@@ -438,6 +439,7 @@ async fn finish<H: Host>(host: &H) {
                         &began,
                         &heads,
                         live,
+                        target,
                         &node,
                         &translator,
                         superseded,
diff --git a/src/remake/delta.rs b/src/remake/delta.rs
index 06508a3..32f5ff7 100644
--- a/src/remake/delta.rs
+++ b/src/remake/delta.rs
@@ -85,12 +85,15 @@ pub(crate) enum Superseded {
 /// stays as `dst` has it (`superseded` says what becomes of the write). Ties go to the
 /// 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.
+/// naming the other. Under a policy, each is judged as every node will judge it (in
+/// `epoch`): one they would refuse is taken back, its paths quarantined.
+#[allow(clippy::too_many_arguments)]
 pub(crate) fn replay_newest(
     src: &Document,
     before: &[ChangeHash],
     after: &[ChangeHash],
     dst: &mut Document,
+    epoch: crate::document::replica::Epoch,
     node: &str,
     translator: &Translator,
     superseded: Superseded,
@@ -140,8 +143,9 @@ pub(crate) fn replay_newest(
     }
     crate::document::replica::authored(dst, node);
     for ((at, user), paths) in by_time {
+        let before = dst.get_heads();
         let mut transaction = dst.transaction();
-        stats.absorb(replay_paths(src, &mut transaction, &paths, translator));
+        let group = replay_paths(src, &mut transaction, &paths, translator);
         let writer = crate::document::replica::Writer {
             node,
             user: user.as_deref(),
@@ -151,6 +155,27 @@ pub(crate) fn replay_newest(
             options = options.with_message(message);
         }
         transaction.commit_with(options);
+        // Judged as every node will judge it (`access::rules`): one they would refuse
+        // goes to the quarantine instead, or everything this node writes after it
+        // would be refused too.
+        match judged(dst, epoch, &before) {
+            Ok(()) => stats.absorb(group),
+            Err(why) => {
+                let actor = dst.get_actor().clone();
+                match dst.fork_at(&before) {
+                    Ok(kept) => *dst = kept.with_actor(actor),
+                    Err(error) => {
+                        warn!(error = %error, "Could not take back a replay every node would refuse");
+                        stats.absorb(group);
+                        continue;
+                    }
+                }
+                let reason = format!("every node would refuse it: {why}");
+                stats
+                    .quarantined
+                    .extend(paths.into_iter().map(|path| (path, reason.clone())));
+            }
+        }
     }
     if stats.superseded > 0 {
         info!(
@@ -166,6 +191,24 @@ pub(crate) fn replay_newest(
     stats
 }
 
+/// The change just committed to `doc` (the one head after `before`), judged under a
+/// policy as every node will judge it on receipt; why it is refused, if it is.
+fn judged(
+    doc: &Document,
+    epoch: crate::document::replica::Epoch,
+    before: &[ChangeHash],
+) -> Result<(), String> {
+    if !crate::access::policy::of(doc).enforced() {
+        return Ok(());
+    }
+    match doc.get_heads()[..] {
+        [hash] if doc.get_heads() != before => {
+            crate::access::rules::judge_held(doc, epoch, &hash).unwrap_or(Ok(()))
+        }
+        _ => Ok(()),
+    }
+}
+
 /// 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> {
@@ -667,6 +710,71 @@ mod tests {
         transaction.commit_with(automerge::transaction::CommitOptions::default().with_time(at));
     }
 
+    /// Under a policy, a replayed write every node would refuse (its user may no longer
+    /// write there in the new epoch) goes to the quarantine, not into the document; the
+    /// rest goes in.
+    #[test]
+    fn replays_every_node_would_refuse_quarantined() {
+        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.
+        let mut new = old.clone().with_actor(automerge::ActorId::random());
+        change(&mut new, |doc| {
+            store::write_path(
+                doc,
+                &path("_continuum/users/bob"),
+                &crate::access::registry::entry("root", 1, &[]),
+            )
+            .unwrap();
+        });
+        let mut commit = |user: &str, write: &str| {
+            let mut transaction = old.transaction();
+            store::write_path(&mut transaction, &path(write), &json!(1)).unwrap();
+            transaction.commit_with(
+                automerge::transaction::CommitOptions::default()
+                    .with_time(10)
+                    .with_message(format!("User: {user}")),
+            );
+        };
+        commit("alice", "notes/x");
+        commit("bob", "notes/y");
+        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.replayed, 1);
+        assert_eq!(stats.quarantined.len(), 1);
+        assert_eq!(stats.quarantined[0].0, path("notes/y"));
+        assert!(store::read_path(&new, &path("notes/x")).unwrap().is_some());
+        assert!(store::read_path(&new, &path("notes/y")).unwrap().is_none());
+    }
+
     /// 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]
@@ -699,6 +807,7 @@ mod tests {
             &base,
             &after,
             &mut new,
+            crate::document::replica::Epoch::GENESIS,
             "node",
             &Translator::default(),
             Superseded::Drop,
@@ -762,6 +871,7 @@ mod tests {
             &base,
             &after,
             &mut new,
+            crate::document::replica::Epoch::GENESIS,
             "node",
             &Translator::default(),
             superseded,
@@ -851,6 +961,7 @@ mod tests {
             &base,
             &after,
             &mut new,
+            crate::document::replica::Epoch::GENESIS,
             "node",
             &Translator::default(),
             Superseded::Drop,
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index dd08b38..34c492f 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -255,6 +255,7 @@ async fn switch<H: Host>(
                 &base,
                 &heads,
                 &mut doc,
+                expected,
                 &node,
                 &translator,
                 superseded,
@@ -328,6 +329,7 @@ pub(super) async fn swap_in<H: Host>(
                 &copy_heads,
                 &live_heads,
                 &mut new_doc,
+                target,
                 &node,
                 &translator,
                 superseded,