//! `/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, /// Optional, so an M4a state file loads unchanged. #[serde(default)] approvals: Vec, } /// The posts of an approval shown in Mattermost and not yet ended (M4b spec, section 8). #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct ApprovalPosts { pub approval: u64, pub channel: String, /// First part first; the last is the one that is edited when it ends. pub posts: 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)); } } for shown in &self.approvals { if !valid_id(&shown.channel) || shown.posts.is_empty() || !shown.posts.iter().all(|post| valid_id(post)) { return Some(format!("approval {}'s posts are not valid", shown.approval)); } } 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) } /// The turn of a session, if one is in flight. pub fn turn_of(&self, session: &str) -> Option { // The entry of in_flight with that session, cloned. self.file .in_flight .iter() .find(|turn| turn.session == session) .cloned() } /// A new posting of an approval shown in Mattermost; it replaces any earlier one. pub fn add_approval(&mut self, shown: ApprovalPosts) -> Result<(), StateError> { // Remove any entry with the same approval, push this one, save. self.file.approvals.retain(|a| a.approval != shown.approval); self.file.approvals.push(shown); self.save() } /// Stop tracking an approval; nothing saved when it was not tracked. pub fn remove_approval(&mut self, approval: u64) -> Result<(), StateError> { // Remove the entries with that approval, save only if one was removed. let before = self.file.approvals.len(); self.file.approvals.retain(|a| a.approval != approval); if self.file.approvals.len() == before { return Ok(()); } self.save() } /// The approval posts left behind by the last run, removed from the state. pub fn take_approvals(&mut self) -> Result, StateError> { // Take them all out (std::mem::take), save only when there were some, return them. let taken = std::mem::take(&mut self.file.approvals); if taken.is_empty() { return Ok(taken); } self.save().map(|()| taken) } }