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

@@ -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;

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 <channel> (thread <root>): <e>", 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 <post.id> from <post.user_id>: 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>: <e>"
+        //    (`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 <channel>: <e>" and returns Ok(()).
+        // 3. When `found.full`, log "gatewayd: <channel>: 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(())
+    }
+}

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<dyn Fn(&str) + Send + Sync>;
+
+/// 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<StateError> 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<Duration>,
+    /// 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<Client>,
+    router: Router,
+    me: Me,
+    state: State,
+    queues: Queues,
+    loop_socket: PathBuf,
+    done_tx: Sender<SessionId>,
+    done_rx: Receiver<SessionId>,
+    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<Client>, 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 <url> as <username>"
+        //    (`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}")),
+            }
+        }
+    }
+}