feat: add the registration mode and the global hourly budgets to state

Mode is persisted rather than read from config alone, so an admin can flip
registration by mail instead of needing a redeploy; config supplies only the
initial value. It defaults to Silent, which is today's behaviour byte for
byte, so landing this cannot start mailing strangers on its own.

Limits holds the two daemon-wide hourly counters. Identity keys are free to
mint, so a per-key budget does not bind an attacker and these are what
actually do; each vector is pruned to the hour on read and so is bounded by
its own cap.

Also fixes reply_allowed's ledger growth, which stops being a wart here and
becomes a vulnerability: until now only accepted authors reached it, but once
a stranger can draw a reply any throwaway keypair appends a permanent row.
Ledgers idle for an hour are dropped and the count is capped, with a
regression test that drives it past the cap.

drained carries the one-time backlog guard: a live mail.db holds every
requests-tier mail ever received, and the first poll after upgrading must not
mail all of them.

daemon/src/state.rs

@@ -12,6 +12,14 @@ use serde::{Deserialize, Serialize};
 pub const REPLIES_PER_HOUR: usize = 10;
 /// Rejected-payload keys remembered for reply dedup, a bounded ring.
 pub const DEDUPE_MAX: usize = 256;
+/// At most this many per-sender ledgers are kept. Before self-service
+/// registration only accepted authors reached `reply_allowed`, so this vector
+/// grew with real authors; once a stranger can draw a reply, any throwaway
+/// keypair appends a row, which makes an unbounded ledger a remote
+/// disk-growth primitive against state.json.
+pub const REPLY_LEDGERS_MAX: usize = 1024;
+/// The window every rate limit here measures over.
+const HOUR: i64 = 3600;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct FeedState {
@@ -56,6 +64,44 @@ pub enum PruneMode {
     Fifo,
 }
 
+/// Whether strangers may register themselves. Persisted so an admin can flip
+/// it by mail; `Config::registration` only supplies the initial value, because
+/// a config-only switch would need a redeploy to change.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum Mode {
+    /// Today's behaviour: an unknown sender is dropped without a reply. The
+    /// default, so deploying this cannot start mailing strangers on its own.
+    #[default]
+    Silent,
+    /// Unknown senders may register themselves.
+    Open,
+    /// Unknown senders are told registration is closed.
+    Closed,
+}
+
+/// Which global budget a request draws on.
+// Wired up by the registration handler; until then only the tests use it.
+#[allow(dead_code)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Global {
+    /// Info replies to unknown senders.
+    Informed,
+    /// Registration attempts that reach the network. Attempts, not successes:
+    /// a registration that fails its key check costs the same outbound dial.
+    Attempted,
+}
+
+/// Daemon-wide hourly counters. Identity keys are free to mint, so a per-key
+/// budget does not bind an attacker — these do.
+#[derive(Debug, Default, Serialize, Deserialize)]
+struct Limits {
+    #[serde(default)]
+    informed: Vec<i64>,
+    #[serde(default)]
+    attempted: Vec<i64>,
+}
+
 #[derive(Debug, Default, Serialize, Deserialize)]
 pub struct State {
     pub feeds: Vec<FeedState>,
@@ -63,6 +109,16 @@ pub struct State {
     replies: Vec<ReplyLedger>,
     #[serde(default)]
     dedupe: Vec<String>,
+    #[serde(default)]
+    limits: Limits,
+    /// The one-time requests-tier backlog drain has run. A live mail.db holds
+    /// every requests-tier mail ever received, so without this the first poll
+    /// after upgrading would mail every stranger in that history.
+    #[serde(default)]
+    pub drained: bool,
+    /// An admin's `Twt-Registration` override; `None` means follow the config.
+    #[serde(default)]
+    registration: Option<Mode>,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -102,6 +158,19 @@ impl State {
         smolboid_core::render_timestamp(now)[..10].to_owned()
     }
 
+    /// The effective registration mode: an admin's override if one was set,
+    /// otherwise the configured default.
+    // Read by the registration handler; until then only the tests call these.
+    #[allow(dead_code)]
+    pub fn registration(&self, configured: Mode) -> Mode {
+        self.registration.unwrap_or(configured)
+    }
+
+    #[allow(dead_code)]
+    pub fn set_registration(&mut self, mode: Mode) {
+        self.registration = Some(mode);
+    }
+
     /// Rolls the daily counters when the day changed.
     pub fn roll_counters(feed: &mut FeedState, now: i64) {
         let today = State::day(now);
@@ -120,9 +189,39 @@ pub fn payload_key(sender: &[u8; 32], time: i64, body: &[u8]) -> String {
     b32(&sha256(&[sender, &time.to_be_bytes(), body]))
 }
 
+/// Whether a daemon-wide budget still has room, recording the draw if so.
+/// Each vector is pruned to the hour on read, so it is bounded by its own cap
+/// rather than by how often it is called.
+#[allow(dead_code)]
+pub fn global_allowed(state: &mut State, which: Global, cap: u32, now: i64) -> bool {
+    let times = match which {
+        Global::Informed => &mut state.limits.informed,
+        Global::Attempted => &mut state.limits.attempted,
+    };
+    times.retain(|t| now - t < HOUR);
+    if times.len() >= cap as usize {
+        return false;
+    }
+    times.push(now);
+    true
+}
+
 /// Whether an error reply may still be sent: within the per-sender hourly
 /// budget and never for a payload already answered. Records the decision.
 pub fn reply_allowed(state: &mut State, sender: &[u8; 32], payload: &str, now: i64) -> bool {
+    // Drop ledgers with nothing left inside the window, then cap how many are
+    // kept at all: see REPLY_LEDGERS_MAX. Pruning first means a steady stream
+    // of distinct senders evicts itself instead of accumulating.
+    state.replies.retain_mut(|l| {
+        l.times.retain(|t| now - t < HOUR);
+        !l.times.is_empty()
+    });
+    // Drain to one below the cap, so the ledger this call may add still
+    // leaves the total at or under it.
+    if state.replies.len() >= REPLY_LEDGERS_MAX {
+        let excess = state.replies.len() - REPLY_LEDGERS_MAX + 1;
+        state.replies.drain(..excess);
+    }
     if state.dedupe.iter().any(|seen| seen == payload) {
         return false;
     }
@@ -221,6 +320,67 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn global_budgets_prune_at_the_hour_boundary() {
+        let mut state = State::default();
+        let t0 = 1_760_000_000;
+        for i in 0..3 {
+            assert!(global_allowed(&mut state, Global::Informed, 3, t0 + i));
+        }
+        // Cap reached.
+        assert!(!global_allowed(&mut state, Global::Informed, 3, t0 + 3));
+        // The two budgets are independent.
+        assert!(global_allowed(&mut state, Global::Attempted, 1, t0 + 3));
+        assert!(!global_allowed(&mut state, Global::Attempted, 1, t0 + 4));
+        // An hour on, the earlier draws have aged out.
+        assert!(global_allowed(&mut state, Global::Informed, 3, t0 + 3601));
+    }
+
+    /// The regression for the growth bug: once a stranger can draw a reply, an
+    /// unpruned ledger is a remote way to grow state.json without limit.
+    #[test]
+    fn reply_ledgers_are_pruned_and_capped() {
+        let mut state = State::default();
+        let t0 = 1_760_000_000;
+        for i in 0..(REPLY_LEDGERS_MAX + 50) {
+            let mut sender = [0u8; 32];
+            sender[..8].copy_from_slice(&(i as u64).to_be_bytes());
+            assert!(reply_allowed(&mut state, &sender, &format!("payload-{i}"), t0));
+        }
+        assert!(
+            state.replies.len() <= REPLY_LEDGERS_MAX,
+            "{} ledgers kept",
+            state.replies.len()
+        );
+        // An hour later every ledger is stale, so the next call clears them all.
+        let mut fresh = [9u8; 32];
+        fresh[0] = 7;
+        assert!(reply_allowed(&mut state, &fresh, "later", t0 + 3601));
+        assert_eq!(state.replies.len(), 1, "stale ledgers should be dropped");
+    }
+
+    #[test]
+    fn registration_mode_defaults_to_silent_and_takes_an_override() {
+        let mut state = State::default();
+        // Nothing persisted: the configured default wins, whatever it is.
+        assert_eq!(state.registration(Mode::Open), Mode::Open);
+        assert_eq!(state.registration(Mode::Silent), Mode::Silent);
+        // An admin override outranks the config.
+        state.set_registration(Mode::Closed);
+        assert_eq!(state.registration(Mode::Open), Mode::Closed);
+        // And the enum's own default is the quiet one.
+        assert_eq!(Mode::default(), Mode::Silent);
+    }
+
+    /// A state.json written before these fields existed must still load.
+    #[test]
+    fn older_state_files_load_without_the_new_fields() {
+        let json = r#"{"feeds":[],"replies":[],"dedupe":[]}"#;
+        let state: State = serde_json::from_str(json).unwrap();
+        assert!(!state.drained);
+        assert_eq!(state.registration(Mode::Silent), Mode::Silent);
+    }
+
     #[test]
     fn state_round_trips_through_json() {
         let mut state = State::default();