From ed361db7185f8823b8362d29d1d1cd4b717ed1ee Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Thu, 24 Sep 2026 00:55:58 -0700 Subject: [PATCH] gatewayd: serve, the event loop, typing, catch-up and reconnecting Implemented-By: OpenCode session (model recorded in docs/implementer-log.md) --- crates/gatewayd/src/lib.rs | 1 + crates/gatewayd/src/serve/handle.rs | 213 +++++++++++++++ crates/gatewayd/src/serve/mod.rs | 297 ++++++++++++++++++++ crates/gatewayd/tests/serve.rs | 213 +++++++++++++++ crates/gatewayd/tests/serve_restart.rs | 192 +++++++++++++ crates/gatewayd/tests/support/fake_mm.rs | 331 +++++++++++++++++++++++ crates/gatewayd/tests/support/gateway.rs | 141 ++++++++++ docs/implementer-log.md | 1 + 8 files changed, 1389 insertions(+) create mode 100644 crates/gatewayd/src/serve/handle.rs create mode 100644 crates/gatewayd/src/serve/mod.rs create mode 100644 crates/gatewayd/tests/serve.rs create mode 100644 crates/gatewayd/tests/serve_restart.rs create mode 100644 crates/gatewayd/tests/support/fake_mm.rs create mode 100644 crates/gatewayd/tests/support/gateway.rs diff --git a/crates/gatewayd/src/lib.rs b/crates/gatewayd/src/lib.rs index 90717bb..a23379c 100644 --- a/crates/gatewayd/src/lib.rs +++ b/crates/gatewayd/src/lib.rs @@ -7,6 +7,7 @@ pub mod http; pub mod mm; pub mod net; pub mod secrets; +pub mod serve; pub mod sessions; pub mod state; pub mod ws; diff --git a/crates/gatewayd/src/serve/handle.rs b/crates/gatewayd/src/serve/handle.rs new file mode 100644 index 0000000..30f1e2a --- /dev/null +++ b/crates/gatewayd/src/serve/handle.rs @@ -0,0 +1,213 @@ +//! What the event loop does with a post, a finished turn, and the time between: routing, starting +//! turns, typing, and catching up after a gap. + +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; + +use crate::deliver::{LOOP_DOWN, deliver}; +use crate::mm::{Post, typing}; +use crate::serve::{Gateway, Stop}; +use crate::sessions::{BUSY, Batch, Ignored, Pushed, Route}; +use crate::state::InFlight; +use crate::ws::WsError; +use crate::ws::conn::Ws; + +/// Now, in Mattermost's milliseconds. +fn now_ms() -> i64 { + // `let ms = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_millis()).unwrap_or(0);` + // then `i64::try_from(ms).unwrap_or(i64::MAX)`. + let ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis()) + .unwrap_or(0); + i64::try_from(ms).unwrap_or(i64::MAX) +} + +impl Gateway { + /// Post in a thread; a failure is logged, not fatal. + pub(super) fn post(&self, channel: &str, root: &str, text: &str) { + // `if let Err(e) = self.client.create_post(channel, root, text)`: log + // "gatewayd: cannot post in (thread ): ", with `(self.log)(&line)`. + // Nothing else: posting records nothing in the state. + if let Err(e) = self.client.create_post(channel, root, text) { + (self.log)(&format!( + "gatewayd: cannot post in {channel} (thread {root}): {e}" + )); + } + } + + /// Is this channel one whose posts `gatewayd` keeps track of? + fn tracked(&self, channel: &str, channel_type: &str) -> bool { + // `channel_type == "D" || self.config.allow.channels.iter().any(|c| c == channel)`. + channel_type == "D" || self.config.allow.channels.iter().any(|c| c == channel) + } + + /// One new post, live or caught up. + pub(super) fn handle_post(&mut self, post: &Post, channel_type: &str) -> Result<(), Stop> { + // 1. `if !self.tracked(&post.channel_id, channel_type) || self.state.seen(&post.id)`: + // return Ok(()). + // 2. `self.state.handled(&post.id, &post.channel_id, post.create_at)?;` before anything + // else: after a crash, a post is never answered twice. + // 3. Route it, with the state answering "is this thread ours?": + // `let state = &self.state;` + // `let route = self.router.route(post, channel_type, &|root| state.knows_thread(root));` + // then `match route`: + // - Route::Ignore(Ignored::NotAllowed): log + // "gatewayd: ignored post from : not allowed" + // (never the text). + // - Route::Ignore(_): nothing. + // - Route::Reply { thread, text }: `self.post(&thread.channel, &thread.root, &text)`. + // - Route::Queue(message): if `message.joins_thread`, + // `self.state.join_thread(&message.thread.root)?`; then + // `match self.queues.push(message)`: + // Pushed::Start(batch) -> `self.start(batch)?`; Pushed::Waiting -> nothing; + // Pushed::Full(thread) -> `self.post(&thread.channel, &thread.root, BUSY)`. + // 4. Ok(()). + if !self.tracked(&post.channel_id, channel_type) || self.state.seen(&post.id) { + return Ok(()); + } + self.state + .handled(&post.id, &post.channel_id, post.create_at)?; + let state = &self.state; + let route = self + .router + .route(post, channel_type, &|root| state.knows_thread(root)); + match route { + Route::Ignore(Ignored::NotAllowed) => { + (self.log)(&format!( + "gatewayd: ignored post {} from {}: not allowed", + post.id, post.user_id + )); + } + Route::Ignore(_) => {} + Route::Reply { thread, text } => self.post(&thread.channel, &thread.root, &text), + Route::Queue(message) => { + if message.joins_thread { + self.state.join_thread(&message.thread.root)?; + } + match self.queues.push(message) { + Pushed::Start(batch) => self.start(batch)?, + Pushed::Waiting => {} + Pushed::Full(thread) => self.post(&thread.channel, &thread.root, BUSY), + } + } + } + Ok(()) + } + + /// Record the turn as in flight and run it on its own thread. + fn start(&mut self, batch: Batch) -> Result<(), Stop> { + // 1. `self.state.start_turn(InFlight { session: batch.session.as_str().to_string(), + // channel: batch.thread.channel.clone(), root: batch.thread.root.clone() })?;` + // 2. Clone what the thread takes, before the `move`: + // `let (client, socket, log, done) = (Arc::clone(&self.client), + // self.loop_socket.clone(), Arc::clone(&self.log), self.done_tx.clone());` + // `let (session, thread) = (batch.session.clone(), batch.thread.clone());` + // 3. `let spawned = std::thread::Builder::new().spawn(move || { + // deliver(client.as_ref(), &socket, &batch, &|line| log(line)); + // let _ = done.send(batch.session); });` + // 4. `if let Err(e) = spawned`: log "gatewayd: cannot start a thread for : " + // (`session.as_str()`), `self.post(&thread.channel, &thread.root, LOOP_DOWN)`, and + // `let _ = self.done_tx.send(session);` so the session does not stay busy. + // 5. Ok(()). + self.state.start_turn(InFlight { + session: batch.session.as_str().to_string(), + channel: batch.thread.channel.clone(), + root: batch.thread.root.clone(), + })?; + let (client, socket, log, done) = ( + Arc::clone(&self.client), + self.loop_socket.clone(), + Arc::clone(&self.log), + self.done_tx.clone(), + ); + let (session, thread) = (batch.session.clone(), batch.thread.clone()); + let spawned = std::thread::Builder::new().spawn(move || { + deliver(client.as_ref(), &socket, &batch, &|line| log(line)); + let _ = done.send(batch.session); + }); + if let Err(e) = spawned { + (self.log)(&format!( + "gatewayd: cannot start a thread for {}: {e}", + session.as_str() + )); + self.post(&thread.channel, &thread.root, LOOP_DOWN); + let _ = self.done_tx.send(session); + } + Ok(()) + } + + /// Turns that ended: out of flight, and the next batch of each session started. + pub(super) fn finished(&mut self) -> Result<(), Stop> { + // `while let Ok(session) = self.done_rx.try_recv() {` (never blocks) + // `self.state.end_turn(session.as_str())?;` + // `if let Some(batch) = self.queues.finish(&session) { self.start(batch)?; }` `}` + // Then Ok(()). + while let Ok(session) = self.done_rx.try_recv() { + self.state.end_turn(session.as_str())?; + if let Some(batch) = self.queues.finish(&session) { + self.start(batch)?; + } + } + Ok(()) + } + + /// Show this bot as typing in every thread with a turn running. + pub(super) fn typing(&mut self, ws: &mut Ws) -> Result<(), WsError> { + // `for thread in self.queues.threads() { self.seq += 1; + // ws.send_text(&typing(self.seq, &thread.channel, &thread.root))?; }` then Ok(()). + for thread in self.queues.threads() { + self.seq += 1; + ws.send_text(&typing(self.seq, &thread.channel, &thread.root))?; + } + Ok(()) + } + + /// Answer what arrived while `gatewayd` was away: the direct channel with each allowed user, + /// and each allowed channel. Written for you: it is the glue. + pub(super) fn catch_up(&mut self) -> Result<(), Stop> { + let mut channels = Vec::new(); + for user in self.config.allow.users.clone() { + match self.client.direct_channel(&self.me.id, &user) { + Ok(id) => channels.push((id, "D")), + Err(e) => (self.log)(&format!("gatewayd: no direct channel with {user}: {e}")), + } + } + channels.extend(self.config.allow.channels.iter().map(|c| (c.clone(), "O"))); + for (channel, channel_type) in channels { + self.catch_up_channel(&channel, channel_type)?; + } + Ok(()) + } + + /// One channel's posts since its mark. A channel seen for the first time starts from now. + fn catch_up_channel(&mut self, channel: &str, channel_type: &str) -> Result<(), Stop> { + // 1. `let Some(since) = self.state.since(channel) else { ... }`: with no mark, + // `self.state.mark(channel, now_ms())?` and return Ok(()). + // 2. `self.client.posts_since(channel, since)`: an Err(e) is logged + // "gatewayd: cannot catch up : " and returns Ok(()). + // 3. When `found.full`, log "gatewayd: : too many posts to catch up; some may be + // missed". + // 4. `for post in &found.posts { self.handle_post(post, channel_type)?; }` and Ok(()). + let Some(since) = self.state.since(channel) else { + self.state.mark(channel, now_ms())?; + return Ok(()); + }; + let found = match self.client.posts_since(channel, since) { + Ok(found) => found, + Err(e) => { + (self.log)(&format!("gatewayd: cannot catch up {channel}: {e}")); + return Ok(()); + } + }; + if found.full { + (self.log)(&format!( + "gatewayd: {channel}: too many posts to catch up; some may be missed" + )); + } + for post in &found.posts { + self.handle_post(post, channel_type)?; + } + Ok(()) + } +} diff --git a/crates/gatewayd/src/serve/mod.rs b/crates/gatewayd/src/serve/mod.rs new file mode 100644 index 0000000..32f0b9b --- /dev/null +++ b/crates/gatewayd/src/serve/mod.rs @@ -0,0 +1,297 @@ +//! Startup, the event loop and reconnecting (M4a spec, section 9). `run` returns only when +//! `gatewayd` must stop: a refused token, a state file it cannot keep, or a stop asked by a test. + +mod handle; + +use std::fs::File; +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::{self, Receiver, Sender}; +use std::time::{Duration, Instant}; + +use proto::SessionId; + +use crate::config::Config; +use crate::mm::rest::Client; +use crate::mm::{Event, Me, MmError, parse_event}; +use crate::net::Connector; +use crate::secrets::Secret; +use crate::sessions::{Queues, Router}; +use crate::state::{State, StateError}; +use crate::ws::conn::{Timing, Ws}; + +pub const UNREACHABLE: &str = "see docs/runbook.md#mattermost-unreachable"; +pub const AUTH_FAILED: &str = "see docs/runbook.md#mattermost-auth-failed"; +pub const START_FAILED: &str = "see docs/runbook.md#gatewayd-start-failed"; +pub const INTERRUPTED: &str = + "interrupted: gatewayd restarted before the answer arrived; ask again"; + +/// A log line: `stderr` in `main`, a record in tests. +pub type Log = Arc; + +/// Why `run` returned. +#[derive(Debug)] +pub enum Stop { + /// Mattermost refused the token (401 or 403). + Auth(u16), + State(StateError), + /// Something `gatewayd` needs at start is missing; the message names it. + Start(String), + /// The stop flag was set. + Asked, +} + +impl std::fmt::Display for Stop { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Stop::Auth(status) => { + write!( + f, + "gatewayd: Mattermost refused the token ({status})\n{AUTH_FAILED}" + ) + } + Stop::State(e) => write!(f, "gatewayd: {e}"), + Stop::Start(why) => write!(f, "gatewayd: {why}\n{START_FAILED}"), + Stop::Asked => write!(f, "gatewayd: stopped"), + } + } +} + +impl From for Stop { + fn from(e: StateError) -> Stop { + Stop::State(e) + } +} + +/// Timings a test shortens. +#[derive(Debug, Clone)] +pub struct Tuning { + /// The waits between connection attempts; the last repeats. + pub backoff: Vec, + /// How long one wait for a WebSocket message lasts, at most. + pub poll: Duration, + /// Connecting and each REST read. + pub rest_timeout: Duration, +} + +impl Default for Tuning { + fn default() -> Self { + let secs = [1, 2, 5, 10, 30].map(Duration::from_secs); + Tuning { + backoff: secs.to_vec(), + poll: Duration::from_millis(200), + rest_timeout: Duration::from_secs(30), + } + } +} + +/// Everything the event loop works with. +pub(crate) struct Gateway { + config: Config, + client: Arc, + router: Router, + me: Me, + state: State, + queues: Queues, + loop_socket: PathBuf, + done_tx: Sender, + done_rx: Receiver, + log: Log, + /// The typing requests' sequence number. + seq: u64, + /// The turns a restart cut off have been answered. + restarted: bool, +} + +/// The wait before attempt `n` (from 0) after a loss. +pub fn backoff(tuning: &Tuning, n: usize) -> Duration { + // `let last = tuning.backoff.last().copied().unwrap_or(Duration::from_secs(30));` then + // `tuning.backoff.get(n).copied().unwrap_or(last)`. + let last = tuning + .backoff + .last() + .copied() + .unwrap_or(Duration::from_secs(30)); + tuning.backoff.get(n).copied().unwrap_or(last) +} + +/// Sleep for `d`, in short steps, unless `stop` is set. +fn sleep_unless(stop: &AtomicBool, d: Duration) { + // `let until = Instant::now() + d;` then, while `!stop.load(Ordering::SeqCst)` and + // `Instant::now() < until`, sleep + // `Duration::from_millis(20).min(until.saturating_duration_since(Instant::now()))`. + let until = Instant::now() + d; + while !stop.load(Ordering::SeqCst) && Instant::now() < until { + std::thread::sleep( + Duration::from_millis(20).min(until.saturating_duration_since(Instant::now())), + ); + } +} + +/// Run `gatewayd` with its token, until it must stop. +pub fn run(config: Config, token: Secret, tuning: Tuning, log: Log, stop: &AtomicBool) -> Stop { + let server = match config.server() { + Ok(s) => s, + Err(why) => return Stop::Start(why), + }; + let connector = match Connector::new(server, config.mattermost.ca_file.as_deref()) { + Ok(c) => c, + Err(e) => return Stop::Start(e.to_string()), + }; + let state = match State::load(&config.state_path()) { + Ok(s) => s, + Err(e) => return Stop::State(e), + }; + let client = Arc::new(Client::new(connector, token, tuning.rest_timeout)); + let mut g = Gateway::new(config, client, state, log); + let mut failures = 0; + loop { + if stop.load(Ordering::SeqCst) { + return Stop::Asked; + } + let stopped = match connect(&g.config, &g.client, &tuning) { + Ok((me, mut ws)) => { + failures = 0; + g.connected(me) + .and_then(|()| g.catch_up()) + .and_then(|()| g.event_loop(&mut ws, &tuning, stop)) + } + Err(MmError::Auth(status)) => Err(Stop::Auth(status)), + Err(e) => { + let wait = backoff(&tuning, failures); + failures += 1; + (g.log)(&format!( + "gatewayd: cannot reach {}: {e}; trying again in {} s\n{UNREACHABLE}", + g.config.mattermost.url, + wait.as_secs() + )); + sleep_unless(stop, wait); + Ok(()) + } + }; + if let Err(stop) = stopped { + return stop; + } + } +} + +/// Who we are, and a WebSocket that has said hello. +fn connect(config: &Config, client: &Client, tuning: &Tuning) -> Result<(Me, Ws), MmError> { + let me = client.me()?; + let timing = Timing { + ping_every: Duration::from_millis(config.limits.ping_every_ms), + dead_after: Duration::from_millis(config.limits.dead_after_ms), + }; + let random = + File::open("/dev/urandom").map_err(|e| MmError::Net(format!("/dev/urandom: {e}")))?; + let net = |e: String| MmError::Net(format!("websocket: {e}")); + let mut ws = Ws::open( + client.connector(), + client.token().expose(), + timing, + Box::new(random), + ) + .map_err(|e| net(e.to_string()))?; + // The server says hello first; anything before it is skipped, and silence is an error. + let until = Instant::now() + timing.dead_after; + while Instant::now() < until { + if let Some(text) = ws.poll(tuning.poll).map_err(|e| net(e.to_string()))? + && matches!(parse_event(&text), Ok(Event::Hello)) + { + return Ok((me, ws)); + } + } + Err(net("no hello from the server".to_string())) +} + +impl Gateway { + fn new(config: Config, client: Arc, state: State, log: Log) -> Gateway { + let (done_tx, done_rx) = mpsc::channel(); + let limit = usize::try_from(config.limits.queue).unwrap_or(usize::MAX); + Gateway { + // Both are set on each connection, from `users/me`. + router: Router::new("", "", &[], &[]), + me: Me { + id: String::new(), + username: String::new(), + }, + loop_socket: config.loop_socket(), + queues: Queues::new(limit), + config, + client, + state, + done_tx, + done_rx, + log, + seq: 0, + restarted: false, + } + } + + /// A connection is up: say so, route as this user, and on the first one, answer the turns a + /// restart cut off. + fn connected(&mut self, me: Me) -> Result<(), Stop> { + // 1. Log exactly "gatewayd: connected to as " + // (`self.config.mattermost.url`, `me.username`), with `(self.log)(&line)`. + // 2. `let allow = &self.config.allow;` + // `self.router = Router::new(&me.id, &me.username, &allow.users, &allow.channels);` + // then `self.me = me;`. + // 3. The first time only (`if !self.restarted { self.restarted = true; ... }`): for each + // `turn` in `self.state.take_in_flight()?`, `self.post(&turn.channel, &turn.root, + // INTERRUPTED)`. A later reconnect must not: those turns are still running. + // 4. Ok(()). + (self.log)(&format!( + "gatewayd: connected to {} as {}", + self.config.mattermost.url, me.username + )); + let allow = &self.config.allow; + self.router = Router::new(&me.id, &me.username, &allow.users, &allow.channels); + self.me = me; + if !self.restarted { + self.restarted = true; + for turn in self.state.take_in_flight()? { + self.post(&turn.channel, &turn.root, INTERRUPTED); + } + } + Ok(()) + } + + /// Read events until the connection is lost (`Ok`) or `gatewayd` must stop. + fn event_loop(&mut self, ws: &mut Ws, tuning: &Tuning, stop: &AtomicBool) -> Result<(), Stop> { + let typing_every = Duration::from_millis(self.config.limits.typing_every_ms); + let mut last_typing = Instant::now(); + loop { + if stop.load(Ordering::SeqCst) { + return Err(Stop::Asked); + } + self.finished()?; + if last_typing.elapsed() >= typing_every { + last_typing = Instant::now(); + if let Err(e) = self.typing(ws) { + (self.log)(&format!( + "gatewayd: lost the connection: {e}\n{UNREACHABLE}" + )); + return Ok(()); + } + } + let text = match ws.poll(tuning.poll.min(typing_every)) { + Ok(Some(text)) => text, + Ok(None) => continue, + Err(e) => { + (self.log)(&format!( + "gatewayd: lost the connection: {e}\n{UNREACHABLE}" + )); + return Ok(()); + } + }; + match parse_event(&text) { + Ok(Event::Posted { post, channel_type }) => { + self.handle_post(&post, &channel_type)? + } + Ok(_) => {} + Err(e) => (self.log)(&format!("gatewayd: ignored an event: {e}")), + } + } + } +} diff --git a/crates/gatewayd/tests/serve.rs b/crates/gatewayd/tests/serve.rs new file mode 100644 index 0000000..6f29a84 --- /dev/null +++ b/crates/gatewayd/tests/serve.rs @@ -0,0 +1,213 @@ +//! `gatewayd` end to end, against a fake Mattermost and a fake `loopd`: who is answered, where, +//! and how (M4a spec, sections 7 and 8). Do not edit. + +#[path = "support/fake_loop.rs"] +mod fake_loop; +#[path = "support/fake_mm.rs"] +mod fake_mm; +#[path = "support/gateway.rs"] +mod gateway; +#[path = "support/tmp.rs"] +mod tmp; + +use std::sync::{Mutex, mpsc}; +use std::time::Duration; + +use fake_loop::{done, serve_loop}; +use fake_mm::{BOT, BOT_NAME, DM, EVE, EVE_DM, FakeMm, KYLE, id, post}; +use gateway::{OTHER, SHARED, WAIT, config, loop_dir, read_state, start, up}; +use gatewayd::serve::Stop; +use gatewayd::sessions::{BUSY, M4B_COMMAND}; +use tmp::TempDir; + +#[test] +fn a_direct_message_is_answered_in_its_thread_while_typing() { + let home = TempDir::new("serve-dm"); + let (mm, running, turns, mut ws) = up(&home, Duration::from_millis(600)); + let p1 = id('p', 1); + ws.posted(&post(&p1, KYLE, DM, "", "hello there", 5), "D"); + let turn = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (turn.session.as_str(), turn.content.as_str(), turn.resume), + (format!("mm-{p1}").as_str(), "hello there", false) + ); + let typing = ws.typing_within(Duration::from_millis(500)); + assert!(typing.len() >= 2, "{typing:?}"); + assert!( + typing.iter().all(|(c, p)| c == DM && *p == p1), + "{typing:?}" + ); + let posts = mm.wait_posts(1, WAIT); + assert_eq!( + posts, + [( + DM.to_string(), + p1.clone(), + "answer to hello there".to_string() + )] + ); + // Typing sent just before the answer may still be on its way; after that, it stops. + ws.typing_within(Duration::from_millis(300)); + assert!( + ws.typing_within(Duration::from_millis(400)).is_empty(), + "typing stops after the answer" + ); + let log = running.log(); + assert!( + log.iter() + .any(|l| l == &format!("gatewayd: connected to {} as {BOT_NAME}", mm.url())), + "{log:?}" + ); + assert!(matches!(running.finish(), Stop::Asked)); +} + +#[test] +fn anyone_else_gets_nothing_at_all() { + let home = TempDir::new("serve-stranger"); + let (mm, running, turns, mut ws) = up(&home, Duration::ZERO); + let p1 = id('p', 1); + ws.posted(&post(&p1, EVE, EVE_DM, "", "secret words", 5), "D"); + let mut own = post(&id('p', 2), BOT, DM, "", "my own post", 6); + own["user_id"] = serde_json::json!(BOT); + ws.posted(&own, "D"); + let log = running.wait_log("not allowed"); + assert!(turns.recv_timeout(Duration::from_millis(300)).is_err()); + assert!(ws.typing_within(Duration::from_millis(200)).is_empty()); + assert!(mm.posts().is_empty()); + assert!( + log.contains(&format!( + "gatewayd: ignored post {p1} from {EVE}: not allowed" + )), + "{log:?}" + ); + assert!(!log.iter().any(|l| l.contains("secret words")), "{log:?}"); +} + +#[test] +fn messages_during_a_turn_go_together_in_the_next() { + let home = TempDir::new("serve-burst"); + loop_dir(&home); + let (release_tx, release_rx) = mpsc::channel::<()>(); + let release = Mutex::new(release_rx); + let turns = serve_loop(&home.path().join("run/loop/loop.sock"), move |n, turn| { + if n == 0 { + let _ = release.lock().unwrap().recv_timeout(WAIT); + } + vec![done(&format!("answer to {}", turn.content))] + }); + let mm = FakeMm::start(); + let _running = start(config(&home, &mm.url(), "")); + let mut ws = mm.next_ws(WAIT); + let p1 = id('p', 1); + ws.posted(&post(&p1, KYLE, DM, "", "one", 5), "D"); + assert_eq!(turns.recv_timeout(WAIT).unwrap().content, "one"); + let saved = read_state(&home); + assert_eq!( + saved["in_flight"], + serde_json::json!([{"session": format!("mm-{p1}"), "channel": DM, "root": p1}]) + ); + ws.posted(&post(&id('p', 2), KYLE, DM, &p1, "two", 6), "D"); + ws.posted(&post(&id('p', 3), KYLE, DM, &p1, "three", 7), "D"); + std::thread::sleep(Duration::from_millis(200)); + release_tx.send(()).unwrap(); + let second = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + ( + second.session.as_str(), + second.content.as_str(), + second.resume + ), + (format!("mm-{p1}").as_str(), "two\n\nthree", true) + ); + let posts = mm.wait_posts(2, WAIT); + let texts: Vec<&str> = posts.iter().map(|(_, _, t)| t.as_str()).collect(); + assert_eq!(texts, ["answer to one", "answer to two\n\nthree"]); +} + +#[test] +fn a_full_queue_says_busy() { + let home = TempDir::new("serve-busy"); + loop_dir(&home); + let (release_tx, release_rx) = mpsc::channel::<()>(); + let release = Mutex::new(release_rx); + let _turns = serve_loop(&home.path().join("run/loop/loop.sock"), move |_, _| { + let _ = release.lock().unwrap().recv_timeout(WAIT); + vec![done("ok")] + }); + let mm = FakeMm::start(); + let _running = start(config(&home, &mm.url(), "queue = 1")); + let mut ws = mm.next_ws(WAIT); + let p1 = id('p', 1); + for (n, text) in ["run", "waits", "too many"].iter().enumerate() { + let n = u32::try_from(n).unwrap(); + let root = if n == 0 { String::new() } else { p1.clone() }; + ws.posted( + &post(&id('p', n + 1), KYLE, DM, &root, text, i64::from(n) + 5), + "D", + ); + } + let posts = mm.wait_posts(1, WAIT); + assert_eq!(posts, [(DM.to_string(), p1, BUSY.to_string())]); + release_tx.send(()).unwrap(); + release_tx.send(()).unwrap(); +} + +#[test] +fn channels_are_answered_only_when_named_or_in_our_thread() { + let home = TempDir::new("serve-channel"); + let (mm, _running, turns, mut ws) = up(&home, Duration::ZERO); + let p1 = id('p', 1); + ws.posted( + &post(&id('p', 9), KYLE, SHARED, "", "hello everyone", 4), + "O", + ); + ws.posted( + &post( + &id('p', 8), + KYLE, + OTHER, + "", + "@boxmaker-straylight elsewhere", + 4, + ), + "O", + ); + ws.posted( + &post(&p1, KYLE, SHARED, "", "@boxmaker-straylight start", 5), + "O", + ); + assert_eq!( + turns.recv_timeout(WAIT).unwrap().content, + "@boxmaker-straylight start" + ); + mm.wait_posts(1, WAIT); + ws.posted( + &post(&id('p', 2), KYLE, SHARED, &p1, "@hermes your turn", 6), + "O", + ); + ws.posted(&post(&id('p', 3), KYLE, SHARED, &p1, "and more", 7), "O"); + let next = turns.recv_timeout(WAIT).unwrap(); + assert_eq!((next.content.as_str(), next.resume), ("and more", true)); + let posts = mm.wait_posts(2, WAIT); + assert!( + posts.iter().all(|(c, r, _)| c == SHARED && *r == p1), + "{posts:?}" + ); + assert!(turns.recv_timeout(Duration::from_millis(200)).is_err()); + let saved = read_state(&home); + assert!(saved["channels"].get(OTHER).is_none(), "{saved}"); + assert_eq!(saved["threads"], serde_json::json!([p1])); +} + +#[test] +fn commands_are_answered_without_a_turn() { + let home = TempDir::new("serve-command"); + let (mm, _running, turns, mut ws) = up(&home, Duration::ZERO); + let p1 = id('p', 1); + ws.posted(&post(&p1, KYLE, DM, "", "!approve 3", 5), "D"); + assert_eq!( + mm.wait_posts(1, WAIT), + [(DM.to_string(), p1, M4B_COMMAND.to_string())] + ); + assert!(turns.recv_timeout(Duration::from_millis(200)).is_err()); +} diff --git a/crates/gatewayd/tests/serve_restart.rs b/crates/gatewayd/tests/serve_restart.rs new file mode 100644 index 0000000..39442d8 --- /dev/null +++ b/crates/gatewayd/tests/serve_restart.rs @@ -0,0 +1,192 @@ +//! `gatewayd` end to end across gaps: a restart, a lost connection, an unreachable server, a +//! refused token and a damaged state file (M4a spec, section 9). Do not edit. + +#[path = "support/fake_loop.rs"] +mod fake_loop; +#[path = "support/fake_mm.rs"] +mod fake_mm; +#[path = "support/gateway.rs"] +mod gateway; +#[path = "support/tmp.rs"] +mod tmp; + +use std::sync::{Mutex, mpsc}; +use std::time::Duration; + +use fake_loop::{done, serve_loop}; +use fake_mm::{DM, FakeMm, KYLE, id, post}; +use gateway::{SHARED, WAIT, answering, config, loop_dir, start, up}; +use gatewayd::serve::{INTERRUPTED, Stop}; +use tmp::TempDir; + +#[test] +fn a_restart_reports_the_cut_off_turn_and_catches_up() { + let home = TempDir::new("serve-restart"); + let (cut, seen, new1, new2) = (id('r', 1), id('p', 2), id('p', 3), id('p', 4)); + let state = serde_json::json!({ + "channels": {DM: 1000}, "recent": [seen], "threads": [], + "in_flight": [{"session": format!("mm-{cut}"), "channel": DM, "root": cut}] + }); + home.write("gateway/state.json", &state.to_string()); + loop_dir(&home); + let turns = answering(&home, Duration::ZERO); + let mm = FakeMm::start(); + mm.set_since( + DM, + &[ + post(&new2, KYLE, DM, "", "second", 2000), + post(&seen, KYLE, DM, "", "already answered", 1500), + post(&new1, KYLE, DM, "", "first", 1800), + ], + ); + let _running = start(config(&home, &mm.url(), "")); + let _ws = mm.next_ws(WAIT); + let first = turns.recv_timeout(WAIT).unwrap(); + let second = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (first.content.as_str(), second.content.as_str()), + ("first", "second") + ); + let posts = mm.wait_posts(3, WAIT); + assert_eq!(posts[0], (DM.to_string(), cut, INTERRUPTED.to_string())); + assert_eq!(posts.len(), 3, "{posts:?}"); + std::thread::sleep(Duration::from_millis(100)); + let saved: serde_json::Value = serde_json::from_str( + &std::fs::read_to_string(home.path().join("gateway/state.json")).unwrap(), + ) + .unwrap(); + assert_eq!(saved["in_flight"], serde_json::json!([])); + assert_eq!(saved["channels"][DM], 2000); +} + +#[test] +fn a_first_start_answers_no_history() { + let home = TempDir::new("serve-first"); + loop_dir(&home); + let turns = answering(&home, Duration::ZERO); + let mm = FakeMm::start(); + mm.set_since(DM, &[post(&id('p', 1), KYLE, DM, "", "old", 5)]); + let running = start(config(&home, &mm.url(), "")); + let _ws = mm.next_ws(WAIT); + running.wait_log("connected to"); + assert!(turns.recv_timeout(Duration::from_millis(300)).is_err()); + assert!( + !mm.calls().iter().any(|(_, p)| p.contains("since=")), + "{:?}", + mm.calls() + ); + let saved: serde_json::Value = serde_json::from_str( + &std::fs::read_to_string(home.path().join("gateway/state.json")).unwrap(), + ) + .unwrap(); + assert!( + saved["channels"][DM].as_i64().unwrap() > 1_700_000_000_000, + "marked from now" + ); + assert!(saved["channels"][SHARED].as_i64().is_some()); +} + +#[test] +fn a_lost_connection_is_made_again() { + let home = TempDir::new("serve-reconnect"); + let (mm, running, turns, ws) = up(&home, Duration::ZERO); + ws.drop_connection(); + let mut again = mm.next_ws(WAIT); + let log = running.wait_log("lost the connection"); + assert!( + log.iter() + .any(|l| l.ends_with("see docs/runbook.md#mattermost-unreachable")), + "{log:?}" + ); + let p1 = id('p', 1); + again.posted(&post(&p1, KYLE, DM, "", "still there?", 5), "D"); + assert_eq!(turns.recv_timeout(WAIT).unwrap().content, "still there?"); + assert_eq!(mm.wait_posts(1, WAIT).len(), 1); +} + +#[test] +fn a_refused_token_stops_gatewayd() { + let home = TempDir::new("serve-auth"); + let mm = FakeMm::start(); + mm.refuse_token(); + let running = start(config(&home, &mm.url(), "")); + let stop = running.join_within(); + assert!(matches!(stop, Stop::Auth(401)), "{stop}"); + assert_eq!( + stop.to_string(), + "gatewayd: Mattermost refused the token (401)\nsee docs/runbook.md#mattermost-auth-failed" + ); +} + +#[test] +fn an_unreachable_server_is_tried_again() { + let home = TempDir::new("serve-unreachable"); + let port = std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port(); + let url = format!("http://127.0.0.1:{port}"); + let running = start(config(&home, &url, "")); + std::thread::sleep(Duration::from_millis(300)); + let log = running.log(); + let tries: Vec<&String> = log + .iter() + .filter(|l| l.starts_with(&format!("gatewayd: cannot reach {url}: "))) + .collect(); + assert!(tries.len() >= 2, "{log:?}"); + assert!( + tries + .iter() + .all(|l| l + .ends_with("; trying again in 0 s\nsee docs/runbook.md#mattermost-unreachable")), + "{tries:?}" + ); + assert!(matches!(running.finish(), Stop::Asked)); +} + +#[test] +fn a_damaged_state_file_stops_at_once() { + let home = TempDir::new("serve-damaged"); + home.write("gateway/state.json", "{not json"); + let mm = FakeMm::start(); + let running = start(config(&home, &mm.url(), "")); + let stop = running.join_within(); + assert!(matches!(stop, Stop::State(_)), "{stop}"); + assert!( + stop.to_string() + .ends_with("see docs/runbook.md#gateway-state-damaged"), + "{stop}" + ); + assert!(mm.calls().is_empty(), "nothing is asked of Mattermost"); +} + +#[test] +fn a_reconnect_does_not_interrupt_a_running_turn() { + let home = TempDir::new("serve-reconnect-turn"); + loop_dir(&home); + let (release_tx, release_rx) = mpsc::channel::<()>(); + let release = Mutex::new(release_rx); + let turns = serve_loop(&home.path().join("run/loop/loop.sock"), move |_, _| { + let _ = release.lock().unwrap().recv_timeout(WAIT); + vec![done("the answer")] + }); + let mm = FakeMm::start(); + let running = start(config(&home, &mm.url(), "")); + let mut ws = mm.next_ws(WAIT); + let p1 = id('p', 1); + ws.posted(&post(&p1, KYLE, DM, "", "a long one", 5), "D"); + turns.recv_timeout(WAIT).unwrap(); + ws.drop_connection(); + let _again = mm.next_ws(WAIT); + running.wait_log("lost the connection"); + std::thread::sleep(Duration::from_millis(100)); + release_tx.send(()).unwrap(); + let posts = mm.wait_posts(1, WAIT); + std::thread::sleep(Duration::from_millis(200)); + assert_eq!( + mm.posts(), + [(DM.to_string(), p1, "the answer".to_string())], + "{posts:?}" + ); +} diff --git a/crates/gatewayd/tests/support/fake_mm.rs b/crates/gatewayd/tests/support/fake_mm.rs new file mode 100644 index 0000000..48cba6b --- /dev/null +++ b/crates/gatewayd/tests/support/fake_mm.rs @@ -0,0 +1,331 @@ +//! A fake Mattermost on 127.0.0.1, plain TCP: the four REST calls `gatewayd` makes, and the +//! WebSocket, whose events the test sends and whose requests it reads. Do not edit. + +#![allow(dead_code)] // each test file uses a different part of this module + +use std::collections::HashMap; +use std::io::{Read, Write}; +use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream}; +use std::sync::{Arc, Mutex, mpsc}; +use std::time::{Duration, Instant}; + +use gatewayd::ws::handshake::accept_for; +use serde_json::{Value, json}; + +pub const BOT: &str = "b0000000000000000000000000"; +pub const BOT_NAME: &str = "boxmaker-straylight"; +pub const KYLE: &str = "k0000000000000000000000000"; +pub const EVE: &str = "e0000000000000000000000000"; +/// The direct channel between the bot and Kyle, and between the bot and anyone else. +pub const DM: &str = "d0000000000000000000000000"; +pub const EVE_DM: &str = "f0000000000000000000000000"; + +#[derive(Default)] +struct Inner { + /// Status for `users/me`: 200 unless a test sets another. + me_status: u16, + /// The body for `channels//posts?since=`, by channel. + since: HashMap, + /// Every post made: channel, root, message. + posts: Vec<(String, String, String)>, + /// Every REST call: method and path. + calls: Vec<(String, String)>, +} + +pub struct FakeMm { + pub addr: SocketAddr, + inner: Arc>, + sockets: Mutex>, +} + +/// One WebSocket connection from `gatewayd`. +pub struct WsPeer { + writer: TcpStream, + /// The text of every text frame `gatewayd` sends. + pub texts: mpsc::Receiver, +} + +/// A post as Mattermost sends it. +pub fn post( + id: &str, + user: &str, + channel: &str, + root: &str, + message: &str, + create_at: i64, +) -> Value { + json!({ + "id": id, "create_at": create_at, "update_at": create_at, "delete_at": 0, "user_id": user, + "channel_id": channel, "root_id": root, "message": message, "type": "", "props": {} + }) +} + +pub fn id(prefix: char, n: u32) -> String { + format!("{prefix}{n:025}") +} + +fn frame(opcode: u8, payload: &[u8]) -> Vec { + let mut out = vec![0x80 | opcode]; + match payload.len() { + n if n < 126 => out.push(n as u8), + n => { + out.push(126); + out.extend_from_slice(&(n as u16).to_be_bytes()); + } + } + out.extend_from_slice(payload); + out +} + +impl WsPeer { + pub fn event(&mut self, value: &Value) { + let _ = self + .writer + .write_all(&frame(0x1, value.to_string().as_bytes())); + } + + pub fn posted(&mut self, post: &Value, channel_type: &str) { + let data = json!({"post": post.to_string(), "channel_type": channel_type, "team_id": ""}); + self.event(&json!({"event": "posted", "data": data, "broadcast": {}, "seq": 1})); + } + + /// End the connection without a close frame. + pub fn drop_connection(self) { + let _ = self.writer.shutdown(Shutdown::Both); + } + + /// The `user_typing` requests received within `wait`, as (channel, parent). + pub fn typing_within(&self, wait: Duration) -> Vec<(String, String)> { + let until = Instant::now() + wait; + let mut got = Vec::new(); + while let Ok(text) = self + .texts + .recv_timeout(until.saturating_duration_since(Instant::now())) + { + let v: Value = serde_json::from_str(&text).unwrap(); + if v["action"] == "user_typing" { + let data = &v["data"]; + got.push(( + data["channel_id"].as_str().unwrap().to_string(), + data["parent_id"].as_str().unwrap().to_string(), + )); + } + } + got + } +} + +fn read_head(stream: &mut TcpStream) -> Option { + let mut head = Vec::new(); + let mut byte = [0u8; 1]; + while !head.ends_with(b"\r\n\r\n") { + if stream.read(&mut byte).ok()? == 0 { + return None; + } + head.push(byte[0]); + } + String::from_utf8(head).ok() +} + +/// Unmask the client's frames and send each text on `tx`, until the connection ends. +fn read_frames(mut stream: TcpStream, tx: mpsc::Sender) { + let mut exact = |n: usize| -> Option> { + let mut buf = vec![0u8; n]; + stream.read_exact(&mut buf).ok().map(|()| buf) + }; + loop { + let Some(head) = exact(2) else { return }; + let len = match head[1] & 0x7F { + 126 => u16::from_be_bytes(exact(2).unwrap().try_into().unwrap()) as usize, + 127 => return, + n => n as usize, + }; + let Some(mask) = exact(4) else { return }; + let Some(raw) = exact(len) else { return }; + let payload: Vec = raw + .iter() + .zip(mask.iter().cycle()) + .map(|(b, m)| b ^ m) + .collect(); + if head[0] & 0x0F == 0x1 { + let _ = tx.send(String::from_utf8(payload).unwrap()); + } + } +} + +impl FakeMm { + pub fn start() -> FakeMm { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let inner = Arc::new(Mutex::new(Inner { + me_status: 200, + ..Inner::default() + })); + let (ws_tx, ws_rx) = mpsc::channel(); + let shared = Arc::clone(&inner); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(stream) = stream else { continue }; + let (inner, ws_tx) = (Arc::clone(&shared), ws_tx.clone()); + std::thread::spawn(move || connection(stream, &inner, &ws_tx)); + } + }); + FakeMm { + addr, + inner, + sockets: Mutex::new(ws_rx), + } + } + + pub fn url(&self) -> String { + format!("http://127.0.0.1:{}", self.addr.port()) + } + + pub fn refuse_token(&self) { + self.inner.lock().unwrap().me_status = 401; + } + + pub fn set_since(&self, channel: &str, posts: &[Value]) { + let order: Vec = posts.iter().map(|p| p["id"].clone()).collect(); + let map: serde_json::Map = posts + .iter() + .map(|p| (p["id"].as_str().unwrap().to_string(), p.clone())) + .collect(); + self.inner + .lock() + .unwrap() + .since + .insert(channel.to_string(), json!({"order": order, "posts": map})); + } + + /// The next WebSocket `gatewayd` opens, after its hello. + pub fn next_ws(&self, wait: Duration) -> WsPeer { + self.sockets + .lock() + .unwrap() + .recv_timeout(wait) + .expect("no WebSocket connection") + } + + pub fn posts(&self) -> Vec<(String, String, String)> { + self.inner.lock().unwrap().posts.clone() + } + + /// Wait until at least `n` posts were made, for at most `wait`. + pub fn wait_posts(&self, n: usize, wait: Duration) -> Vec<(String, String, String)> { + let until = Instant::now() + wait; + while self.posts().len() < n && Instant::now() < until { + std::thread::sleep(Duration::from_millis(10)); + } + self.posts() + } + + pub fn calls(&self) -> Vec<(String, String)> { + self.inner.lock().unwrap().calls.clone() + } +} + +fn connection(mut stream: TcpStream, inner: &Mutex, ws_tx: &mpsc::Sender) { + let Some(head) = read_head(&mut stream) else { + return; + }; + let mut words = head.split_whitespace(); + let (method, path) = ( + words.next().unwrap_or("").to_string(), + words.next().unwrap_or("").to_string(), + ); + if path == "/api/v4/websocket" { + let key = head + .lines() + .find_map(|l| l.strip_prefix("Sec-WebSocket-Key: ")) + .unwrap_or("") + .trim() + .to_string(); + let reply = format!( + "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {}\r\n\r\n", + accept_for(&key) + ); + let _ = stream.write_all(reply.as_bytes()); + let _ = stream.write_all(&frame( + 0x1, + br#"{"event":"hello","data":{},"broadcast":{},"seq":0}"#, + )); + let (tx, texts) = mpsc::channel(); + let reader = stream.try_clone().unwrap(); + std::thread::spawn(move || read_frames(reader, tx)); + let _ = ws_tx.send(WsPeer { + writer: stream, + texts, + }); + return; + } + let length = head + .lines() + .find_map(|l| { + l.to_ascii_lowercase() + .strip_prefix("content-length:") + .map(|v| v.trim().parse::().unwrap_or(0)) + }) + .unwrap_or(0); + let mut body = vec![0u8; length]; + let _ = stream.read_exact(&mut body); + let (status, answer) = rest(inner, &method, &path, &body); + let text = answer.to_string(); + let reply = format!( + "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{text}", + text.len() + ); + let _ = stream.write_all(reply.as_bytes()); +} + +fn rest(inner: &Mutex, method: &str, path: &str, body: &[u8]) -> (u16, Value) { + let mut inner = inner.lock().unwrap(); + inner.calls.push((method.to_string(), path.to_string())); + match (method, path) { + ("GET", "/api/v4/users/me") if inner.me_status == 200 => { + (200, json!({"id": BOT, "username": BOT_NAME})) + } + ("GET", "/api/v4/users/me") => ( + inner.me_status, + json!({"id": "api.context.session_expired.app_error"}), + ), + ("POST", "/api/v4/channels/direct") => { + let users: Vec = serde_json::from_slice(body).unwrap(); + let channel = if users.iter().any(|u| u == KYLE) { + DM + } else { + EVE_DM + }; + (201, json!({"id": channel, "type": "D"})) + } + ("POST", "/api/v4/posts") => { + let p: Value = serde_json::from_slice(body).unwrap(); + let n = u32::try_from(inner.posts.len()).unwrap(); + let (channel, root, message) = ( + p["channel_id"].as_str().unwrap(), + p["root_id"].as_str().unwrap(), + p["message"].as_str().unwrap(), + ); + inner + .posts + .push((channel.to_string(), root.to_string(), message.to_string())); + (201, post(&id('x', n), BOT, channel, root, message, 1)) + } + ("GET", p) if p.contains("/posts?since=") => { + let channel = p + .trim_start_matches("/api/v4/channels/") + .split('/') + .next() + .unwrap_or(""); + ( + 200, + inner + .since + .get(channel) + .cloned() + .unwrap_or(json!({"order": [], "posts": {}})), + ) + } + _ => (404, json!({"message": "not found"})), + } +} diff --git a/crates/gatewayd/tests/support/gateway.rs b/crates/gatewayd/tests/support/gateway.rs new file mode 100644 index 0000000..6af0432 --- /dev/null +++ b/crates/gatewayd/tests/support/gateway.rs @@ -0,0 +1,141 @@ +//! Running `gatewayd`'s serve loop in a test, against the fake Mattermost and the fake `loopd`. +//! Do not edit. + +#![allow(dead_code)] // each test file uses a different part of this module + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, mpsc}; +use std::thread::JoinHandle; +use std::time::Duration; + +use gatewayd::config::Config; +use gatewayd::secrets::Secret; +use gatewayd::serve::{Stop, Tuning, run}; +use proto::Turn; + +use crate::fake_loop::{done, serve_loop}; +use crate::fake_mm::{FakeMm, KYLE, WsPeer}; +use crate::tmp::TempDir; + +pub const SHARED: &str = "c0000000000000000000000000"; +pub const OTHER: &str = "o0000000000000000000000000"; +pub const WAIT: Duration = Duration::from_secs(5); + +pub struct Running { + pub stop: Arc, + pub log: Arc>>, + pub handle: Option>, +} + +impl Running { + pub fn log(&self) -> Vec { + self.log.lock().unwrap().clone() + } + + pub fn wait_log(&self, part: &str) -> Vec { + let until = std::time::Instant::now() + WAIT; + while !self.log().iter().any(|l| l.contains(part)) { + assert!( + std::time::Instant::now() < until, + "no log line with {part:?}: {:?}", + self.log() + ); + std::thread::sleep(Duration::from_millis(10)); + } + self.log() + } + + /// The `Stop` `run` returns by itself within 5 s; after that it is stopped, and the test fails. + pub fn join_within(mut self) -> Stop { + let handle = self.handle.take().unwrap(); + let until = std::time::Instant::now() + WAIT; + while !handle.is_finished() && std::time::Instant::now() < until { + std::thread::sleep(Duration::from_millis(10)); + } + self.stop.store(true, Ordering::SeqCst); + let stop = handle.join().unwrap(); + assert!(!matches!(stop, Stop::Asked), "run did not stop by itself"); + stop + } + + pub fn finish(mut self) -> Stop { + self.stop.store(true, Ordering::SeqCst); + self.handle.take().unwrap().join().unwrap() + } +} + +impl Drop for Running { + fn drop(&mut self) { + self.stop.store(true, Ordering::SeqCst); + } +} + +pub fn config(home: &TempDir, url: &str, extra: &str) -> Config { + let text = format!( + r#" +[mattermost] +url = "{url}" +[secrets.mattermost_token] +env = "NOT_READ_BY_RUN" +[allow] +users = ["{KYLE}"] +channels = ["{SHARED}"] +[paths] +home = "{}" +[limits] +typing_every_ms = 100 +{extra} +"#, + home.path().display() + ); + Config::parse(&text).unwrap() +} + +pub fn start(config: Config) -> Running { + let stop = Arc::new(AtomicBool::new(false)); + let log = Arc::new(Mutex::new(Vec::new())); + let tuning = Tuning { + backoff: vec![Duration::from_millis(50)], + poll: Duration::from_millis(20), + rest_timeout: WAIT, + }; + let (s, l) = (Arc::clone(&stop), Arc::clone(&log)); + let handle = std::thread::spawn(move || { + let sink: gatewayd::serve::Log = + Arc::new(move |line: &str| l.lock().unwrap().push(line.to_string())); + run(config, Secret::new("TOKEN".to_string()), tuning, sink, &s) + }); + Running { + stop, + log, + handle: Some(handle), + } +} + +/// A fake loop that answers every turn with "answer to ", after `delay`. +pub fn answering(home: &TempDir, delay: Duration) -> mpsc::Receiver { + serve_loop(&home.path().join("run/loop/loop.sock"), move |_, turn| { + std::thread::sleep(delay); + vec![done(&format!("answer to {}", turn.content))] + }) +} + +pub fn read_state(home: &TempDir) -> serde_json::Value { + let text = std::fs::read_to_string(home.path().join("gateway/state.json")).unwrap(); + serde_json::from_str(&text).unwrap() +} + +pub fn loop_dir(home: &TempDir) { + std::fs::create_dir_all(home.path().join("run/loop")).unwrap(); +} + +/// Start with a fake Mattermost and a fake loop; the first WebSocket is returned. +pub fn up(home: &TempDir, delay: Duration) -> (FakeMm, Running, mpsc::Receiver, WsPeer) { + loop_dir(home); + let turns = answering(home, delay); + let mm = FakeMm::start(); + let running = start(config(home, &mm.url(), "")); + let ws = mm.next_ws(WAIT); + running.wait_log("gatewayd: connected to "); + (mm, running, turns, ws) +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 4aa0a8e..7743ba7 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/14-gatewayd-serve | 2026-09-24 | done | 1 | pass | none | Copied `tests/serve.rs`, `tests/serve_restart.rs`, `tests/support/fake_mm.rs`, `tests/support/gateway.rs` and the `src/serve/mod.rs`/`handle.rs` skeletons, added `pub mod serve;` to `lib.rs` (before `secrets`, alphabetical). Filled `mod.rs`: `From for Stop` → `Stop::State(e)`; `backoff` (the last backoff repeats past its end); `sleep_unless` (interruptible 20 ms steps until the deadline); `Gateway::connected` (log "connected to as ", route as this user from `users/me`, and on the first connection only, answer each in-flight turn with INTERRUPTED so a later reconnect does not). Filled `handle.rs`: `now_ms` (millis with an i64 clamp); `post` (a failing post logs, never throws); `tracked` ("D" or an allowed channel); `handle_post` (skip untracked/seen, record `handled` before routing so a crash never answers twice, then route with the state answering `knows_thread` — an `NotAllowed` stranger is logged by post id and user id only, never the text, a reply posts, a queued message joins the thread then pushes with Start/Waiting/Full); `start` (record the in-flight turn, spawn `deliver` on its own thread, and on a spawn error post LOOP_DOWN and un-busy the session); `finished` (drain done, end each turn, start the next batch); `typing` (a `user_typing` per running thread, seq incremented); `catch_up_channel` (no mark → mark "now" and stop, history not answered; else `posts_since`, log when `full`, replay each post). `SessionId` has no `Display`, so the start-error line formats `session.as_str()`. `serve` 6 and `serve_restart` 7 pass five runs in ~1.3 s; `make gate` prints `gate: ok` first run. | ? | | M4a/13-gatewayd-deliver | 2026-09-23 | done | 1 | pass | none | Copied `tests/deliver.rs`, `tests/support/fake_loop.rs` and the `src/deliver.rs` skeleton, added `pub mod deliver;` to `lib.rs` (before `http`, alphabetical). Filled `error_text` (the snake_case name serde gives the ErrorCode via `serde_json::to_value`, falling back to `` without ever formatting `ErrorCode` with `{}` since it has no Display); `split_answer` (blank/whitespace-only → [EMPTY_ANSWER]; else while the rest is longer than `MAX_POST` *characters*, cut at the last newline within the first `MAX_POST` chars past position 0 — dropping that newline — else at the byte index of the `MAX_POST`-th char via `char_indices().nth(MAX_POST)`, never inside one char); `one_turn` (connect fails → LoopDown("cannot connect to : "), write one id-1 final Turn envelope (write error → LoopDown), then read: (1, not final, TurnEvent)→on_event, (1, final, TurnDone)→Answer(content), (1, final, Error)→Refused, a read error → LoopDown("the turn ended early: "), anything else → LoopDown("an unexpected frame")); `run_turn` (one_turn with `batch.resume`, and when Refused(NoSuchSession) with `batch.resume` true, one more with resume false to create the session, as `bxctl chat --session`); `deliver` (post in the thread — ApprovalPending posts `approval_text` at once, then Answer→every split part in order, Refused→error_text, LoopDown→log "gatewayd: : " and post LOOP_DOWN; a failing post logs "gatewayd: cannot post in (thread ): " through a small `post` helper). All 9 tests pass; `make gate` prints `gate: ok` first run. | ? | | 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. | ? |