Plan M4a: gatewayd in 15 tasks, with skeletons and given tests
Each task's tests were run against a reference at its end state; the end states were replayed from master in order with the gate at each step (650 to 762 tests); each skeleton compiles against its tests and fails them. The reference is kept off this machine. Lessons T27 (every wait in a test has a limit) and T28 (mutate the reference before hand-over) come from this work. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,78 @@
|
||||
//! 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 {
|
||||
// Milliseconds since the Unix epoch as i64, with try_from (i64::MAX if it does not fit).
|
||||
todo!()
|
||||
}
|
||||
|
||||
impl Gateway {
|
||||
/// Post in a thread; a failure is logged, not fatal.
|
||||
pub(super) fn post(&self, channel: &str, root: &str, text: &str) {
|
||||
// `self.client.create_post`; an error is logged, "gatewayd: cannot post in <channel>
|
||||
// (thread <root>): <error>".
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// Is this channel one whose posts `gatewayd` keeps track of?
|
||||
fn tracked(&self, channel: &str, channel_type: &str) -> bool {
|
||||
// channel_type "D", or the channel is in allow.channels.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// One new post, live or caught up.
|
||||
pub(super) fn handle_post(&mut self, post: &Post, channel_type: &str) -> Result<(), Stop> {
|
||||
// 1. Not tracked, or seen -> nothing. 2. `state.handled(...)?` before anything else. 3.
|
||||
// Route it: NotAllowed -> log "gatewayd: ignored post <id> from <user>: not allowed"
|
||||
// (never the message); other Ignore -> nothing; Reply -> post it; Queue -> join the
|
||||
// thread when joins_thread, then push: Start -> `start`, Waiting -> nothing, Full ->
|
||||
// post BUSY.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// Record the turn as in flight and run it on its own thread.
|
||||
fn start(&mut self, batch: Batch) -> Result<(), Stop> {
|
||||
// `state.start_turn` (session, channel, root). Spawn with std::thread::Builder: `deliver`
|
||||
// with the client, the loop socket, the batch and the log, then send the session on
|
||||
// done_tx. If spawning fails: log "gatewayd: cannot start a thread for <session>: <e>",
|
||||
// post LOOP_DOWN in the thread, and send the session on done_tx.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// Turns that ended: out of flight, and the next batch of each session started.
|
||||
pub(super) fn finished(&mut self) -> Result<(), Stop> {
|
||||
// For each session on done_rx (try_recv, never blocking): `end_turn`, then `queues.finish`;
|
||||
// a batch it returns is started.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// Show this bot as typing in every thread with a turn running.
|
||||
pub(super) fn typing(&mut self, ws: &mut Ws) -> Result<(), WsError> {
|
||||
// For each of `queues.threads()`: seq += 1, send `typing(seq, channel, root)` as text.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// Answer what arrived while `gatewayd` was away: the direct channel with each allowed user,
|
||||
/// and each allowed channel. A channel seen for the first time starts from now.
|
||||
pub(super) fn catch_up(&mut self) -> Result<(), Stop> {
|
||||
// Channels: the direct channel with each allowed user ("D"; an error is logged "gatewayd:
|
||||
// no direct channel with <user>: <e>" and skipped), then each allowed channel ("O"). For
|
||||
// each: no mark -> `mark(channel, now_ms())` and skip; else `posts_since(channel, mark)`
|
||||
// (an error is logged "gatewayd: cannot catch up <channel>: <e>" and skipped); when full,
|
||||
// log "gatewayd: <channel>: too many posts to catch up; some may be missed"; `handle_post`
|
||||
// each post in order.
|
||||
todo!()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,232 @@
|
||||
//! 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).
|
||||
todo!()
|
||||
}
|
||||
}
|
||||
|
||||
/// 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 {
|
||||
// tuning.backoff[n], or its last entry when n is past the end (30 s if the list is empty). No
|
||||
// indexing.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// Sleep for `d`, in short steps, unless `stop` is set.
|
||||
fn sleep_unless(stop: &AtomicBool, d: Duration) {
|
||||
// Sleep in steps of at most 20 ms until `d` has passed or `stop` is set.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// 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> {
|
||||
// 1. `client.me()?`. 2. Timing from limits (ping_every_ms, dead_after_ms). 3. Open /dev/urandom
|
||||
// (an error is Net("/dev/urandom: <e>")). 4. `Ws::open(client.connector(),
|
||||
// client.token().expose(), timing, ..)`.
|
||||
// 5. Poll with tuning.poll until a text that `parse_event`s to Hello, for at most dead_after.
|
||||
// Every WebSocket error, and no hello in time, is Net("websocket: <why>").
|
||||
todo!()
|
||||
}
|
||||
|
||||
impl Gateway {
|
||||
fn new(config: Config, client: Arc<Client>, state: State, log: Log) -> Gateway {
|
||||
// A done channel, a Router with empty ids (set on connect), an empty Me, loop_socket from
|
||||
// config, Queues with limit limits.queue (usize::try_from), seq 0, restarted false.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// 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> {
|
||||
// Log exactly "gatewayd: connected to <url> as <username>". A new Router from me and the
|
||||
// allow lists; store me. The first time only (`restarted`): for each turn `take_in_flight`
|
||||
// gives back, post INTERRUPTED in its channel and root. A later reconnect must not: those
|
||||
// turns are still running.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// 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}")),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user