refusals: a drop taking much waits for a second member, and shows

A node told (REFUSALS) that it holds a change the swarm's rules refuse,
and whose own rules refuse it too, dropped it and everything built on
it on one peer's word, however much that was; its own writes among them
went to the quarantine, the others' nowhere.

* A drop taking more than 100 changes, or more than ten and a tenth of
  the document, now
  waits until a second member names the same change: the node's rules
  and one peer's word could both be off. Logged as an error, counted in
  /admin/stats (sync.dropsHeld).
* The others' changes dropped are listed with the refused (GET
  /admin/refused, "taken unjudged, dropped"), for operators to see.

The cheap half of asking the others (future-work.md, Telling a
malicious change from a wrong policy).

1 files changed, 58 insertions, 4 deletions

diff --git a/src/state/host.rs b/src/state/host.rs
index 38be831..feda8a9 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -95,12 +95,21 @@ impl AppState {
     /// (`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`; nor if `root` is an epoch's root, or nothing
-    /// would be kept, which no rule can mean (logged as an error).
-    pub(crate) async fn drop_taken(&self, root: ChangeHash) {
+    /// them go to the quarantine, the others' to the refused (`GET /admin/refused`, saying
+    /// so). Every peer syncs anew. Nothing while a rejoin runs, or if the document doesn't
+    /// hold `root`; nor if `root` is an epoch's root, or nothing would be kept, which no
+    /// rule can mean (logged as an error). A drop taking much (`BIG_DROP`) waits for a
+    /// second member to say so (`peer` the first, `why` its rules' reason): its rules and
+    /// one peer's word might both be off, the swarm's not.
+    pub(crate) async fn drop_taken(&self, root: ChangeHash, peer: &str, why: &str) {
+        /// Changes a drop takes, at most, on one peer's word: more, or more than a tenth
+        /// of the document (past `SMALL_DROP`), wait for another's.
+        const BIG_DROP: usize = 100;
+        const SMALL_DROP: usize = 10;
         let own = self.remake.own_actors();
         let state = self.clone();
+        let peer = peer.to_owned();
+        let why = why.to_owned();
         let dropped = tokio::task::spawn_blocking(move || {
             let mut replica = state.doc.blocking_write();
             if !replica.writable {
@@ -127,6 +136,19 @@ impl AppState {
                 error!(change = %root, "Dropping a change would leave nothing; kept it");
                 return None;
             }
+            if tainted.len() > BIG_DROP
+                || (tainted.len() > SMALL_DROP && tainted.len() * 10 > changes.len())
+            {
+                use crate::locks::MutexExt;
+                let mut screening = state.screening.lock_or_recover();
+                let said = screening
+                    .of(replica.epoch_seq)
+                    .hold_drop(root, &peer);
+                if said < 2 {
+                    error!(change = %root, peer = %peer, changes = tainted.len(), why = %why, "A peer says we hold a change the swarm's rules refuse, and ours do too, but dropping it takes much: waiting for another member to say so");
+                    return None;
+                }
+            }
             let built_on: std::collections::HashSet<ChangeHash> = kept
                 .iter()
                 .flat_map(|change| change.deps().iter().copied())
@@ -162,9 +184,41 @@ impl AppState {
                     return None;
                 }
             };
+            // The others' changes dropped, for operators (own: to the quarantine).
+            let theirs: Vec<crate::access::screen::Refused> = changes
+                .iter()
+                .filter(|change| tainted.contains(&change.hash()))
+                .filter(|change| !actors.contains(change.actor_id()))
+                .map(|change| crate::access::screen::Refused {
+                    change: (*change).clone(),
+                    committer: replica
+                        .doc
+                        .get_author_for_actor(change.actor_id())
+                        .and_then(|author| {
+                            std::str::from_utf8(author.as_bytes())
+                                .ok()
+                                .map(str::to_owned)
+                        }),
+                    user: change
+                        .message()
+                        .and_then(crate::document::replica::user_of)
+                        .map(str::to_owned),
+                    paths: Default::default(),
+                    why: if change.hash() == root {
+                        format!("taken unjudged, dropped: {why}")
+                    } else {
+                        format!("taken unjudged, dropped: it builds on {root}")
+                    },
+                })
+                .collect();
             state.remake.add_own_actor(actor);
             let epoch = replica.epoch;
             let old = state.swap_document(&mut replica, epoch, fresh);
+            {
+                use crate::locks::MutexExt;
+                let mut screening = state.screening.lock_or_recover();
+                screening.of(replica.epoch_seq).keep_dropped(theirs);
+            }
             Some((epoch, old, paths, tainted.len()))
         })
         .await