state: refused writes limited for the node too

Review 3, R5: --refused-writes-per-minute bounds each user, but the
cost it guards, a document copy under the write lock, is the node's:
ten users, or ten accounts of one, are ten times the budget.

The node now takes six times a user's limit within a minute in all
(refused::NODE_FACTOR); past it, every user with a write refused
within the minute is answered 429 untried, until the oldest leave it.
A user none of whose writes was refused writes on, so honest writers
aren't held up by others' refusals.

2 files changed, 53 insertions, 17 deletions

diff --git a/src/config/mod.rs b/src/config/mod.rs
index 34ed709..4230b65 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -184,7 +184,7 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         Kind::Integer,
         "COUNT",
         Some("10"),
-        "Writes a user may have refused within a minute because every node would refuse them (docs/access.md); past it, its writes are answered 429 untried for a while. Each refusal costs a copy of the document.",
+        "Writes a user may have refused within a minute because every node would refuse them (docs/access.md); past it, its writes are answered 429 untried for a while, and past six times as many on the node in all, those of every user with one refused. Each refusal costs a copy of the document.",
     ),
     option(
         "genesis-root",
diff --git a/src/state/refused.rs b/src/state/refused.rs
index bb37af0..b01a84a 100644
--- a/src/state/refused.rs
+++ b/src/state/refused.rs
@@ -1,42 +1,62 @@
 //! Writes every node would refuse, counted by user (`AppState::mutate_doc_as`): each is
 //! rolled back by forking the document at the heads before it, which costs as much as
 //! the document's history, under the write lock. A user may have so many refused within
-//! a minute (`--refused-writes-per-minute`); past it, its writes are answered `429`
-//! untried until the oldest leave the minute.
+//! a minute (`--refused-writes-per-minute`), and the node `NODE_FACTOR` times as many in
+//! all; past either, the user's writes (past the node's, those of every user with one
+//! refused within the minute) are answered `429` untried until the oldest leave it. A
+//! user none of whose writes was refused within the minute writes on.
 
 use std::collections::{HashMap, VecDeque};
 use std::time::{Duration, Instant};
 
-/// How long a refused write counts against its user.
+/// How long a refused write counts against its user, and the node.
 const WINDOW: Duration = Duration::from_secs(60);
 
-/// Each user's writes refused within the last `WINDOW`, oldest first.
+/// How many refused writes the node takes within a minute, in all, as many times the
+/// limit of one user: ten users can't hold the write lock ten times as long.
+pub(crate) const NODE_FACTOR: usize = 6;
+
+/// The writes refused within the last `WINDOW`, oldest first: each user's, and all.
 #[derive(Debug, Default)]
-pub(crate) struct RefusedWrites(HashMap<String, VecDeque<Instant>>);
+pub(crate) struct RefusedWrites {
+    by_user: HashMap<String, VecDeque<Instant>>,
+    all: VecDeque<Instant>,
+}
+
+/// `refused` without those that left the minute before `now`.
+fn prune(refused: &mut VecDeque<Instant>, now: Instant) {
+    while refused
+        .front()
+        .is_some_and(|at| now.duration_since(*at) >= WINDOW)
+    {
+        refused.pop_front();
+    }
+}
 
 impl RefusedWrites {
     /// Whether `user` may try another write at `now`: fewer than `limit` of its writes
-    /// refused within the last minute.
+    /// refused within the last minute; and, if any was, fewer than `NODE_FACTOR` times
+    /// `limit` of all users'.
     pub(crate) fn allows(&mut self, user: &str, limit: usize, now: Instant) -> bool {
-        let Some(refused) = self.0.get_mut(user) else {
+        prune(&mut self.all, now);
+        let Some(refused) = self.by_user.get_mut(user) else {
             return true;
         };
-        while refused
-            .front()
-            .is_some_and(|at| now.duration_since(*at) >= WINDOW)
-        {
-            refused.pop_front();
-        }
+        prune(refused, now);
         if refused.is_empty() {
-            self.0.remove(user);
+            self.by_user.remove(user);
             return true;
         }
-        refused.len() < limit
+        refused.len() < limit && self.all.len() < limit.saturating_mul(NODE_FACTOR)
     }
 
     /// One of `user`'s writes refused at `now`.
     pub(crate) fn note(&mut self, user: &str, now: Instant) {
-        self.0.entry(user.to_owned()).or_default().push_back(now);
+        self.by_user
+            .entry(user.to_owned())
+            .or_default()
+            .push_back(now);
+        self.all.push_back(now);
     }
 }
 
@@ -56,4 +76,20 @@ mod tests {
         assert!(refused.allows("bob", 2, start));
         assert!(refused.allows("alice", 2, start + WINDOW));
     }
+
+    /// Past the node's limit, every user with a write refused within the minute waits;
+    /// one without writes on.
+    #[test]
+    fn refused_writes_per_node() {
+        let mut refused = RefusedWrites::default();
+        let start = Instant::now();
+        // Two each, below a user's limit of three; the node's taken in all.
+        for user in 0..3 * NODE_FACTOR / 2 {
+            refused.note(&format!("u{user}"), start);
+            refused.note(&format!("u{user}"), start);
+        }
+        assert!(!refused.allows("u0", 3, start));
+        assert!(refused.allows("carol", 3, start));
+        assert!(refused.allows("u0", 3, start + WINDOW));
+    }
 }