2 files changed, 83 insertions, 0 deletions

diff --git a/src/state/mod.rs b/src/state/mod.rs
index d3f5859..e079340 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -20,6 +20,7 @@
 mod host;
 mod peer;
 mod read;
+pub(crate) mod refused;
 mod timing;
 
 pub(crate) use peer::{HelloState, Incidents, Keepalive, Liveness, PeerState, Standing};
@@ -161,6 +162,10 @@ pub(crate) struct AppState {
     /// may make within it (`access::challenge`).
     pub(crate) challenge_lifetime: Duration,
     pub(crate) key_logins_per_user: usize,
+    /// Writes every node would refuse, by user, and how many a user may have within a
+    /// minute (`refused`).
+    pub(crate) refused_writes: Arc<std::sync::Mutex<refused::RefusedWrites>>,
+    pub(crate) refused_writes_per_minute: usize,
     /// Its epoch's root, as peers name it and as it knows its own (`swarm::roots`), and
     /// the one it was given (`--genesis-root`).
     pub(crate) roots: Arc<std::sync::Mutex<crate::swarm::roots::Roots>>,
@@ -636,6 +641,18 @@ impl AppState {
                     node: state.local_id(),
                     user: user.as_deref(),
                 };
+                // A user whose writes every node refused, again and again: answered untried
+                // for a while, as each refusal costs a fork of the document (`refused`).
+                if let Some(user) = &user {
+                    use crate::locks::MutexExt;
+                    if !state.refused_writes.lock_or_recover().allows(
+                        user,
+                        state.refused_writes_per_minute,
+                        std::time::Instant::now(),
+                    ) {
+                        return Err(E::from(StoreError::TooManyRefused));
+                    }
+                }
                 let before = replica.doc.get_heads();
                 let result = replica.transact(state.clock.now_millis(), &writer, f);
                 let result = match result {
@@ -648,6 +665,13 @@ impl AppState {
                         }
                         Err(why) => {
                             roll_back(&mut replica, &before)?;
+                            if let Some(user) = &user {
+                                use crate::locks::MutexExt;
+                                state
+                                    .refused_writes
+                                    .lock_or_recover()
+                                    .note(user, std::time::Instant::now());
+                            }
                             tracing::debug!(reason = %why, "Refused a write every node would refuse");
                             Err(E::from(StoreError::Refused(why)))
                         }
diff --git a/src/state/refused.rs b/src/state/refused.rs
new file mode 100644
index 0000000..bb37af0
--- /dev/null
+++ b/src/state/refused.rs
@@ -0,0 +1,59 @@
+//! 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.
+
+use std::collections::{HashMap, VecDeque};
+use std::time::{Duration, Instant};
+
+/// How long a refused write counts against its user.
+const WINDOW: Duration = Duration::from_secs(60);
+
+/// Each user's writes refused within the last `WINDOW`, oldest first.
+#[derive(Debug, Default)]
+pub(crate) struct RefusedWrites(HashMap<String, VecDeque<Instant>>);
+
+impl RefusedWrites {
+    /// Whether `user` may try another write at `now`: fewer than `limit` of its writes
+    /// refused within the last minute.
+    pub(crate) fn allows(&mut self, user: &str, limit: usize, now: Instant) -> bool {
+        let Some(refused) = self.0.get_mut(user) else {
+            return true;
+        };
+        while refused
+            .front()
+            .is_some_and(|at| now.duration_since(*at) >= WINDOW)
+        {
+            refused.pop_front();
+        }
+        if refused.is_empty() {
+            self.0.remove(user);
+            return true;
+        }
+        refused.len() < limit
+    }
+
+    /// 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);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Past the limit within a minute, a user waits; others don't; a minute on, it may.
+    #[test]
+    fn refused_writes_per_user() {
+        let mut refused = RefusedWrites::default();
+        let start = Instant::now();
+        assert!(refused.allows("alice", 2, start));
+        refused.note("alice", start);
+        refused.note("alice", start);
+        assert!(!refused.allows("alice", 2, start));
+        assert!(refused.allows("bob", 2, start));
+        assert!(refused.allows("alice", 2, start + WINDOW));
+    }
+}