//! Protocol v5: bounded frames and the request/reply/event envelope. //! //! Every packet is `[u32 length][UTF-8 payload]`. The length is big-endian and //! capped, so a peer cannot make the reader allocate without bound. Inside a //! frame the payload is text: //! //! ```text //! REQ [idem=] //! RES //! EVT //! ``` //! //! The command and body grammar is the same text protocol the games already //! use (`HELLO`, `PLAY_START`, `GAME_RESULT`, ...). The envelope only adds //! correlation, event numbering and a hard size limit. use std::fmt; use std::io::{self, Read, Write}; /// Largest payload a peer may send in one frame. A game message is a few /// hundred bytes; 64 KiB is generous and still bounds the reader. pub const MAX_FRAME_LEN: usize = 64 * 1024; /// How many request ids a connection remembers for idempotent retries. pub const REPLAY_CACHE_LEN: usize = 256; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum FrameError { /// The declared payload length exceeds [`MAX_FRAME_LEN`]. TooLarge(usize), /// The payload was not valid UTF-8. InvalidUtf8, } impl fmt::Display for FrameError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { FrameError::TooLarge(length) => { write!( f, "frame of {length} bytes exceeds the {MAX_FRAME_LEN}-byte limit" ) } FrameError::InvalidUtf8 => write!(f, "frame payload is not valid UTF-8"), } } } impl std::error::Error for FrameError {} /// A bounded frame decoder. Feed it bytes from a socket, take whole frames. #[derive(Debug, Default)] pub struct FrameBuffer { buffer: Vec, } impl FrameBuffer { pub fn new() -> Self { Self::default() } pub fn push(&mut self, bytes: &[u8]) { self.buffer.extend_from_slice(bytes); } /// Return the next complete frame, if the buffer holds one. /// /// A declared length above the limit is reported immediately instead of /// waiting for bytes that will never be accepted; the caller should close /// the connection. pub fn next_frame(&mut self) -> Option> { if self.buffer.len() < 4 { return None; } let length = u32::from_be_bytes([ self.buffer[0], self.buffer[1], self.buffer[2], self.buffer[3], ]) as usize; if length > MAX_FRAME_LEN { self.buffer.clear(); return Some(Err(FrameError::TooLarge(length))); } if self.buffer.len() < 4 + length { return None; } let payload: Vec = self.buffer[4..4 + length].to_vec(); self.buffer.drain(..4 + length); match String::from_utf8(payload) { Ok(text) => Some(Ok(text)), Err(_) => Some(Err(FrameError::InvalidUtf8)), } } /// Read one frame from a blocking reader, preserving leftover bytes. /// `None` means EOF with no complete frame buffered. pub fn read_frame(&mut self, reader: &mut R) -> io::Result> { loop { match self.next_frame() { Some(Ok(frame)) => return Ok(Some(frame)), Some(Err(error)) => return Err(io::Error::new(io::ErrorKind::InvalidData, error)), None => {} } let mut chunk = [0u8; 4096]; match reader.read(&mut chunk) { Ok(0) => return Ok(None), Ok(n) => self.push(&chunk[..n]), Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, Err(error) => return Err(error), } } } /// True when a partial frame is buffered (used by tests and diagnostics). pub fn is_empty(&self) -> bool { self.buffer.is_empty() } } /// Encode one payload as a length-prefixed frame. pub fn encode_frame(payload: &str) -> Vec { let length = payload.len() as u32; let mut bytes = Vec::with_capacity(4 + payload.len()); bytes.extend_from_slice(&length.to_be_bytes()); bytes.extend_from_slice(payload.as_bytes()); bytes } /// Write one payload as a length-prefixed frame. pub fn write_frame(writer: &mut W, payload: &str) -> io::Result<()> { writer.write_all(&encode_frame(payload))?; writer.flush() } /// One client request. `id` is the correlation key; `idem` is optional /// metadata for logs and a future cross-connection cache. #[derive(Debug, Clone, PartialEq, Eq)] pub struct Request { pub id: u64, pub idem: Option, pub payload: String, } /// A frame received by a client. #[derive(Debug, Clone, PartialEq, Eq)] pub enum Packet { Reply { id: u64, body: String }, Event { seq: u64, body: String }, } pub fn request_line(id: u64, idem: Option<&str>, payload: &str) -> String { match idem { Some(idem) if !idem.is_empty() => format!("REQ {id} idem={idem} {payload}"), _ => format!("REQ {id} {payload}"), } } pub fn reply_line(id: u64, body: &str) -> String { format!("RES {id} {body}") } pub fn event_line(seq: u64, body: &str) -> String { format!("EVT {seq} {body}") } /// Parse a `REQ` payload. Errors are safe to return to the peer. pub fn parse_request(frame: &str) -> Result { let frame = frame.trim_end(); let rest = frame .strip_prefix("REQ ") .ok_or_else(|| "Malformed request: expected REQ".to_string())?; let (id_token, mut rest) = split_token(rest); let id: u64 = id_token .parse() .map_err(|_| "Malformed request id".to_string())?; let mut idem = None; let (next, after) = split_token(rest); if let Some(value) = next.strip_prefix("idem=") { if value.is_empty() { return Err("Malformed idempotency key".to_string()); } idem = Some(value.to_string()); rest = after; } Ok(Request { id, idem, payload: rest.to_string(), }) } /// Parse a `RES` or `EVT` payload on the client side. pub fn parse_packet(frame: &str) -> Result { let frame = frame.trim_end(); if let Some(rest) = frame.strip_prefix("RES ") { let (id_token, body) = split_token(rest); let id: u64 = id_token .parse() .map_err(|_| "Malformed reply id".to_string())?; return Ok(Packet::Reply { id, body: body.to_string(), }); } if let Some(rest) = frame.strip_prefix("EVT ") { let (seq_token, body) = split_token(rest); let seq: u64 = seq_token .parse() .map_err(|_| "Malformed event sequence".to_string())?; return Ok(Packet::Event { seq, body: body.to_string(), }); } Err(format!("Unexpected packet: {frame}")) } fn split_token(rest: &str) -> (&str, &str) { let rest = rest.trim_start(); match rest.find(' ') { Some(index) => (&rest[..index], rest[index + 1..].trim_start()), None => (rest, ""), } } /// A stable, user-facing message for a failed request. pub fn io_failure_message(error: &io::Error) -> &'static str { match error.kind() { io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock => "ERROR Request timed out", _ => "ERROR Connection lost", } }