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.

8 files changed, 105 insertions, 0 deletions

diff --git a/src/api/mod.rs b/src/api/mod.rs
index 43d3c20..0dde808 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -341,6 +341,7 @@ pub(crate) fn store_status(error: StoreError, context: &str) -> ApiError {
         StoreError::Rejoining => StatusCode::SERVICE_UNAVAILABLE,
         StoreError::UnknownPolicy(_) => StatusCode::CONFLICT,
         StoreError::Refused(_) => StatusCode::FORBIDDEN,
+        StoreError::TooManyRefused => StatusCode::TOO_MANY_REQUESTS,
         other => {
             error!(error = %format!("{other:#}"), "{context}");
             return StatusCode::INTERNAL_SERVER_ERROR.into();
diff --git a/src/config/mod.rs b/src/config/mod.rs
index f6d3e15..572960b 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -180,6 +180,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         "With bootstrap-users: their SSH keys, as authorized_keys lines (ssh-ed25519 <base64> [comment], named by the comment), recorded in their entries: to log in by key and sign values (docs/access.md), so a swarm can be strict from its start.",
     ),
     option(
+        "refused-writes-per-minute",
+        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.",
+    ),
+    option(
         "genesis-root",
         Kind::Text,
         "HASH",
@@ -633,6 +640,8 @@ pub(crate) struct Settings {
     pub(crate) challenge_lifetime: Duration,
     #[serde(default = "default_key_logins_per_user")]
     pub(crate) key_logins_per_user: usize,
+    #[serde(default = "default_refused_writes_per_minute")]
+    pub(crate) refused_writes_per_minute: usize,
     #[serde(default)]
     pub(crate) debug_api: bool,
     #[serde(default = "default_log_format")]
@@ -799,6 +808,9 @@ fn default_session_lifetime() -> Duration {
 fn default_challenge_lifetime() -> Duration {
     Duration::from_secs(5 * 60)
 }
+fn default_refused_writes_per_minute() -> usize {
+    10
+}
 fn default_key_logins_per_user() -> usize {
     100
 }
diff --git a/src/document/store.rs b/src/document/store.rs
index e04ef41..fb28a4b 100644
--- a/src/document/store.rs
+++ b/src/document/store.rs
@@ -185,6 +185,9 @@ pub(crate) enum StoreError {
     /// the write is rolled back.
     #[error("every node would refuse this write: {0}")]
     Refused(String),
+    /// The user's writes were refused too often of late (`state::refused`): not tried.
+    #[error("too many of your writes refused of late: try again in a while")]
+    TooManyRefused,
     #[error(transparent)]
     Doc(#[from] automerge::AutomergeError),
     #[error(transparent)]
diff --git a/src/main.rs b/src/main.rs
index 8c12f41..880d793 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -267,6 +267,7 @@ async fn main() -> Result<()> {
             .filter_map(|line| access::sshsig::PublicKey::parse(line).ok())
             .collect(),
         key_logins_per_user: args.key_logins_per_user,
+        refused_writes_per_minute: args.refused_writes_per_minute,
         cors_origins: args.cors_origins.clone(),
         files,
         state_file,
diff --git a/src/node.rs b/src/node.rs
index 84a5c01..c74d7c0 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -78,6 +78,8 @@ pub(crate) struct NodeConfig {
     /// `--challenge-lifetime`, `--key-logins-per-user`.
     pub(crate) challenge_lifetime: Duration,
     pub(crate) key_logins_per_user: usize,
+    /// `--refused-writes-per-minute`.
+    pub(crate) refused_writes_per_minute: usize,
     /// `--genesis-root`.
     pub(crate) genesis_root: Option<crate::document::ChangeHash>,
     /// `--strict-keys`.
@@ -182,6 +184,8 @@ impl Node {
             session_lifetime: config.session_lifetime,
             challenge_lifetime: config.challenge_lifetime,
             key_logins_per_user: config.key_logins_per_user,
+            refused_writes: Default::default(),
+            refused_writes_per_minute: config.refused_writes_per_minute,
             strict_keys: Arc::new(config.strict_keys),
             cors_origins: Arc::new(config.cors_origins),
             state_file: config.state_file,
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index 253ec54..8d6437a 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -222,6 +222,7 @@ impl Sim {
             session_lifetime: Duration::from_secs(86400),
             challenge_lifetime: Duration::from_secs(300),
             key_logins_per_user: 100,
+            refused_writes_per_minute: 10,
             genesis_root: None,
             strict_keys: Vec::new(),
             cors_origins: Vec::new(),
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));
+    }
+}