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).

16 files changed, 462 insertions, 38 deletions

diff --git a/docs/access.md b/docs/access.md
index 31efe4e..44c89f3 100644
--- a/docs/access.md
+++ b/docs/access.md
@@ -246,7 +246,18 @@ from then on builds on what was refused, and logs it once. It shows in `GET /pee
 (`refused`, with the change and why), in `/admin/stats` (`sync.refusedPeers`) and on the
 dashboard. A node that only passed a refused change on isn't: the committer is, whoever
 brought it. It lasts until a change of the member's is taken again (it rejoined and
-writes anew), or the epoch changes; forgetting or banning it is the operators'. A change whose
+writes anew), or the epoch changes; forgetting or banning it is the operators'.
+
+A node takes a change the others refuse only if it didn't judge it as they do: under a
+policy it didn't know yet (it receives without judging then), or before the policy
+came. Everything it writes from then on builds on that change, and is refused too. So
+a node refusing changes a peer brought tells it which broke a rule (REFUSALS, at most
+every 30 seconds). The peer doesn't take the word for it: it judges each it holds
+itself, by its rules as they are now. If they refuse one too, it drops it with
+everything built on it, its document becoming what it would be without them, and puts
+its own writes among them in the quarantine (`GET /admin/quarantine`), to be replayed
+or discarded; what it writes from then on goes everywhere. If its rules take the
+change, the two policies disagree, which it logs. A change whose
 dependencies haven't come yet waits for them.
 
 What it doesn't do yet:
diff --git a/docs/architecture.md b/docs/architecture.md
index 1713d40..ef7928e 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -77,7 +77,7 @@ it switched to the right document without trusting anyone.
 ## Modules
 
 **[`transport/`](../src/transport/)** — [`mod.rs`](../src/transport/mod.rs) defines the message types (HELLO, HELLO_REPLY, sync
-envelope, STATUS/STATUS_REPLY, PING/PONG, blob messages, PROBE/PROBE_REPLY, SIGNATURES) and the `Transport` trait;
+envelope, STATUS/STATUS_REPLY, PING/PONG, blob messages, PROBE/PROBE_REPLY, SIGNATURES, REFUSALS) and the `Transport` trait;
 [`wire.rs`](../src/transport/wire.rs) is the wire format (a version byte, bumped with every change to the format,
 and a kind byte, then the message in MessagePack, derived from the types in [`mod.rs`](../src/transport/mod.rs)
 with codecs for hashes and IDs); [`mycelium/`](../src/transport/mycelium/) (with the `mycelium` feature) is the REST
diff --git a/docs/measurements.md b/docs/measurements.md
index 8e36a7e..4550357 100644
--- a/docs/measurements.md
+++ b/docs/measurements.md
@@ -181,7 +181,7 @@ Messages and payload bytes by direction (`sent`, `received`, `refused`) and kind
 (`hello`, `helloReply`, `sync`, `status`, `statusReply`, `blob`, `identity` for WHO and
 ME, `paths` for PING and PONG, `probe` for PROBE
 and PROBE_REPLY, `plumtree` for PRUNE and GRAFT, `hyparview` for NEIGHBOUR,
-NEIGHBOUR_REPLY and DISCONNECT, `signatures` for SIGNATURES). The layers below the node count WHO and ME, and the
+NEIGHBOUR_REPLY and DISCONNECT, `signatures` for SIGNATURES, `refusals` for REFUSALS). The layers below the node count WHO and ME, and the
 paths layer's own PINGs and PONGs, under the address they went to or came from; a peer
 (`/peers`) has those of its addresses too, where it is known to be at them. Payload bytes are what a transport's message carries (the wire
 encoding): not the transport's framing (for Mycelium, its topic and the base64 of its
diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index 832d941..9b75618 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -2230,7 +2230,7 @@ components:
 
     ByKind:
       type: object
-      required: [hello, helloReply, sync, status, statusReply, blob, identity, paths, probe, plumtree, hyparview, signatures]
+      required: [hello, helloReply, sync, status, statusReply, blob, identity, paths, probe, plumtree, hyparview, signatures, refusals]
       properties:
         hello:
           $ref: '#/components/schemas/Counter'
@@ -2263,6 +2263,9 @@ components:
         signatures:
           $ref: '#/components/schemas/Counter'
           description: SIGNATURES, signatures of changes no sync message took.
+        refusals:
+          $ref: '#/components/schemas/Counter'
+          description: REFUSALS, the changes a peer brought that broke the swarm's rules.
 
     Traffic:
       type: object
diff --git a/src/access/screen.rs b/src/access/screen.rs
index 6773d05..8d23b47 100644
--- a/src/access/screen.rs
+++ b/src/access/screen.rs
@@ -76,20 +76,26 @@ pub(crate) struct Refusing {
     pub(crate) why: String,
 }
 
-/// The hashes of the changes refused, to refuse them again by hash.
+/// The hashes of the changes refused, to refuse them again by hash, each with the one
+/// that broke a rule (itself, or the one it builds on).
 #[derive(Debug, Default)]
 pub(crate) struct Refusals {
-    known: HashSet<ChangeHash>,
+    known: HashMap<ChangeHash, ChangeHash>,
     order: std::collections::VecDeque<ChangeHash>,
 }
 
 impl Refusals {
     fn contains(&self, hash: &ChangeHash) -> bool {
-        self.known.contains(hash)
+        self.known.contains_key(hash)
     }
 
-    fn insert(&mut self, hash: ChangeHash) {
-        if self.known.insert(hash) {
+    /// The change that broke a rule, of the refused change `hash`.
+    fn root(&self, hash: &ChangeHash) -> Option<ChangeHash> {
+        self.known.get(hash).copied()
+    }
+
+    fn insert(&mut self, hash: ChangeHash, root: ChangeHash) {
+        if self.known.insert(hash, root).is_none() {
             self.order.push_back(hash);
             if self.order.len() > MAX_REFUSALS
                 && let Some(oldest) = self.order.pop_front()
@@ -184,6 +190,8 @@ pub(crate) struct Screened {
     pub(crate) broke_rules: Vec<(String, ChangeHash, String)>,
     /// The committers of changes taken.
     pub(crate) taken_from: BTreeSet<String>,
+    /// The changes that broke a rule, of those refused (now, or again).
+    pub(crate) roots: BTreeSet<ChangeHash>,
 }
 
 /// Judges `received` (and those `waiting` before) and applies to `doc` (of `epoch`)
@@ -208,10 +216,13 @@ pub(crate) fn apply(
         .chain(received)
         .filter(|change| seen.insert(change.hash()))
         .filter(|change| doc.get_change_by_hash(&change.hash()).is_none())
-        .filter(|change| {
-            let again = refusals.contains(&change.hash());
-            screened.refused_again += usize::from(again);
-            !again
+        .filter(|change| match refusals.root(&change.hash()) {
+            Some(root) => {
+                screened.refused_again += 1;
+                screened.roots.insert(root);
+                false
+            }
+            None => true,
         })
         .collect();
     // Building on one refused before: refused too, without waiting for it (it never
@@ -226,7 +237,9 @@ pub(crate) fn apply(
             .find(|dep| refusals.contains(dep))
             .copied()
             .expect("partitioned so");
-        refusals.insert(change.hash());
+        let root = refusals.root(&dep).unwrap_or(dep);
+        refusals.insert(change.hash(), root);
+        screened.roots.insert(root);
         screened.refused.push(Refused {
             committer: change.author().and_then(|author| {
                 std::str::from_utf8(author.as_bytes())
@@ -281,7 +294,8 @@ pub(crate) fn apply(
 
     // The changes of the document each accepted one, not applied yet, builds on.
     let mut frontier: HashMap<ChangeHash, Vec<ChangeHash>> = HashMap::new();
-    let mut refused: HashSet<ChangeHash> = HashSet::new();
+    // Refused in this batch, each with the change that broke a rule.
+    let mut refused: HashMap<ChangeHash, ChangeHash> = HashMap::new();
     let mut unsigned: HashSet<ChangeHash> = HashSet::new();
     let mut accepted: Vec<Change> = Vec::new();
     for (index, change) in ready.iter().enumerate() {
@@ -298,9 +312,14 @@ pub(crate) fn apply(
         if let Some(dep) = change
             .deps()
             .iter()
-            .find(|dep| refused.contains(dep) || refusals.contains(dep))
+            .find(|dep| refused.contains_key(dep) || refusals.contains(dep))
         {
-            refused.insert(hash);
+            let root = refused
+                .get(dep)
+                .copied()
+                .or_else(|| refusals.root(dep))
+                .unwrap_or(*dep);
+            refused.insert(hash, root);
             screened
                 .refused
                 .push(refuse(format!("it builds on refused change {dep}")));
@@ -359,11 +378,11 @@ pub(crate) fn apply(
             }
             Err(Refusal::Unready) => {
                 // Can't be: its dependencies are in the document or among these.
-                refused.insert(hash);
+                refused.insert(hash, hash);
                 screened.refused.push(refuse(Refusal::Unready.to_string()));
             }
             Err(Refusal::Rule(why)) => {
-                refused.insert(hash);
+                refused.insert(hash, hash);
                 // Its committer's own, as a signature by it vouches.
                 if signatures.is_some()
                     && let Some(committer) = committer
@@ -380,21 +399,24 @@ pub(crate) fn apply(
     if !accepted.is_empty() {
         doc.apply_changes_log_patches(accepted, log)?;
     }
-    for hash in refused {
-        refusals.insert(hash);
+    for (hash, root) in refused {
+        refusals.insert(hash, root);
+        screened.roots.insert(root);
     }
     waiting.0.truncate(MAX_WAITING);
     Ok(screened)
 }
 
 /// What `apply_bytes` did.
-#[derive(Debug, Default, Clone, Copy)]
+#[derive(Debug, Default, Clone)]
 pub(crate) struct Outcome {
     /// Changes applied (received, or waiting before).
     pub(crate) applied: usize,
     /// Whether the bytes carried changes, and all of them refused before: their sender
     /// offers again what the node never takes, and its sync session backs off.
     pub(crate) refused_again: bool,
+    /// The changes that broke a rule, of those refused (`swarm::refusals`).
+    pub(crate) roots: Vec<ChangeHash>,
 }
 
 /// [`apply`] of the changes in `bytes`, as sync messages and downloads carry them, with
@@ -439,6 +461,7 @@ pub(crate) fn apply_bytes(
     let outcome = Outcome {
         applied: screened.applied,
         refused_again: received > 0 && screened.refused_again == received,
+        roots: screened.roots.iter().copied().collect(),
     };
     screening.judge_members(&screened);
     screening.record(screened.refused);
diff --git a/src/document/sync.rs b/src/document/sync.rs
index 823d4a3..b172dd1 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -212,6 +212,9 @@ pub(crate) struct SyncSession {
     /// The wait before answering a peer that offers again only changes we refused
     /// (`refused_again`).
     refused_backoff: Backoff,
+    /// When we last told the peer which of the changes it brought we refused
+    /// (`refusals_due`).
+    refusals_told: Option<Instant>,
 
     /// The answer we expect to our last sync message(s), if any.
     pub(crate) reply: Option<SyncReplyWait>,
@@ -264,6 +267,7 @@ impl SyncSession {
             assumed_lately: std::collections::VecDeque::new(),
             said_heads: None,
             refused_backoff: Backoff::new(REFUSED_WAIT, REFUSED_WAIT_MAX),
+            refusals_told: None,
             reply: None,
             reply_backoff,
         }
@@ -993,6 +997,18 @@ impl SyncSession {
         delay
     }
 
+    /// Whether to tell the peer which of the changes it brought we refused, now
+    /// (`swarm::refusals`): at most every `refusals::TELL_EVERY`.
+    pub(crate) fn refusals_due(&mut self, now: Instant) -> bool {
+        let due = self.refusals_told.is_none_or(|told| {
+            now.saturating_duration_since(told) >= crate::swarm::refusals::TELL_EVERY
+        });
+        if due {
+            self.refusals_told = Some(now);
+        }
+        due
+    }
+
     /// The peer brought changes we took (or none refused before): answered at once again.
     pub(crate) fn refusals_settled(&mut self) {
         self.refused_backoff.reset();
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index 342eab3..840322a 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -39,6 +39,23 @@ mod vote;
 
 #[cfg(test)]
 pub(crate) use api::testing;
+
+/// Quarantines `paths` of `document` (of `epoch`): this node's writes on a change the
+/// swarm's rules refuse, which it took unjudged and has dropped since
+/// (`AppState::drop_taken`). The quarantine's id, if any.
+pub(crate) fn quarantine_taken<H: Host>(
+    host: &H,
+    epoch: crate::document::replica::Epoch,
+    document: &crate::document::Document,
+    paths: Vec<Vec<String>>,
+) -> Option<String> {
+    let reason = "written on a change the swarm's rules refuse, which this node took unjudged";
+    let quarantined = paths
+        .into_iter()
+        .map(|path| (path, reason.to_owned()))
+        .collect();
+    quarantine::add(host, epoch, epoch, document, quarantined)
+}
 pub(crate) use delta::Superseded;
 pub(crate) use kind::remake_genesis;
 pub(crate) use local::RemakeState;
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);
+    }
+}
diff --git a/src/state/host.rs b/src/state/host.rs
index fb2e2e9..d23e07d 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -8,7 +8,7 @@ use std::time::Duration;
 use anyhow::Result;
 use tokio::sync::broadcast;
 use tokio::time::Instant;
-use tracing::debug;
+use tracing::{debug, error, info};
 
 use super::{AppState, HelloState, NodeLifecycle, StaleView, Standing};
 use crate::document::replica::{Epoch, Replica};
@@ -90,6 +90,90 @@ impl AppState {
         (doc, true)
     }
 
+    /// Drops `root`, a change the swarm's rules refuse that this node took unjudged
+    /// (`swarm::refusals`), and every change built on it: the document becomes the one
+    /// without them (forked where none is), written by a fresh actor of this node's, as
+    /// the old one's later changes are among those dropped; this node's own writes among
+    /// them go to the quarantine. Every peer syncs anew. Nothing while a rejoin runs, or
+    /// if the document doesn't hold `root`.
+    pub(crate) async fn drop_taken(&self, root: ChangeHash) {
+        let own = self.remake.own_actors();
+        let state = self.clone();
+        let dropped = tokio::task::spawn_blocking(move || {
+            let mut replica = state.doc.blocking_write();
+            if !replica.writable {
+                return None;
+            }
+            let changes = replica.doc.get_changes(&[]);
+            if !changes.iter().any(|change| change.hash() == root) {
+                return None;
+            }
+            // In causal order: a change after those it builds on.
+            let mut tainted: std::collections::HashSet<ChangeHash> = [root].into();
+            for change in &changes {
+                if change.deps().iter().any(|dep| tainted.contains(dep)) {
+                    tainted.insert(change.hash());
+                }
+            }
+            let kept: Vec<_> = changes
+                .iter()
+                .filter(|change| !tainted.contains(&change.hash()))
+                .collect();
+            let built_on: std::collections::HashSet<ChangeHash> = kept
+                .iter()
+                .flat_map(|change| change.deps().iter().copied())
+                .collect();
+            let heads: Vec<ChangeHash> = kept
+                .iter()
+                .map(|change| change.hash())
+                .filter(|hash| !built_on.contains(hash))
+                .collect();
+            // This node's own writes among those dropped: the paths they touched.
+            let mut actors = own;
+            actors.push(replica.doc.get_actor().clone());
+            let ours: Vec<crate::document::ExpandedChange> = changes
+                .iter()
+                .filter(|change| tainted.contains(&change.hash()))
+                .filter(|change| actors.contains(change.actor_id()))
+                .map(|change| change.decode())
+                .collect();
+            let mut paths: Vec<Vec<String>> =
+                crate::document::inspect::Inspector::new(&replica.doc)
+                    .ops(&ours)
+                    .into_iter()
+                    .flatten()
+                    .map(|op| op.path)
+                    .collect();
+            paths.sort();
+            paths.dedup();
+            let actor = crate::document::ActorId::random();
+            let fresh = match replica.doc.fork_at(&heads) {
+                Ok(fresh) => fresh.with_actor(actor.clone()),
+                Err(error) => {
+                    error!(error = %error, "Couldn't drop a change taken unjudged");
+                    return None;
+                }
+            };
+            state.remake.add_own_actor(actor);
+            let epoch = replica.epoch;
+            let old = state.swap_document(&mut replica, epoch, fresh);
+            Some((epoch, old, paths, tainted.len()))
+        })
+        .await
+        .ok()
+        .flatten();
+        let Some((epoch, old, paths, count)) = dropped else {
+            return;
+        };
+        info!(
+            changes = count,
+            own_paths = paths.len(),
+            "Dropped a change taken unjudged and what built on it"
+        );
+        crate::remake::quarantine_taken(self, epoch, &old, paths);
+        self.after_epoch_switch().await;
+    }
+
     /// After the document was replaced: every peer's sync state refers to the old
     /// document, so start over with all of them. Peers still on the old epoch are
     /// skipped by the sync worker until they switch too.
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index 0cb4fb8..d872474 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -15,6 +15,7 @@ pub(crate) mod pair;
 pub(crate) mod peer;
 pub(crate) mod plumtree;
 pub(crate) mod reach;
+pub(crate) mod refusals;
 pub(crate) mod relay;
 pub(crate) mod rtt;
 pub(crate) mod signatures;
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 9b5db8f..a7633b3 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -272,6 +272,16 @@ pub(crate) async fn handle_message(
                 crate::swarm::signatures::received(state, message).await;
             }
         }
+        transport::ParsedMessage::Refusals(message) => {
+            // From members, of our epoch: judging costs (`swarm::refusals`).
+            let member = matches!(
+                peer_state.lock().await.standing,
+                crate::state::Standing::Member { .. }
+            );
+            if member && same_epoch {
+                crate::swarm::refusals::received(state, &peer, message).await;
+            }
+        }
         transport::ParsedMessage::Probe(request) => {
             crate::swarm::working_links::answer(state, &peer, request).await;
         }
diff --git a/src/swarm/refusals.rs b/src/swarm/refusals.rs
new file mode 100644
index 0000000..2e5399a
--- /dev/null
+++ b/src/swarm/refusals.rs
@@ -0,0 +1,139 @@
+//! Telling a peer which of the changes it brought were refused (REFUSALS), so a node
+//! that took a change the swarm's rules refuse finds out and sets itself right.
+//!
+//! A node takes a change the others refuse only if it didn't judge it as they do: under
+//! a policy it didn't know yet (it receives without judging then, `access::policy`), or
+//! one that came after. Everything it writes from then on builds on that change and is
+//! refused too, and nobody tells it so. Hence: a node that refuses changes a peer
+//! brought tells it which broke a rule (`tell`), at most every `TELL_EVERY` and
+//! `MAX_LISTED` of them. The peer doesn't take the word for it (a member could tell
+//! anything): it judges each it holds itself, as its rules have them now
+//! (`received`). If its rules refuse one too, it took it unjudged, and drops it with
+//! everything built on it: its document as it would be without them, its own writes
+//! among them quarantined (`AppState::drop_taken`). If its rules take it, the two
+//! policies disagree, which it logs; finding out which is right is left
+//! (`future-work.md`, Write protection by checking changes).
+
+use std::time::Duration;
+
+use tracing::{debug, warn};
+
+use crate::document::replica::Epoch;
+use crate::document::{ChangeHash, ReadDoc};
+use crate::state::AppState;
+use crate::transport::{ParsedMessage, RefusalsMessage};
+
+/// Changes listed in one REFUSALS, at most.
+pub(crate) const MAX_LISTED: usize = 64;
+/// How often a peer is told, at most.
+pub(crate) const TELL_EVERY: Duration = Duration::from_secs(30);
+
+/// Tells `peer` that `changes` it brought, of our epoch, broke the swarm's rules.
+pub(crate) async fn tell(
+    state: &AppState,
+    peer: &str,
+    epoch: Epoch,
+    epoch_seq: u32,
+    mut changes: Vec<ChangeHash>,
+) {
+    changes.truncate(MAX_LISTED);
+    let message = ParsedMessage::Refusals(RefusalsMessage {
+        epoch,
+        epoch_seq,
+        changes,
+    });
+    if let Err(error) = state.transport.send_message(peer, &message).await {
+        // Told again with the next refusal.
+        debug!(peer = %peer, error = %format!("{error:#}"), "Couldn't tell refusals");
+    }
+}
+
+/// REFUSALS from `peer`, of our epoch: each change listed that we hold judged as our
+/// rules have it now; if they refuse one, we took it unjudged, and drop it.
+pub(crate) async fn received(state: &AppState, peer: &str, message: RefusalsMessage) {
+    let doc = std::sync::Arc::clone(&state.doc);
+    let changes: Vec<ChangeHash> = message.changes.into_iter().take(MAX_LISTED).collect();
+    let judged = tokio::task::spawn_blocking(move || {
+        let replica = doc.blocking_read();
+        if replica.epoch != message.epoch {
+            return None;
+        }
+        let policy = crate::access::policy::of(&replica.doc);
+        if !policy.enforced() || matches!(policy, crate::access::policy::Policy::Unknown(_)) {
+            // We can't judge as they do: receiving without judging until upgraded.
+            return None;
+        }
+        let mut refused = None;
+        let mut taken = Vec::new();
+        for hash in changes {
+            match judge_held(&replica.doc, replica.epoch, &hash) {
+                Some(Err(why)) => {
+                    refused = Some((hash, why));
+                    break;
+                }
+                Some(Ok(())) => taken.push(hash),
+                None => {}
+            }
+        }
+        Some((refused, taken))
+    })
+    .await
+    .ok()
+    .flatten();
+    let Some((refused, taken)) = judged else {
+        return;
+    };
+    if let Some((hash, why)) = refused {
+        warn!(peer = %peer, change = %hash, why = %why, "We hold a change the swarm's rules refuse, taken unjudged: dropping it and what builds on it (our writes among them go to the quarantine)");
+        state.drop_taken(hash).await;
+    } else if !taken.is_empty() {
+        warn!(peer = %peer, changes = ?taken, "A peer refuses changes our rules take: its policy or ours is off");
+    }
+}
+
+/// `hash`, which `doc` (of `epoch`) holds, judged by its rules now: why they refuse it,
+/// if they do; None if `doc` doesn't hold it.
+fn judge_held(
+    doc: &crate::document::Document,
+    epoch: Epoch,
+    hash: &ChangeHash,
+) -> Option<Result<(), String>> {
+    let change = doc.get_change_by_hash(hash)?;
+    let decoded = change.decode();
+    let ops = crate::document::inspect::Inspector::new(doc)
+        .ops(std::slice::from_ref(&decoded))
+        .pop()
+        .unwrap_or_default();
+    let paths = ops.iter().map(|op| op.path.clone()).collect();
+    let committer = doc
+        .get_author_for_actor(change.actor_id())
+        .and_then(|author| {
+            std::str::from_utf8(author.as_bytes())
+                .ok()
+                .map(str::to_owned)
+        });
+    let user = change
+        .message()
+        .and_then(crate::document::replica::user_of)
+        .map(str::to_owned);
+    let deps = change.deps().to_vec();
+    let judged = crate::access::rules::judge(
+        doc,
+        epoch,
+        &crate::access::rules::Change {
+            hash: *hash,
+            deps: &deps,
+            seen: &deps,
+            committer: committer.as_deref(),
+            user: user.as_deref(),
+            paths: &paths,
+            ops: &ops,
+        },
+    );
+    Some(match judged {
+        Ok(()) => Ok(()),
+        Err(crate::access::rules::Refusal::Rule(why)) => Err(why),
+        // Its dependencies are in the document: it is.
+        Err(crate::access::rules::Refusal::Unready) => Ok(()),
+    })
+}
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 9f928ad..7601d69 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -571,6 +571,7 @@ pub(crate) async fn handle_sync(
             bool,
             (Vec<(String, Vec<crate::document::ChangeHash>)>, bool),
             Option<std::time::Duration>,
+            Option<(u32, Vec<crate::document::ChangeHash>)>,
         );
         tokio::task::spawn_blocking(move || -> Result<Option<Applied>> {
             // Before the lock: inflating and decoding a large message takes a while. No
@@ -618,6 +619,7 @@ pub(crate) async fn handle_sync(
             let epoch_seq = replica.epoch_seq;
             let mut message = message;
             let mut refused_again = false;
+            let mut roots = Vec::new();
             replica
                 .logged_signed(|doc, known, log| {
                     let result = (|| {
@@ -630,7 +632,7 @@ pub(crate) async fn handle_sync(
                             if required {
                                 known.offer(doc, signatures.clone());
                             }
-                            refused_again = crate::access::screen::apply_bytes(
+                            let outcome = crate::access::screen::apply_bytes(
                                 doc,
                                 epoch,
                                 epoch_seq,
@@ -638,8 +640,9 @@ pub(crate) async fn handle_sync(
                                 &screening,
                                 required.then_some(&mut *known),
                                 log,
-                            )?
-                            .refused_again;
+                            )?;
+                            refused_again = outcome.refused_again;
+                            roots = outcome.roots;
                         }
                         sync.receive_logged(doc, message, log)
                     })();
@@ -649,6 +652,10 @@ pub(crate) async fn handle_sync(
             peer_state.last_sync_rx = Some(Instant::now());
             // It offers again only what we refused, and would as fast as we answer: we
             // answer later, later each time (`SyncSession::refused_again`).
+            // What it brought that broke the rules, told to it now and then
+            // (`swarm::refusals`).
+            let tell = (!roots.is_empty() && peer_state.sync.refusals_due(Instant::now()))
+                .then(|| (replica.epoch_seq, std::mem::take(&mut roots)));
             let back_off = if refused_again {
                 Some(peer_state.sync.refused_again(Instant::now()))
             } else {
@@ -744,16 +751,20 @@ pub(crate) async fn handle_sync(
                 carried,
                 authors,
                 back_off,
+                tell,
             )))
         })
         .await
         .context("Sync task panicked")??
     };
 
-    let Some((caught_up, changed, carried, (authors, unauthored), back_off)) = applied else {
+    let Some((caught_up, changed, carried, (authors, unauthored), back_off, tell)) = applied else {
         debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
         return Ok(());
     };
+    if let Some((epoch_seq, roots)) = tell {
+        crate::swarm::refusals::tell(state, peer, epoch, epoch_seq, roots).await;
+    }
     if let Some(delay) = back_off {
         debug!(
             peer = %peer,
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index a1fb9d5..e0359b0 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -254,6 +254,23 @@ pub(crate) enum ParsedMessage {
     Disconnect(NeighbourAnswer),
     /// Signatures of changes, sent on their own (`swarm::signatures`).
     Signatures(SignaturesMessage),
+    /// Changes the sender refused, of those the recipient brought (`swarm::refusals`).
+    Refusals(RefusalsMessage),
+}
+
+/// REFUSALS: changes the sender refused by the swarm's rules (`access::screen`), of
+/// those the recipient brought it: the recipient judges them itself, and rejoins if its
+/// own rules refuse them too, as it took them under another policy (`swarm::refusals`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct RefusalsMessage {
+    #[serde(with = "crate::transport::wire::codec::epoch")]
+    pub(crate) epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub(crate) epoch_seq: u32,
+    /// The changes that broke the rules (not those refused for building on them), at
+    /// most `swarm::refusals::MAX_LISTED`.
+    #[serde(with = "crate::transport::wire::codec::change_hashes")]
+    pub(crate) changes: Vec<crate::document::ChangeHash>,
 }
 
 /// SIGNATURES: signatures of changes no sync message took (`swarm::signatures`).
@@ -413,6 +430,7 @@ pub(crate) enum MessageKind {
     NeighbourReply,
     Disconnect,
     Signatures,
+    Refusals,
 }
 
 impl ParsedMessage {
@@ -436,6 +454,7 @@ impl ParsedMessage {
             ParsedMessage::NeighbourReply(_) => MessageKind::NeighbourReply,
             ParsedMessage::Disconnect(_) => MessageKind::Disconnect,
             ParsedMessage::Signatures(_) => MessageKind::Signatures,
+            ParsedMessage::Refusals(_) => MessageKind::Refusals,
         }
     }
 
@@ -462,6 +481,7 @@ impl ParsedMessage {
                 Some((answer.epoch, answer.epoch_seq))
             }
             ParsedMessage::Signatures(message) => Some((message.epoch, message.epoch_seq)),
+            ParsedMessage::Refusals(message) => Some((message.epoch, message.epoch_seq)),
             ParsedMessage::Who(_)
             | ParsedMessage::Me(_)
             | ParsedMessage::Ping(_)
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 180d3b5..e810595 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -59,6 +59,8 @@ pub(crate) struct ByKind {
     pub(crate) hyparview: Counter,
     /// SIGNATURES (`swarm::signatures`).
     pub(crate) signatures: Counter,
+    /// REFUSALS (`swarm::refusals`).
+    pub(crate) refusals: Counter,
 }
 
 impl ByKind {
@@ -76,6 +78,7 @@ impl ByKind {
             MessageKind::Prune,
             MessageKind::Neighbour,
             MessageKind::Signatures,
+            MessageKind::Refusals,
         ] {
             let theirs = *other.get(kind);
             self.get_mut(kind).absorb(theirs);
@@ -98,6 +101,7 @@ impl ByKind {
                 &self.hyparview
             }
             MessageKind::Signatures => &self.signatures,
+            MessageKind::Refusals => &self.refusals,
         }
     }
 
@@ -117,6 +121,7 @@ impl ByKind {
                 &mut self.hyparview
             }
             MessageKind::Signatures => &mut self.signatures,
+            MessageKind::Refusals => &mut self.refusals,
         }
     }
 }
@@ -327,6 +332,7 @@ fn names(kind: MessageKind) -> (&'static str, &'static str) {
         MessageKind::NeighbourReply => ("NEIGHBOUR_REPLY", "hyparview"),
         MessageKind::Disconnect => ("DISCONNECT", "hyparview"),
         MessageKind::Signatures => ("SIGNATURES", "signatures"),
+        MessageKind::Refusals => ("REFUSALS", "refusals"),
     }
 }
 
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 88b2916..7ff7835 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 41;
+pub(crate) const VERSION: u8 = 42;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -37,6 +37,7 @@ fn tag(kind: MessageKind) -> u8 {
         MessageKind::NeighbourReply => 17,
         MessageKind::Disconnect => 18,
         MessageKind::Signatures => 19,
+        MessageKind::Refusals => 20,
     }
 }
 
@@ -99,6 +100,7 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
             rmp_serde::encode::write(&mut payload, answer)?
         }
         ParsedMessage::Signatures(message) => rmp_serde::encode::write(&mut payload, message)?,
+        ParsedMessage::Refusals(message) => rmp_serde::encode::write(&mut payload, message)?,
     }
     Ok(payload)
 }
@@ -180,6 +182,8 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
         ParsedMessage::Disconnect(rmp_serde::from_slice(body)?)
     } else if is(MessageKind::Signatures) {
         ParsedMessage::Signatures(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Refusals) {
+        ParsedMessage::Refusals(rmp_serde::from_slice(body)?)
     } else {
         return Err(WireError::UnknownKind(kind));
     })
@@ -544,6 +548,11 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
             epoch_seq: 1,
             signatures: serde_bytes::ByteBuf::from(vec![9; 96]),
         }),
+        ParsedMessage::Refusals(crate::transport::RefusalsMessage {
+            epoch,
+            epoch_seq: 1,
+            changes: vec![crate::document::ChangeHash([7; 32])],
+        }),
     ]
 }
 
@@ -575,7 +584,7 @@ mod tests {
     #[test]
     fn every_kind_round_trips() {
         let samples = samples();
-        assert_eq!(samples.len(), 18, "a kind is missing from `samples`");
+        assert_eq!(samples.len(), 19, "a kind is missing from `samples`");
         for message in samples {
             let bytes = encode(&message).unwrap();
             let decoded = decode(&bytes).unwrap();