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, 15 insertions, 4 deletions

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,