diff --git a/crates/gatewayd/src/lib.rs b/crates/gatewayd/src/lib.rs index 3f502d6..b257ff2 100644 --- a/crates/gatewayd/src/lib.rs +++ b/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; diff --git a/crates/gatewayd/src/state.rs b/crates/gatewayd/src/state.rs new file mode 100644 index 0000000..0cfc438 --- /dev/null +++ b/crates/gatewayd/src/state.rs @@ -0,0 +1,253 @@ +//! `/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, + recent: Vec, + threads: Vec, + in_flight: Vec, +} + +impl StateFile { + /// Every id is a Mattermost id; a session is `mm-`. + fn problem(&self) -> Option { + // Every key of channels, every entry of recent and threads must be `valid_id` ("not a + // Mattermost id: "). Each in_flight entry: session is "mm-" + a valid id, + // channel and root valid ids ("a turn in flight is not valid: "). 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 { + // 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::(&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. ".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 { + // The mark of the channel. + self.file.channels.get(channel).copied() + } + + /// The channels with a mark, for catching up. + pub fn channels(&self) -> Vec { + // 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, 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) + } +} diff --git a/crates/gatewayd/tests/state.rs b/crates/gatewayd/tests/state.rs new file mode 100644 index 0000000..fa02297 --- /dev/null +++ b/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 = (0..RECENT_KEPT).map(id).collect(); + let threads: Vec = (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); +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 858a4bd..cf8af16 100644 --- a/docs/implementer-log.md +++ b/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: "; each `in_flight` entry, session `mm-`+valid id with channel and root valid → "a turn in flight is not valid: "); `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-`, `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 ` (the only `expose`), `Accept`/`Content-Type` headers and `host_header`, mapping every error to `Net(" : ")`; `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. | ? |