state: limit refused writes per user

A write every node would refuse is rolled back by forking the document
at the heads before it, under the write lock, at a cost that grows with
the document's history. A user could ask for such writes in a loop.

Count each user's refused writes over the last minute. Past
--refused-writes-per-minute (default 10), its writes are answered 429
without being tried, until the oldest refusal leaves the minute.

1 files changed, 24 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)))
                         }