swarm: changes taken unjudged, dropped (REFUSALS)

A node refusing changes a peer brought tells it which broke a rule; the
peer judges them itself, and if its rules refuse them too, drops them
with all built on them (its document forked without them, a fresh
actor), quarantining its own writes among them. Wire version 42.

Auth plan M6 (rejoin on taint).

1 files changed, 83 insertions, 9 deletions

diff --git a/src/simtest/policy.rs b/src/simtest/policy.rs
index eb706bc..5ad05fd 100644
--- a/src/simtest/policy.rs
+++ b/src/simtest/policy.rs
@@ -90,15 +90,6 @@ async fn a_swarm_judging_every_change() {
         })
         .await;
         assert!(!holds(node, "notes/forged").await);
-        // Its own change, signed: c refused as a whole, sent nothing more.
-        use crate::locks::MutexExt;
-        assert!(
-            node.screening
-                .lock_or_recover()
-                .refusing(c.local_id())
-                .is_some(),
-            "c isn't refused as a whole"
-        );
     }
 }
 
@@ -404,3 +395,86 @@ async fn sealed_values() {
         assert!(!holds(node, "secrets/api").await);
     }
 }
+
+/// A node that took a change the rules refuse, unjudged (as under a policy it didn't
+/// know yet): the others refuse what it writes on it, tell it so (REFUSALS), and it
+/// drops the change and what it built on it, its own write among them quarantined, and
+/// what it writes from then on goes everywhere.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn a_change_taken_unjudged() {
+    let sim = Sim::with_identity(75);
+    let a = sim.start_with("a", &[], Start::Genesis, |config| {
+        config.first_users = [("alice".to_owned(), Default::default())].into();
+    });
+    let b = sim.start("b", &["a"], Start::Join);
+    let c = sim.start("c", &["a"], Start::Join);
+    sim.wait_ready(&[&a, &b, &c], SOON).await;
+    let nodes = [a.clone(), b.clone(), c.clone()];
+    sim.wait_until("every node is registered", SOON, || {
+        let nodes = nodes.clone();
+        async move {
+            for node in &nodes {
+                if crate::swarm::members::members(&node.doc.read().await.doc).len() < 3 {
+                    return false;
+                }
+            }
+            true
+        }
+    })
+    .await;
+    write_as(
+        &a,
+        "alice",
+        "_continuum/access/policy",
+        json!({"version": 1}),
+    )
+    .await;
+    sim.wait_converged(&[&a, &b, &c], SOON).await;
+
+    // A change of a node that isn't a member, into c's document as if received
+    // unjudged.
+    let forged = {
+        let mut doc = c.doc.read().await.doc.fork();
+        let writer = crate::document::replica::Writer {
+            node: "continuum:nobody",
+            user: Some("alice"),
+        };
+        crate::document::replica::transact(&mut doc, crate::clock::UnixMillis(1), &writer, |doc| {
+            store::write_path(doc, &path("notes/forged"), &json!(1))
+        })
+        .unwrap();
+        doc.get_last_local_change().unwrap().clone()
+    };
+    {
+        let doc = std::sync::Arc::clone(&c.doc);
+        tokio::task::spawn_blocking(move || {
+            let mut replica = doc.blocking_write();
+            replica.doc.apply_changes([forged]).unwrap();
+            replica.refresh_heads();
+        })
+        .await
+        .unwrap();
+    }
+    assert!(holds(&c, "notes/forged").await);
+    // c's write builds on it: refused by the others, who tell c.
+    write_as(&c, "alice", "notes/after", json!(2)).await;
+    sim.wait_until("c drops the change taken unjudged", SOON, || async {
+        !holds(&c, "notes/forged").await
+    })
+    .await;
+    assert!(
+        !holds(&c, "notes/after").await,
+        "built on the change dropped"
+    );
+    assert_eq!(c.remake.list_quarantine(|entry| entry.id.clone()).len(), 1);
+
+    write_as(&c, "alice", "notes/later", json!(3)).await;
+    for node in [&a, &b] {
+        sim.wait_until("c's later write is everywhere", SOON, move || async move {
+            holds(node, "notes/later").await
+        })
+        .await;
+        assert!(!holds(node, "notes/forged").await);
+        assert!(!holds(node, "notes/after").await);
+    }
+}