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, 41 insertions, 18 deletions

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