gatewayd: state, what was handled, our threads and turns in flight

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)

crates/gatewayd/src/lib.rs

@@ -7,4 +7,5 @@ pub mod mm;
 pub mod net;
 pub mod secrets;
 pub mod sessions;
+pub mod state;
 pub mod ws;

crates/gatewayd/src/state.rs

@@ -0,0 +1,253 @@
+//! `<home>/gateway/state.json`: what `gatewayd` has handled, the threads it takes part in, and the
+//! turns in flight (M4a spec, section 9). Written atomically after every change. A file that
+//! cannot be read, parsed or written stops `gatewayd`: guessing would answer posts twice.
+
+use std::collections::BTreeMap;
+use std::fs;
+use std::io::{self, Write};
+use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt};
+use std::path::{Path, PathBuf};
+
+use serde::{Deserialize, Serialize};
+
+use crate::config::valid_id;
+
+pub const RECENT_KEPT: usize = 500;
+pub const THREADS_KEPT: usize = 5000;
+pub const RUNBOOK: &str = "see docs/runbook.md#gateway-state-damaged";
+
+#[derive(Debug)]
+pub enum StateError {
+    Read(PathBuf, String),
+    Write(PathBuf, io::Error),
+}
+
+impl std::fmt::Display for StateError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            StateError::Read(path, why) => write!(f, "{}: {why}\n{RUNBOOK}", path.display()),
+            StateError::Write(path, err) => {
+                write!(f, "{}: cannot write: {err}\n{RUNBOOK}", path.display())
+            }
+        }
+    }
+}
+
+impl std::error::Error for StateError {}
+
+/// A turn sent to `loopd` and not yet answered.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct InFlight {
+    pub session: String,
+    pub channel: String,
+    pub root: String,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct StateFile {
+    channels: BTreeMap<String, i64>,
+    recent: Vec<String>,
+    threads: Vec<String>,
+    in_flight: Vec<InFlight>,
+}
+
+impl StateFile {
+    /// Every id is a Mattermost id; a session is `mm-<id>`.
+    fn problem(&self) -> Option<String> {
+        // Every key of channels, every entry of recent and threads must be `valid_id` ("not a
+        // Mattermost id: <id with {:?}>"). Each in_flight entry: session is "mm-" + a valid id,
+        // channel and root valid ids ("a turn in flight is not valid: <session with {:?}>"). None
+        // when all are good.
+        for channel in self.channels.keys() {
+            if !valid_id(channel) {
+                return Some(format!("not a Mattermost id: {channel:?}"));
+            }
+        }
+        for post in &self.recent {
+            if !valid_id(post) {
+                return Some(format!("not a Mattermost id: {post:?}"));
+            }
+        }
+        for root in &self.threads {
+            if !valid_id(root) {
+                return Some(format!("not a Mattermost id: {root:?}"));
+            }
+        }
+        for turn in &self.in_flight {
+            let ok = match turn.session.strip_prefix("mm-") {
+                Some(rest) => valid_id(rest) && valid_id(&turn.channel) && valid_id(&turn.root),
+                None => false,
+            };
+            if !ok {
+                return Some(format!("a turn in flight is not valid: {:?}", turn.session));
+            }
+        }
+        None
+    }
+}
+
+pub struct State {
+    path: PathBuf,
+    file: StateFile,
+}
+
+impl State {
+    /// Read the state; a missing file is a first start.
+    pub fn load(path: &Path) -> Result<State, StateError> {
+        // Read the file. Missing (NotFound only) -> an empty StateFile. Any other read error, a
+        // parse error (serde_json), or `problem()` -> Read(path, why).
+        let text = match fs::read_to_string(path) {
+            Ok(text) => text,
+            Err(err) if err.kind() == io::ErrorKind::NotFound => {
+                return Ok(State {
+                    path: path.to_path_buf(),
+                    file: StateFile::default(),
+                });
+            }
+            Err(err) => return Err(StateError::Read(path.to_path_buf(), err.to_string())),
+        };
+        let file = match serde_json::from_str::<StateFile>(&text) {
+            Ok(file) => file,
+            Err(err) => return Err(StateError::Read(path.to_path_buf(), err.to_string())),
+        };
+        if let Some(why) = file.problem() {
+            return Err(StateError::Read(path.to_path_buf(), why));
+        }
+        Ok(State {
+            path: path.to_path_buf(),
+            file,
+        })
+    }
+
+    /// Write the state to its path atomically, in six steps; an error leaves the old file.
+    fn save(&self) -> Result<(), StateError> {
+        // `persist`, its error as Write(path, error).
+        self.persist()
+            .map_err(|err| StateError::Write(self.path.clone(), err))
+    }
+
+    fn persist(&self) -> io::Result<()> {
+        // The six steps of brokerd/src/state.rs, persist: 1. the parent directory, recursive, 0700;
+        // 2. serde_json::to_string plus "\n"; 3. "<path>.tmp" (with_extension("json.tmp")), create
+        //    + truncate, mode 0600; 4. write_all and sync_all; 5. rename over the path; 6. open the
+        //    directory and sync_all.
+        let dir = self
+            .path
+            .parent()
+            .ok_or_else(|| io::Error::other("state file has no parent"))?;
+        fs::DirBuilder::new()
+            .recursive(true)
+            .mode(0o700)
+            .create(dir)?;
+        let json = serde_json::to_string(&self.file).map_err(io::Error::other)?;
+        let bytes = format!("{json}\n");
+        let tmp = self.path.with_extension("json.tmp");
+        let mut file = fs::OpenOptions::new()
+            .write(true)
+            .create(true)
+            .truncate(true)
+            .mode(0o600)
+            .open(&tmp)?;
+        file.write_all(bytes.as_bytes())?;
+        file.sync_all()?;
+        fs::rename(&tmp, &self.path)?;
+        fs::File::open(dir)?.sync_all()
+    }
+
+    /// Was this post handled already?
+    pub fn seen(&self, post_id: &str) -> bool {
+        // Is the id in recent?
+        self.file.recent.contains(&post_id.to_string())
+    }
+
+    /// A post was handled (acted on or ignored): remember its id and move its channel's mark.
+    pub fn handled(
+        &mut self,
+        post_id: &str,
+        channel: &str,
+        create_at: i64,
+    ) -> Result<(), StateError> {
+        // If not seen: push the id to recent, then drop the oldest beyond RECENT_KEPT. The channel
+        // mark becomes the larger of its old value and create_at (a new channel starts at
+        // create_at). Save.
+        if !self.file.recent.contains(&post_id.to_string()) {
+            self.file.recent.push(post_id.to_string());
+            while self.file.recent.len() > RECENT_KEPT {
+                self.file.recent.remove(0);
+            }
+        }
+        let entry = self
+            .file
+            .channels
+            .entry(channel.to_string())
+            .or_insert(create_at);
+        *entry = (*entry).max(create_at);
+        self.save()
+    }
+
+    /// The `create_at` of the last post handled in a channel, if any.
+    pub fn since(&self, channel: &str) -> Option<i64> {
+        // The mark of the channel.
+        self.file.channels.get(channel).copied()
+    }
+
+    /// The channels with a mark, for catching up.
+    pub fn channels(&self) -> Vec<String> {
+        // The channel ids that have a mark.
+        self.file.channels.keys().cloned().collect()
+    }
+
+    /// Start catching up a channel from `at` (milliseconds), if it has no mark yet.
+    pub fn mark(&mut self, channel: &str, at: i64) -> Result<(), StateError> {
+        // A channel that has a mark is left alone (nothing saved). Otherwise set it to `at` and
+        // save.
+        if self.file.channels.contains_key(channel) {
+            return Ok(());
+        }
+        self.file.channels.insert(channel.to_string(), at);
+        self.save()
+    }
+
+    pub fn knows_thread(&self, root: &str) -> bool {
+        // Is the root in threads?
+        self.file.threads.contains(&root.to_string())
+    }
+
+    /// This Boxmaker takes part in a thread; the newest `THREADS_KEPT` are kept.
+    pub fn join_thread(&mut self, root: &str) -> Result<(), StateError> {
+        // Known -> nothing. Else push it, drop the oldest beyond THREADS_KEPT, save.
+        if self.file.threads.contains(&root.to_string()) {
+            return Ok(());
+        }
+        self.file.threads.push(root.to_string());
+        while self.file.threads.len() > THREADS_KEPT {
+            self.file.threads.remove(0);
+        }
+        self.save()
+    }
+
+    pub fn start_turn(&mut self, turn: InFlight) -> Result<(), StateError> {
+        // Remove any entry of the same session, push this one, save.
+        self.file.in_flight.retain(|t| t.session != turn.session);
+        self.file.in_flight.push(turn);
+        self.save()
+    }
+
+    pub fn end_turn(&mut self, session: &str) -> Result<(), StateError> {
+        // Remove the entries of the session, save.
+        self.file.in_flight.retain(|t| t.session != session);
+        self.save()
+    }
+
+    /// The turns left in flight by the last run, removed from the state.
+    pub fn take_in_flight(&mut self) -> Result<Vec<InFlight>, StateError> {
+        // Take the whole list out (std::mem::take); save only when it was not empty.
+        let taken = std::mem::take(&mut self.file.in_flight);
+        if taken.is_empty() {
+            return Ok(taken);
+        }
+        self.save().map(|()| taken)
+    }
+}

crates/gatewayd/tests/state.rs

@@ -0,0 +1,178 @@
+//! The state file: a first start, surviving a restart, its limits, and refusing a damaged file
+//! instead of guessing (M4a spec, section 9). Do not edit.
+
+#[path = "support/tmp.rs"]
+mod tmp;
+
+use std::os::unix::fs::PermissionsExt;
+
+use gatewayd::state::{InFlight, RECENT_KEPT, State, StateError, THREADS_KEPT};
+use tmp::TempDir;
+
+const CHAN: &str = "c0000000000000000000000000";
+const DM: &str = "d0000000000000000000000000";
+
+fn id(n: usize) -> String {
+    format!("p{n:025}")
+}
+
+fn turn(n: usize) -> InFlight {
+    InFlight {
+        session: format!("mm-{}", id(n)),
+        channel: DM.to_string(),
+        root: id(n),
+    }
+}
+
+#[test]
+fn a_first_start_then_a_restart() {
+    let dir = TempDir::new("state-restart");
+    let path = dir.path().join("gateway/state.json");
+    let mut s = State::load(&path).unwrap();
+    assert!(!path.exists(), "loading writes nothing");
+    assert_eq!(
+        (s.since(CHAN), s.channels().len(), s.seen(&id(1))),
+        (None, 0, false)
+    );
+    s.handled(&id(1), CHAN, 2000).unwrap();
+    s.handled(&id(2), CHAN, 1500).unwrap();
+    s.mark(DM, 3000).unwrap();
+    s.mark(DM, 9000).unwrap();
+    s.join_thread(&id(1)).unwrap();
+    s.start_turn(turn(1)).unwrap();
+    s.start_turn(turn(2)).unwrap();
+    s.end_turn(&format!("mm-{}", id(2))).unwrap();
+
+    let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
+    let dir_mode = std::fs::metadata(path.parent().unwrap())
+        .unwrap()
+        .permissions()
+        .mode()
+        & 0o777;
+    assert_eq!((mode, dir_mode), (0o600, 0o700));
+    assert!(!dir.path().join("gateway/state.json.tmp").exists());
+
+    let mut again = State::load(&path).unwrap();
+    assert_eq!(again.since(CHAN), Some(2000), "the mark never moves back");
+    assert_eq!(
+        again.since(DM),
+        Some(3000),
+        "mark only sets a channel without one"
+    );
+    assert_eq!(again.channels(), [CHAN, DM]);
+    assert!(again.seen(&id(1)) && again.seen(&id(2)));
+    assert!(again.knows_thread(&id(1)) && !again.knows_thread(&id(2)));
+    assert_eq!(again.take_in_flight().unwrap(), [turn(1)]);
+    assert!(
+        State::load(&path)
+            .unwrap()
+            .take_in_flight()
+            .unwrap()
+            .is_empty(),
+        "taking is saved"
+    );
+}
+
+#[test]
+fn only_the_newest_posts_and_threads_are_kept() {
+    let dir = TempDir::new("state-limits");
+    // A full file to start from: posts 0.. and threads 0.., at their limits.
+    let recent: Vec<String> = (0..RECENT_KEPT).map(id).collect();
+    let threads: Vec<String> = (0..THREADS_KEPT).map(id).collect();
+    let full =
+        serde_json::json!({"channels": {}, "recent": recent, "threads": threads, "in_flight": []});
+    let path = dir.write("state.json", &full.to_string());
+    let mut s = State::load(&path).unwrap();
+    s.handled(&id(RECENT_KEPT), CHAN, 1).unwrap();
+    s.handled(&id(RECENT_KEPT + 1), CHAN, 1).unwrap();
+    s.handled(&id(RECENT_KEPT + 1), CHAN, 1).unwrap();
+    s.join_thread(&id(THREADS_KEPT)).unwrap();
+    s.join_thread(&id(3)).unwrap();
+    let s = State::load(&path).unwrap();
+    assert!(!s.seen(&id(0)) && !s.seen(&id(1)) && s.seen(&id(2)) && s.seen(&id(RECENT_KEPT + 1)));
+    assert!(!s.knows_thread(&id(0)) && s.knows_thread(&id(1)) && s.knows_thread(&id(THREADS_KEPT)));
+    let text = std::fs::read_to_string(&path).unwrap();
+    let v: serde_json::Value = serde_json::from_str(&text).unwrap();
+    assert_eq!(
+        v["recent"].as_array().unwrap().len(),
+        RECENT_KEPT,
+        "a repeat is not stored twice"
+    );
+    assert_eq!(v["threads"].as_array().unwrap().len(), THREADS_KEPT);
+}
+
+#[test]
+fn a_damaged_file_stops_with_its_pointer() {
+    let dir = TempDir::new("state-damaged");
+    let good_turn = r#"{"session":"mm-p0000000000000000000000001","channel":"d0000000000000000000000000","root":"p0000000000000000000000001"}"#;
+    let cases = [
+        "".to_string(),
+        "{".to_string(),
+        "[]".to_string(),
+        r#"{"channels":{},"recent":[],"threads":[]}"#.to_string(),
+        r#"{"channels":{},"recent":[],"threads":[],"in_flight":[],"extra":1}"#.to_string(),
+        r#"{"channels":{"../x":1},"recent":[],"threads":[],"in_flight":[]}"#.to_string(),
+        r#"{"channels":{},"recent":["short"],"threads":[],"in_flight":[]}"#.to_string(),
+        r#"{"channels":{},"recent":[],"threads":["P0000000000000000000000000"],"in_flight":[]}"#
+            .to_string(),
+        format!(
+            r#"{{"channels":{{}},"recent":[],"threads":[],"in_flight":[{}]}}"#,
+            good_turn.replace("mm-", "xx-")
+        ),
+        format!(
+            r#"{{"channels":{{}},"recent":[],"threads":[],"in_flight":[{}]}}"#,
+            good_turn.replace("\"d0", "\"D0")
+        ),
+    ];
+    for (n, text) in cases.iter().enumerate() {
+        let path = dir.write(&format!("s{n}.json"), text);
+        match State::load(&path) {
+            Err(e @ StateError::Read(..)) => {
+                let message = e.to_string();
+                assert!(
+                    message.starts_with(&path.display().to_string()),
+                    "{message}"
+                );
+                assert!(
+                    message.ends_with("\nsee docs/runbook.md#gateway-state-damaged"),
+                    "{message}"
+                );
+            }
+            Err(e) => panic!("{text}: {e}"),
+            Ok(_) => panic!("accepted: {text}"),
+        }
+    }
+    let ok = dir.write(
+        "ok.json",
+        &format!(
+            r#"{{"channels":{{"{CHAN}":5}},"recent":[],"threads":[],"in_flight":[{good_turn}]}}"#
+        ),
+    );
+    assert_eq!(State::load(&ok).unwrap().since(CHAN), Some(5));
+    std::fs::create_dir(dir.path().join("adir")).unwrap();
+    assert!(matches!(
+        State::load(&dir.path().join("adir")),
+        Err(StateError::Read(..))
+    ));
+}
+
+#[test]
+fn a_failed_write_is_an_error_and_keeps_the_old_file() {
+    let dir = TempDir::new("state-readonly");
+    let sub = dir.path().join("gateway");
+    let path = sub.join("state.json");
+    let mut s = State::load(&path).unwrap();
+    s.handled(&id(1), CHAN, 5).unwrap();
+    let before = std::fs::read(&path).unwrap();
+    std::fs::set_permissions(&sub, std::fs::Permissions::from_mode(0o500)).unwrap();
+    let got = s.handled(&id(2), CHAN, 6);
+    std::fs::set_permissions(&sub, std::fs::Permissions::from_mode(0o700)).unwrap();
+    let e = got.unwrap_err();
+    assert!(matches!(e, StateError::Write(..)), "{e}");
+    assert!(
+        e.to_string()
+            .ends_with("see docs/runbook.md#gateway-state-damaged"),
+        "{e}"
+    );
+    assert_eq!(std::fs::read(&path).unwrap(), before);
+}

docs/implementer-log.md

@@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone.
 
 | Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model |
 |---|---|---|---|---|---|---|---|
+| M4a/12-gatewayd-state | 2026-09-23 | done | 1 | pass | none | Copied `tests/state.rs` and the `src/state.rs` skeleton, added `pub mod state;` to `lib.rs`. Filled `problem` (every `channels` key, `recent` and `threads` entry must be `valid_id` → "not a Mattermost id: <id with {:?}>"; each `in_flight` entry, session `mm-`+valid id with channel and root valid → "a turn in flight is not valid: <session with {:?}>"); `load` (NotFound → empty StateFile, any other read error, serde parse error or `problem()` → Read(path, why)); `save`/`persist` (the six atomic steps of brokerd's persist, io error mapped to Write, old file left on failure); `handled` (records unseen ids keeping the newest RECENT_KEPT, moves the channel mark to the max), `mark` (sets only a channel without a mark), `join_thread` (keeps the newest THREADS_KEPT), `start_turn`/`end_turn`/`take_in_flight` (removing by session, save only when in_flight was non-empty). Every mutating method saves before returning. All 4 tests pass in ~0.04 s; `make gate` prints `gate: ok` first run. | ? |
 | M4a/11-gatewayd-sessions | 2026-09-23 | done | 2 | fail | Removed the `resume` field from the copied `Pending` struct (written by the skeleton but never read) | Filled `named` (byte scan for `@`, the longest run of ASCII alnum/`.` `-` `_` after it, trailing `.` trimmed, lower-cased, empty runs skipped, then continue past the name) and the `Router` (`route` is a straight line of early returns in spec order: own -> Ignore(Own); kind not empty -> System; user not in allow.users -> NotAllowed; then the channel where "D" is always ours and "O"/"P"/"G" needs an allowed channel id plus `for_us`; the thread root is `root_id` or the post id; `!!...` keeps one `!` and queues, a lone `!` or `!approve`/`!deny` is a command where approve/deny answers M4B_COMMAND and anything else UNKNOWN_COMMAND, a command never reaches loopd; then Queue with session `mm-<root>`, `resume = root_id not empty`, `joins_thread = channel_type != "D"`). `for_us` is true when it names this bot (case-insensitive), otherwise a reply in a known thread that names only channel/here/all. `Queues`: `push` starts a turn when idle (Start with that message alone), waits while running and returns Full(thread) at the limit; `finish` joins the waiting texts with "\n\n" as a resume:true Batch, clears the queue and stays running, removing the session when nothing waits. `Pending.resume` was dead code (the next turn is always a continuation so `finish` hardcodes resume:true) so I removed it rather than allow a lint. First gate failed on clippy `manual_strip`; switched `starts_with("!!")`/`starts_with('!')` plus `&message[2..]`/`&message[1..]` slicing to `strip_prefix`. All 8 sessions tests pass; `make gate` prints `gate: ok` on the second run. | ? |
 | M4a/10-gatewayd-mm | 2026-09-23 | done | 1 | pass | none | Copied `tests/mm_json.rs`, `tests/mm_rest.rs` and `tests/support/http_server.rs`, added the `src/mm/mod.rs` and `src/mm/rest.rs` skeletons to `crates/gatewayd/src/mm/` and `pub mod mm;` to `lib.rs` (before `secrets`, alphabetical). Filled `mod.rs`: `Post::check` requires `id`/`user_id`/`channel_id` `valid_id` and `root_id` empty-or-`valid_id`, else `Json` quoting the offending id with `{:?}`; `json` is `serde_json::from_slice` mapped to `Json(e.to_string())`; `parse_event` matches `hello`/`posted`/other — `posted` takes `data.post` as a JSON *string* (an object or missing is `Json`), parses it, `check`s it, and reads `data.channel_type` (else ""), any other name (or an empty-name reply) is `Other(name)`; `typing` is `serde_json::json!` compacted; `since_list` walks `order` only (skipping ids not in `posts`, keeping `create_at > since && delete_at == 0` after `check`, deduping, then sorting by `(create_at, id)`), `full` when `order.len() >= SINCE_LIMIT`. Filled `rest.rs`: `Client::new` stores the three fields; `once` connects within `timeout`, sets the read timeout, sends `Authorization: Bearer <token>` (the only `expose`), `Accept`/`Content-Type` headers and `host_header`, mapping every error to `Net("<method> <path>: <e>")`; `call` loops `once` — 2xx returns the body, 401/403 `Auth`, 429 waits `rate_limit_wait` up to `RETRIES` then `RateLimited`, 5xx retried up to `RETRIES` times sleeping `RETRY_5XX`, else `Status` with the first `BODY_KEPT` lossy-UTF-8 chars via a `status_error` helper; `me`/`create_post`/`posts_since`/`direct_channel` build the four calls, `posts_since` and `me`/`direct_channel` reject non-`valid_id` ids as `Json` before sending. All 7 `mm_json` and 10 `mm_rest` tests pass (the latter ~3 s on two deliberate rate-limit waits); `make gate` prints `gate: ok` first run. | ? |
 | M4a/09-gatewayd-ws-conn | 2026-09-23 | done | 1 | pass | none | Filled the eight functions in the copied `crates/gatewayd/src/ws/conn.rs` skeleton (the written `poll` was the glue). `open`: `connector.connect(dead_after)` mapped to `Handshake(e.to_string())`, then `handshake` with `host_header(connector.server())`, a Ws with a new `Decoder` and `last_heard`/`last_ping` both `now`. `send`: `read_exact` 4 mask bytes from `random`, then `encode(opcode, payload, mask)` written and flushed. `send_text`: `send(TEXT, text.as_bytes())`. `take_messages`: loop `next_message`, `Text` returns, `Ping` answered with `send(PONG, &payload)`, `Pong` ignored, `Close` replies the code as 2 big-endian bytes (empty when none) via a best-effort `send(CLOSE, ...)` (the peer may be gone) and returns `Closed`. `keep_alive`: `now.duration_since(last_heard) >= dead_after` is `Dead`, else `now.duration_since(last_ping) >= ping_every` pings and stamps `last_ping`. `read_timeout`: the least of next-ping, next-dead and until-left (each `saturating_duration_since`), then `.max(1ms)`. `read_some`: `set_read_timeout`, a 16 KiB buffer, `Ok(0)` -> `Closed`, `Ok(n)` feeds `buf.get(..n).unwrap_or_default()` and stamps `last_heard`, `WouldBlock`/`TimedOut`/`Interrupted` -> `Ok(())`, any other `Err` -> `Io`. `close`: best-effort `send(CLOSE, &1000u16.to_be_bytes())`. `host_header`: host alone when the port is the scheme default (443 for tls, 80 otherwise) else `host:port`. All 10 tests in `tests/ws_conn.rs` pass five runs under a second; `make gate` prints `gate: ok` first run. | ? |