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

5 files changed, 102 insertions, 6 deletions

diff --git a/src/access/screen.rs b/src/access/screen.rs
index e15b407..89ca007 100644
--- a/src/access/screen.rs
+++ b/src/access/screen.rs
@@ -70,6 +70,9 @@ pub(crate) struct Screening {
     refusals: Refusals,
     /// The members refused as a whole, by node ID (`refusing`).
     refusing: std::collections::BTreeMap<String, Refusing>,
+    /// Changes this node holds that peers say the swarm's rules refuse, and its own do
+    /// too, whose drop would take much: each with the peers that said so (`hold_drop`).
+    held_drops: HashMap<ChangeHash, BTreeSet<String>>,
 }
 
 /// A member refused as a whole: since when, for which change of its, and why.
@@ -139,6 +142,30 @@ impl Screening {
         }
     }
 
+    /// `peer` says `root`, held here, is refused by the swarm's rules, its drop taking
+    /// much (`AppState::drop_taken`): how many peers said so of it, this one with them.
+    pub(crate) fn hold_drop(&mut self, root: ChangeHash, peer: &str) -> usize {
+        let named = self.held_drops.entry(root).or_default();
+        named.insert(peer.to_owned());
+        named.len()
+    }
+
+    /// Drops held, for another peer to say so too (`hold_drop`).
+    pub(crate) fn drops_held(&self) -> usize {
+        self.held_drops.len()
+    }
+
+    /// `dropped`, changes this node took unjudged and dropped (`AppState::drop_taken`),
+    /// kept with those refused, the oldest going past `MAX_REFUSED`; logged once, there.
+    pub(crate) fn keep_dropped(&mut self, dropped: Vec<Refused>) {
+        for dropped in dropped {
+            self.refused.push_back(dropped);
+            if self.refused.len() > MAX_REFUSED {
+                self.refused.pop_front();
+            }
+        }
+    }
+
     /// Those refused, oldest first.
     pub(crate) fn refused(&self) -> impl DoubleEndedIterator<Item = &Refused> {
         self.refused.iter()
diff --git a/src/api/stats.rs b/src/api/stats.rs
index 1f0701d..90b4fc2 100644
--- a/src/api/stats.rs
+++ b/src/api/stats.rs
@@ -130,6 +130,10 @@ struct SyncStats {
     /// Peers refused as a whole, their own changes having broken the swarm's rules
     /// (`access::screen`): sent nothing.
     refused_peers: usize,
+    /// Changes peers say the swarm's rules refuse, which this node took unjudged and its
+    /// rules refuse too, whose drop takes much: held for a second member to say so
+    /// (`AppState::drop_taken`).
+    drops_held: usize,
     /// The largest sync message received since start (its Automerge part). Larger
     /// transfers come as deltas (blobs).
     largest_received_bytes: u64,
@@ -378,6 +382,10 @@ pub(crate) async fn collect(state: &AppState, list: &PeerListResponse) -> StatsR
             last_from_neighbour_secs,
             diverged_peers,
             refused_peers,
+            drops_held: {
+                use crate::locks::MutexExt;
+                state.screening.lock_or_recover().drops_held()
+            },
             largest_received_bytes: node.largest_sync_received,
         },
         relay: RelayStats {
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 41d224a..dd49751 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -640,6 +640,12 @@ pub(super) fn render(stats: &StatsResponse, list: &PeerListResponse) -> String {
         "Peers refused as a whole, their own changes having broken the swarm's rules.",
         stats.sync.refused_peers as f64,
     );
+    out.single(
+        "drops_held",
+        "gauge",
+        "Changes peers say the swarm's rules refuse, ours too, whose drop takes much: held for a second member to say so.",
+        stats.sync.drops_held as f64,
+    );
     let relay = &stats.relay;
     out.single(
         "announce_rounds_total",
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
diff --git a/src/swarm/refusals.rs b/src/swarm/refusals.rs
index c1949eb..af4252c 100644
--- a/src/swarm/refusals.rs
+++ b/src/swarm/refusals.rs
@@ -10,7 +10,8 @@
 //! 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
+//! among them quarantined, the others' listed as refused (`AppState::drop_taken`); one
+//! taking much, once a second member says so too. 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).
 
@@ -85,7 +86,7 @@ pub(crate) async fn received(state: &AppState, peer: &str, message: RefusalsMess
     };
     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;
+        state.drop_taken(hash, peer, &why).await;
     } else if !taken.is_empty() {
         warn!(peer = %peer, changes = ?taken, "A peer refuses changes our rules take: its policy or ours is off");
     }