access: members refused as a whole

With signatures required, a member whose own change broke a rule is
refused as a whole: sent nothing, shown in /peers (refused), in stats
(sync.refusedPeers) and on the dashboard, until a change of its is
taken again or the epoch changes.

Auth plan M6 (diverged standing).

1 files changed, 80 insertions, 0 deletions

diff --git a/src/access/screen.rs b/src/access/screen.rs
index a5a8c8a..6773d05 100644
--- a/src/access/screen.rs
+++ b/src/access/screen.rs
@@ -8,6 +8,12 @@
 //! the policy, bans) is what its writer saw, unless one of those ancestors changed it. So a change that
 //! does (`relevant`) is applied before the next is judged: the rest go in at once.
 //!
+//! A member whose own change broke a rule, as its signature proves (signatures
+//! required), is refused as a whole (`Screening::refusing`): the node sends it nothing
+//! more, as all it commits builds on what was refused, until a change of its is taken
+//! again (it rejoined) or the epoch changes. One that only passed a refused change on
+//! isn't: the committer is, whoever brought it.
+//!
 //! A change refused is remembered by its hash (`Refusals`): when it comes again, as its
 //! writer keeps offering what the node never takes, it is refused again without being
 //! decoded or logged, and the sync session with whoever sent only such changes backs off
@@ -58,6 +64,16 @@ pub(crate) struct Screening {
     pub(crate) waiting: Waiting,
     refused: std::collections::VecDeque<Refused>,
     refusals: Refusals,
+    /// The members refused as a whole, by node ID (`refusing`).
+    refusing: std::collections::BTreeMap<String, Refusing>,
+}
+
+/// A member refused as a whole: since when, for which change of its, and why.
+#[derive(Debug, Clone)]
+pub(crate) struct Refusing {
+    pub(crate) since: tokio::time::Instant,
+    pub(crate) change: ChangeHash,
+    pub(crate) why: String,
 }
 
 /// The hashes of the changes refused, to refuse them again by hash.
@@ -117,6 +133,33 @@ impl Screening {
     pub(crate) fn refused(&self) -> impl DoubleEndedIterator<Item = &Refused> {
         self.refused.iter()
     }
+
+    /// Whether `node` is refused as a whole, and since when, for what.
+    pub(crate) fn refusing(&self, node: &str) -> Option<&Refusing> {
+        self.refusing.get(node)
+    }
+
+    /// The members refused as a whole and taken again of `screened`, logged.
+    fn judge_members(&mut self, screened: &Screened) {
+        for node in &screened.taken_from {
+            if self.refusing.remove(node).is_some() {
+                tracing::info!(node = %node, "A member refused as a whole wrote a change taken: refused no longer");
+            }
+        }
+        for (node, change, why) in &screened.broke_rules {
+            if !self.refusing.contains_key(node) {
+                tracing::warn!(node = %node, change = %change, why = %why, "A member's own change broke the rules: refusing it as a whole, sending it nothing");
+                self.refusing.insert(
+                    node.clone(),
+                    Refusing {
+                        since: tokio::time::Instant::now(),
+                        change: *change,
+                        why: why.clone(),
+                    },
+                );
+            }
+        }
+    }
 }
 
 /// Received changes waiting for their dependencies.
@@ -136,6 +179,11 @@ pub(crate) struct Screened {
     pub(crate) refused: Vec<Refused>,
     /// Received changes refused before (`Refusals`), refused again by their hash.
     pub(crate) refused_again: usize,
+    /// Members whose own changes broke a rule, a signature proving it: with the change,
+    /// and why.
+    pub(crate) broke_rules: Vec<(String, ChangeHash, String)>,
+    /// The committers of changes taken.
+    pub(crate) taken_from: BTreeSet<String>,
 }
 
 /// Judges `received` (and those `waiting` before) and applies to `doc` (of `epoch`)
@@ -297,6 +345,9 @@ pub(crate) fn apply(
         match judged {
             Ok(()) => {
                 accepted.push(change.clone());
+                if let Some(committer) = committer {
+                    screened.taken_from.insert(committer.to_owned());
+                }
                 if relevant(&paths[index]) {
                     // What rules read changed: in before the next is judged.
                     screened.applied += accepted.len();
@@ -313,6 +364,14 @@ pub(crate) fn apply(
             }
             Err(Refusal::Rule(why)) => {
                 refused.insert(hash);
+                // Its committer's own, as a signature by it vouches.
+                if signatures.is_some()
+                    && let Some(committer) = committer
+                {
+                    screened
+                        .broke_rules
+                        .push((committer.to_owned(), hash, why.clone()));
+                }
                 screened.refused.push(refuse(why));
             }
         }
@@ -381,6 +440,7 @@ pub(crate) fn apply_bytes(
         applied: screened.applied,
         refused_again: received > 0 && screened.refused_again == received,
     };
+    screening.judge_members(&screened);
     screening.record(screened.refused);
     Ok(outcome)
 }
@@ -537,6 +597,26 @@ mod tests {
         assert!(waiting.len() == 0);
     }
 
+    /// A member whose own change broke a rule: refused as a whole, until a change of its
+    /// is taken.
+    #[test]
+    fn members_refused_as_a_whole() {
+        let mut screening = Screening::default();
+        let broke = Screened {
+            broke_rules: vec![("b".to_owned(), ChangeHash([1; 32]), "why".to_owned())],
+            ..Screened::default()
+        };
+        screening.judge_members(&broke);
+        assert_eq!(screening.refusing("b").unwrap().change, ChangeHash([1; 32]));
+        assert!(screening.refusing("a").is_none());
+        let taken = Screened {
+            taken_from: ["b".to_owned()].into(),
+            ..Screened::default()
+        };
+        screening.judge_members(&taken);
+        assert!(screening.refusing("b").is_none());
+    }
+
     /// A refused change offered again: refused by its hash, not judged or kept again,
     /// and so is a new one building on it.
     #[test]