session 3 commit 1/N: move transport codec to pmacs-protocol
Surfaced as session 3's first finding: pmacs-gpu can't attach to a
daemon without the length-prefix postcard codec
(read_message / write_message / TransportError / MAX_FRAME_BYTES),
but session 1 left those in the main pmacs crate. The wire-types
crate's boundary as drawn in session 1 didn't include the framing
codec — a real frontend needs both.
Classified as small under rule (iii) and absorbed in session 3.
Structural lesson recorded: transport is part of the wire contract,
not internal to the daemon.
src/transport.rs is now a re-export shim ('pub use
pmacs_protocol::transport::*;') so existing internal callers
(crate::transport::* in attach.rs, daemon.rs, attach_reconnect.rs)
keep working. Net test count unchanged: 11 transport tests now run
under 'cargo test -p pmacs-protocol' instead of 'cargo test --lib',
total 1314 across both crates.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
6a9aa57f69
commit
dd9d36926d
|
|
@ -38,6 +38,7 @@ pub mod cell;
|
|||
pub mod crdt;
|
||||
pub mod ids;
|
||||
pub mod message;
|
||||
pub mod transport;
|
||||
|
||||
pub use cell::{
|
||||
Attachment, Cell, CellCoord, CellSize, Color, DiffSpan, Glyph, Style, UnderlineStyle,
|
||||
|
|
@ -52,3 +53,4 @@ pub use message::{
|
|||
PROTOCOL_VERSION, ResourceBody, SUPPORTED_PROTOCOL_VERSIONS, SelectionSnapshot, StyleSegment,
|
||||
StyleSpan, is_supported_protocol_version, negotiate_capabilities,
|
||||
};
|
||||
pub use transport::{MAX_FRAME_BYTES, TransportError, read_message, write_message};
|
||||
|
|
|
|||
|
|
@ -0,0 +1,329 @@
|
|||
// transport.rs --- Length-prefix postcard codec.
|
||||
|
||||
//! Length-prefix framed postcard codec for the M5.5 frontend ↔ instance
|
||||
//! protocol (T M5.5b).
|
||||
//!
|
||||
//! # Wire format
|
||||
//!
|
||||
//! Each message is encoded as:
|
||||
//!
|
||||
//! ```text
|
||||
//! [u32 big-endian length][postcard bytes]
|
||||
//! ```
|
||||
//!
|
||||
//! - The length is the byte count of the postcard payload that follows.
|
||||
//! Zero-length payloads are valid (postcard encodes some types as
|
||||
//! empty byte strings).
|
||||
//! - Length values exceeding [`MAX_FRAME_BYTES`] are rejected without
|
||||
//! allocation on the read side, and refused before any bytes hit the
|
||||
//! wire on the write side. This caps both worst-case allocation and
|
||||
//! the maximum legitimate message size — large payloads should be
|
||||
//! chunked at a higher layer.
|
||||
//!
|
||||
//! # Encoding choice
|
||||
//!
|
||||
//! Postcard is a Serde-driven, no-std-friendly format chosen for its
|
||||
//! compactness on the cell-stream traffic (60 Hz cell-delta frames
|
||||
//! dominate the wire) and for the future option of a thin attach
|
||||
//! client without `tokio`. Schema evolution is handled by an explicit
|
||||
//! version handshake rather than the encoding itself; see
|
||||
//! [`crate::PROTOCOL_VERSION`].
|
||||
//!
|
||||
//! The worker-protocol encoding (spec §5.5) remains `MessagePack` via
|
||||
//! `rmp-serde`; that subsystem values schema flexibility over wire
|
||||
//! compactness.
|
||||
|
||||
use serde::{Serialize, de::DeserializeOwned};
|
||||
use std::io::{Read, Write};
|
||||
|
||||
/// Maximum legitimate frame payload size, in bytes.
|
||||
///
|
||||
/// 16 MiB. Comfortably above any single cell-delta frame in v0.1: a
|
||||
/// full 4K terminal at 60 Hz with truecolor styling fits well under a
|
||||
/// megabyte per frame. Frames larger than this are presumed bugs or
|
||||
/// hostile peers and are rejected.
|
||||
pub const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024;
|
||||
|
||||
/// Errors produced by [`read_message`] and [`write_message`].
|
||||
#[derive(Debug)]
|
||||
pub enum TransportError {
|
||||
/// Underlying I/O failed (broken pipe, connection reset, etc.).
|
||||
Io(std::io::Error),
|
||||
/// Postcard refused to encode the message (typically a `Serialize`
|
||||
/// implementation returning an error).
|
||||
Encode(postcard::Error),
|
||||
/// Postcard refused to decode the bytes — malformed payload from
|
||||
/// peer, or peer running an incompatible message shape that
|
||||
/// slipped past the version handshake.
|
||||
Decode(postcard::Error),
|
||||
/// Advertised or computed frame length exceeded [`MAX_FRAME_BYTES`].
|
||||
FrameTooLarge {
|
||||
/// The length the peer advertised (read side) or the size of
|
||||
/// the encoded payload (write side).
|
||||
len: usize,
|
||||
},
|
||||
/// Peer disconnected before a full frame could be read. The same
|
||||
/// error is returned for "EOF before any bytes," "EOF mid
|
||||
/// length-prefix," and "EOF mid payload"; the caller treats these
|
||||
/// identically.
|
||||
Eof,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for TransportError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::Io(e) => write!(f, "transport I/O error: {e}"),
|
||||
Self::Encode(e) => write!(f, "transport encode error: {e}"),
|
||||
Self::Decode(e) => write!(f, "transport decode error: {e}"),
|
||||
Self::FrameTooLarge { len } => {
|
||||
write!(f, "frame length {len} exceeds maximum {MAX_FRAME_BYTES}")
|
||||
}
|
||||
Self::Eof => write!(f, "peer disconnected before frame complete"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for TransportError {
|
||||
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
|
||||
match self {
|
||||
Self::Io(e) => Some(e),
|
||||
Self::Encode(e) | Self::Decode(e) => Some(e),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<std::io::Error> for TransportError {
|
||||
fn from(e: std::io::Error) -> Self {
|
||||
Self::Io(e)
|
||||
}
|
||||
}
|
||||
|
||||
/// Read a single framed message from `reader`.
|
||||
///
|
||||
/// Returns [`TransportError::Eof`] if the peer disconnected before a
|
||||
/// full frame arrived (whether at the start, mid length-prefix, or
|
||||
/// mid payload — all three look identical to the caller).
|
||||
pub fn read_message<M: DeserializeOwned>(reader: &mut impl Read) -> Result<M, TransportError> {
|
||||
let mut len_buf = [0u8; 4];
|
||||
read_exact_or_eof(reader, &mut len_buf)?;
|
||||
let len = u32::from_be_bytes(len_buf) as usize;
|
||||
if len > MAX_FRAME_BYTES {
|
||||
return Err(TransportError::FrameTooLarge { len });
|
||||
}
|
||||
let mut buf = vec![0u8; len];
|
||||
read_exact_or_eof(reader, &mut buf)?;
|
||||
postcard::from_bytes(&buf).map_err(TransportError::Decode)
|
||||
}
|
||||
|
||||
/// Write a single framed message to `writer`.
|
||||
///
|
||||
/// Returns [`TransportError::FrameTooLarge`] if the encoded form
|
||||
/// exceeds [`MAX_FRAME_BYTES`]; in that case no bytes are written.
|
||||
pub fn write_message<M: Serialize>(writer: &mut impl Write, msg: &M) -> Result<(), TransportError> {
|
||||
let payload = postcard::to_allocvec(msg).map_err(TransportError::Encode)?;
|
||||
if payload.len() > MAX_FRAME_BYTES {
|
||||
return Err(TransportError::FrameTooLarge { len: payload.len() });
|
||||
}
|
||||
let len = u32::try_from(payload.len()).expect("payload length bounded by MAX_FRAME_BYTES");
|
||||
writer.write_all(&len.to_be_bytes())?;
|
||||
writer.write_all(&payload)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Fill `buf` from `reader`, returning [`TransportError::Eof`] if the
|
||||
/// peer disconnects before the buffer is full. Retries on
|
||||
/// [`std::io::ErrorKind::Interrupted`].
|
||||
///
|
||||
/// `std::io::Read::read_exact` collapses both "read 0 bytes" and "read
|
||||
/// some-but-not-all" into `ErrorKind::UnexpectedEof`, but it is not
|
||||
/// guaranteed to retry on `Interrupted`. This helper makes both
|
||||
/// behaviors explicit.
|
||||
fn read_exact_or_eof(reader: &mut impl Read, buf: &mut [u8]) -> Result<(), TransportError> {
|
||||
let mut filled = 0;
|
||||
while filled < buf.len() {
|
||||
match reader.read(&mut buf[filled..]) {
|
||||
Ok(0) => return Err(TransportError::Eof),
|
||||
Ok(n) => filled += n,
|
||||
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
|
||||
Err(e) => return Err(TransportError::Io(e)),
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::{
|
||||
AttachRequest, FrontendCapabilities, FrontendEvent, FrontendId, Hello,
|
||||
InstanceCapabilities, InstanceIdentity, Key, KeyEvent, Modifiers, PROTOCOL_VERSION,
|
||||
};
|
||||
use std::io::Cursor;
|
||||
|
||||
fn round_trip<M: Serialize + DeserializeOwned + std::fmt::Debug + PartialEq>(msg: &M) {
|
||||
let mut buf = Vec::new();
|
||||
write_message(&mut buf, msg).expect("write");
|
||||
let mut cursor = Cursor::new(buf);
|
||||
let decoded: M = read_message(&mut cursor).expect("read");
|
||||
assert_eq!(&decoded, msg);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hello_round_trips_through_transport() {
|
||||
let h = Hello {
|
||||
protocol_version: PROTOCOL_VERSION,
|
||||
assigned_frontend_id: FrontendId(2),
|
||||
instance_identity: InstanceIdentity {
|
||||
pmacs_version: "0.1.0".into(),
|
||||
build_hash: None,
|
||||
instance_name: None,
|
||||
uptime_secs: 12,
|
||||
working_directory: "/tmp".into(),
|
||||
},
|
||||
instance_capabilities: InstanceCapabilities::default(),
|
||||
};
|
||||
round_trip(&h);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn attach_request_round_trips_through_transport() {
|
||||
let req = AttachRequest {
|
||||
protocol_version: PROTOCOL_VERSION,
|
||||
frontend_capabilities: FrontendCapabilities {
|
||||
synchronized_output: true,
|
||||
unicode_smp: true,
|
||||
true_color: true,
|
||||
mouse: true,
|
||||
bracketed_paste: true,
|
||||
terminal_kind: Some("xterm-256color".into()),
|
||||
multi_frontend: false,
|
||||
crdt_replica: false,
|
||||
semantic_render: false,
|
||||
},
|
||||
initial_size: crate::cell::CellSize::new(24, 80),
|
||||
};
|
||||
round_trip(&req);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn key_event_round_trips_through_transport() {
|
||||
let ev = FrontendEvent::Key(KeyEvent {
|
||||
frontend_id: FrontendId(2),
|
||||
key: Key::Char('a'),
|
||||
mods: Modifiers::CTRL,
|
||||
timestamp_ns: 0,
|
||||
});
|
||||
round_trip(&ev);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_input_returns_eof() {
|
||||
let mut cursor = Cursor::new(Vec::<u8>::new());
|
||||
match read_message::<Hello>(&mut cursor) {
|
||||
Err(TransportError::Eof) => {}
|
||||
other => panic!("expected Eof, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn truncated_length_prefix_returns_eof() {
|
||||
// Two bytes of a four-byte length prefix.
|
||||
let mut cursor = Cursor::new(vec![0x00, 0x10]);
|
||||
match read_message::<Hello>(&mut cursor) {
|
||||
Err(TransportError::Eof) => {}
|
||||
other => panic!("expected Eof, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn truncated_payload_returns_eof() {
|
||||
// Length advertises 100 bytes; only 5 follow.
|
||||
let mut bytes = 100u32.to_be_bytes().to_vec();
|
||||
bytes.extend_from_slice(&[0x01, 0x02, 0x03, 0x04, 0x05]);
|
||||
let mut cursor = Cursor::new(bytes);
|
||||
match read_message::<Hello>(&mut cursor) {
|
||||
Err(TransportError::Eof) => {}
|
||||
other => panic!("expected Eof, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn frame_larger_than_max_rejected_without_allocating() {
|
||||
// Advertise MAX_FRAME_BYTES + 1; we expect rejection before any
|
||||
// body bytes are read.
|
||||
let len = u32::try_from(MAX_FRAME_BYTES + 1).expect("fits in u32");
|
||||
let bytes = len.to_be_bytes().to_vec();
|
||||
let mut cursor = Cursor::new(bytes);
|
||||
match read_message::<Hello>(&mut cursor) {
|
||||
Err(TransportError::FrameTooLarge { len: l }) => {
|
||||
assert_eq!(l, MAX_FRAME_BYTES + 1);
|
||||
}
|
||||
other => panic!("expected FrameTooLarge, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn frame_at_exact_max_size_passes_length_check() {
|
||||
// Advertise exactly MAX_FRAME_BYTES — the boundary case must
|
||||
// not be rejected by the length check. We don't actually have
|
||||
// a payload this large; we expect Eof from the body fetch,
|
||||
// which proves the length check passed.
|
||||
let len = u32::try_from(MAX_FRAME_BYTES).expect("fits in u32");
|
||||
let bytes = len.to_be_bytes().to_vec();
|
||||
let mut cursor = Cursor::new(bytes);
|
||||
match read_message::<Hello>(&mut cursor) {
|
||||
Err(TransportError::Eof) => {}
|
||||
other => panic!("expected Eof at MAX_FRAME_BYTES boundary, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bad_postcard_bytes_return_decode_error() {
|
||||
// Length prefix says 8, payload is garbage bytes that do not
|
||||
// decode as a Hello.
|
||||
let payload = vec![0xFFu8; 8];
|
||||
let mut bytes = u32::try_from(payload.len()).unwrap().to_be_bytes().to_vec();
|
||||
bytes.extend_from_slice(&payload);
|
||||
let mut cursor = Cursor::new(bytes);
|
||||
match read_message::<Hello>(&mut cursor) {
|
||||
Err(TransportError::Decode(_)) => {}
|
||||
other => panic!("expected Decode, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn multiple_messages_back_to_back() {
|
||||
// Two messages share one buffer; framing must not leak state
|
||||
// between them.
|
||||
let h1 = FrontendEvent::Detach(FrontendId(1));
|
||||
let h2 = FrontendEvent::Detach(FrontendId(2));
|
||||
let mut buf = Vec::new();
|
||||
write_message(&mut buf, &h1).expect("write 1");
|
||||
write_message(&mut buf, &h2).expect("write 2");
|
||||
let mut cursor = Cursor::new(buf);
|
||||
let d1: FrontendEvent = read_message(&mut cursor).expect("read 1");
|
||||
let d2: FrontendEvent = read_message(&mut cursor).expect("read 2");
|
||||
match d1 {
|
||||
FrontendEvent::Detach(id) => assert_eq!(id, FrontendId(1)),
|
||||
other => panic!("expected Detach(1), got {other:?}"),
|
||||
}
|
||||
match d2 {
|
||||
FrontendEvent::Detach(id) => assert_eq!(id, FrontendId(2)),
|
||||
other => panic!("expected Detach(2), got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn read_after_consuming_only_message_returns_eof() {
|
||||
let h = FrontendEvent::Detach(FrontendId(7));
|
||||
let mut buf = Vec::new();
|
||||
write_message(&mut buf, &h).expect("write");
|
||||
let mut cursor = Cursor::new(buf);
|
||||
let _: FrontendEvent = read_message(&mut cursor).expect("read");
|
||||
match read_message::<FrontendEvent>(&mut cursor) {
|
||||
Err(TransportError::Eof) => {}
|
||||
other => panic!("expected Eof after consuming the only message, got {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
338
src/transport.rs
338
src/transport.rs
|
|
@ -1,329 +1,15 @@
|
|||
// transport.rs --- Length-prefix postcard codec.
|
||||
|
||||
//! Length-prefix framed postcard codec for the M5.5 frontend ↔ instance
|
||||
//! protocol (T M5.5b).
|
||||
//! Re-export of `pmacs_protocol::transport`.
|
||||
//!
|
||||
//! # Wire format
|
||||
//! Session 3 of the pmacs-gpu arc (`docs/pmacs-gpu-design.md`) moved
|
||||
//! the length-prefix postcard codec into `pmacs-protocol` so
|
||||
//! `pmacs-gpu` can use it without depending on the main `pmacs`
|
||||
//! crate. Existing `crate::transport::...` import paths inside the
|
||||
//! main `pmacs` crate continue to resolve through this re-export.
|
||||
//!
|
||||
//! Each message is encoded as:
|
||||
//!
|
||||
//! ```text
|
||||
//! [u32 big-endian length][postcard bytes]
|
||||
//! ```
|
||||
//!
|
||||
//! - The length is the byte count of the postcard payload that follows.
|
||||
//! Zero-length payloads are valid (postcard encodes some types as
|
||||
//! empty byte strings).
|
||||
//! - Length values exceeding [`MAX_FRAME_BYTES`] are rejected without
|
||||
//! allocation on the read side, and refused before any bytes hit the
|
||||
//! wire on the write side. This caps both worst-case allocation and
|
||||
//! the maximum legitimate message size — large payloads should be
|
||||
//! chunked at a higher layer.
|
||||
//!
|
||||
//! # Encoding choice
|
||||
//!
|
||||
//! Postcard is a Serde-driven, no-std-friendly format chosen for its
|
||||
//! compactness on the cell-stream traffic (60 Hz cell-delta frames
|
||||
//! dominate the wire) and for the future option of a thin attach
|
||||
//! client without `tokio`. Schema evolution is handled by an explicit
|
||||
//! version handshake rather than the encoding itself; see
|
||||
//! [`crate::protocol::PROTOCOL_VERSION`].
|
||||
//!
|
||||
//! The worker-protocol encoding (spec §5.5) remains `MessagePack` via
|
||||
//! `rmp-serde`; that subsystem values schema flexibility over wire
|
||||
//! compactness.
|
||||
//! The move surfaced an early Phase-A-style finding: the wire-types
|
||||
//! crate's boundary as drawn in session 1 didn't include the
|
||||
//! framing codec, but a real frontend needs both. Classified as
|
||||
//! *small* under rule (iii) and absorbed here; structural lesson is
|
||||
//! "transport is part of the wire contract."
|
||||
|
||||
use serde::{Serialize, de::DeserializeOwned};
|
||||
use std::io::{Read, Write};
|
||||
|
||||
/// Maximum legitimate frame payload size, in bytes.
|
||||
///
|
||||
/// 16 MiB. Comfortably above any single cell-delta frame in v0.1: a
|
||||
/// full 4K terminal at 60 Hz with truecolor styling fits well under a
|
||||
/// megabyte per frame. Frames larger than this are presumed bugs or
|
||||
/// hostile peers and are rejected.
|
||||
pub const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024;
|
||||
|
||||
/// Errors produced by [`read_message`] and [`write_message`].
|
||||
#[derive(Debug)]
|
||||
pub enum TransportError {
|
||||
/// Underlying I/O failed (broken pipe, connection reset, etc.).
|
||||
Io(std::io::Error),
|
||||
/// Postcard refused to encode the message (typically a `Serialize`
|
||||
/// implementation returning an error).
|
||||
Encode(postcard::Error),
|
||||
/// Postcard refused to decode the bytes — malformed payload from
|
||||
/// peer, or peer running an incompatible message shape that
|
||||
/// slipped past the version handshake.
|
||||
Decode(postcard::Error),
|
||||
/// Advertised or computed frame length exceeded [`MAX_FRAME_BYTES`].
|
||||
FrameTooLarge {
|
||||
/// The length the peer advertised (read side) or the size of
|
||||
/// the encoded payload (write side).
|
||||
len: usize,
|
||||
},
|
||||
/// Peer disconnected before a full frame could be read. The same
|
||||
/// error is returned for "EOF before any bytes," "EOF mid
|
||||
/// length-prefix," and "EOF mid payload"; the caller treats these
|
||||
/// identically.
|
||||
Eof,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for TransportError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::Io(e) => write!(f, "transport I/O error: {e}"),
|
||||
Self::Encode(e) => write!(f, "transport encode error: {e}"),
|
||||
Self::Decode(e) => write!(f, "transport decode error: {e}"),
|
||||
Self::FrameTooLarge { len } => {
|
||||
write!(f, "frame length {len} exceeds maximum {MAX_FRAME_BYTES}")
|
||||
}
|
||||
Self::Eof => write!(f, "peer disconnected before frame complete"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for TransportError {
|
||||
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
|
||||
match self {
|
||||
Self::Io(e) => Some(e),
|
||||
Self::Encode(e) | Self::Decode(e) => Some(e),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<std::io::Error> for TransportError {
|
||||
fn from(e: std::io::Error) -> Self {
|
||||
Self::Io(e)
|
||||
}
|
||||
}
|
||||
|
||||
/// Read a single framed message from `reader`.
|
||||
///
|
||||
/// Returns [`TransportError::Eof`] if the peer disconnected before a
|
||||
/// full frame arrived (whether at the start, mid length-prefix, or
|
||||
/// mid payload — all three look identical to the caller).
|
||||
pub fn read_message<M: DeserializeOwned>(reader: &mut impl Read) -> Result<M, TransportError> {
|
||||
let mut len_buf = [0u8; 4];
|
||||
read_exact_or_eof(reader, &mut len_buf)?;
|
||||
let len = u32::from_be_bytes(len_buf) as usize;
|
||||
if len > MAX_FRAME_BYTES {
|
||||
return Err(TransportError::FrameTooLarge { len });
|
||||
}
|
||||
let mut buf = vec![0u8; len];
|
||||
read_exact_or_eof(reader, &mut buf)?;
|
||||
postcard::from_bytes(&buf).map_err(TransportError::Decode)
|
||||
}
|
||||
|
||||
/// Write a single framed message to `writer`.
|
||||
///
|
||||
/// Returns [`TransportError::FrameTooLarge`] if the encoded form
|
||||
/// exceeds [`MAX_FRAME_BYTES`]; in that case no bytes are written.
|
||||
pub fn write_message<M: Serialize>(writer: &mut impl Write, msg: &M) -> Result<(), TransportError> {
|
||||
let payload = postcard::to_allocvec(msg).map_err(TransportError::Encode)?;
|
||||
if payload.len() > MAX_FRAME_BYTES {
|
||||
return Err(TransportError::FrameTooLarge { len: payload.len() });
|
||||
}
|
||||
let len = u32::try_from(payload.len()).expect("payload length bounded by MAX_FRAME_BYTES");
|
||||
writer.write_all(&len.to_be_bytes())?;
|
||||
writer.write_all(&payload)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Fill `buf` from `reader`, returning [`TransportError::Eof`] if the
|
||||
/// peer disconnects before the buffer is full. Retries on
|
||||
/// [`std::io::ErrorKind::Interrupted`].
|
||||
///
|
||||
/// `std::io::Read::read_exact` collapses both "read 0 bytes" and "read
|
||||
/// some-but-not-all" into `ErrorKind::UnexpectedEof`, but it is not
|
||||
/// guaranteed to retry on `Interrupted`. This helper makes both
|
||||
/// behaviors explicit.
|
||||
fn read_exact_or_eof(reader: &mut impl Read, buf: &mut [u8]) -> Result<(), TransportError> {
|
||||
let mut filled = 0;
|
||||
while filled < buf.len() {
|
||||
match reader.read(&mut buf[filled..]) {
|
||||
Ok(0) => return Err(TransportError::Eof),
|
||||
Ok(n) => filled += n,
|
||||
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
|
||||
Err(e) => return Err(TransportError::Io(e)),
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::protocol::{
|
||||
AttachRequest, FrontendCapabilities, FrontendEvent, FrontendId, Hello,
|
||||
InstanceCapabilities, InstanceIdentity, Key, KeyEvent, Modifiers, PROTOCOL_VERSION,
|
||||
};
|
||||
use std::io::Cursor;
|
||||
|
||||
fn round_trip<M: Serialize + DeserializeOwned + std::fmt::Debug + PartialEq>(msg: &M) {
|
||||
let mut buf = Vec::new();
|
||||
write_message(&mut buf, msg).expect("write");
|
||||
let mut cursor = Cursor::new(buf);
|
||||
let decoded: M = read_message(&mut cursor).expect("read");
|
||||
assert_eq!(&decoded, msg);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hello_round_trips_through_transport() {
|
||||
let h = Hello {
|
||||
protocol_version: PROTOCOL_VERSION,
|
||||
assigned_frontend_id: FrontendId(2),
|
||||
instance_identity: InstanceIdentity {
|
||||
pmacs_version: "0.1.0".into(),
|
||||
build_hash: None,
|
||||
instance_name: None,
|
||||
uptime_secs: 12,
|
||||
working_directory: "/tmp".into(),
|
||||
},
|
||||
instance_capabilities: InstanceCapabilities::default(),
|
||||
};
|
||||
round_trip(&h);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn attach_request_round_trips_through_transport() {
|
||||
let req = AttachRequest {
|
||||
protocol_version: PROTOCOL_VERSION,
|
||||
frontend_capabilities: FrontendCapabilities {
|
||||
synchronized_output: true,
|
||||
unicode_smp: true,
|
||||
true_color: true,
|
||||
mouse: true,
|
||||
bracketed_paste: true,
|
||||
terminal_kind: Some("xterm-256color".into()),
|
||||
multi_frontend: false,
|
||||
crdt_replica: false,
|
||||
semantic_render: false,
|
||||
},
|
||||
initial_size: crate::cell::CellSize::new(24, 80),
|
||||
};
|
||||
round_trip(&req);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn key_event_round_trips_through_transport() {
|
||||
let ev = FrontendEvent::Key(KeyEvent {
|
||||
frontend_id: FrontendId(2),
|
||||
key: Key::Char('a'),
|
||||
mods: Modifiers::CTRL,
|
||||
timestamp_ns: 0,
|
||||
});
|
||||
round_trip(&ev);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_input_returns_eof() {
|
||||
let mut cursor = Cursor::new(Vec::<u8>::new());
|
||||
match read_message::<Hello>(&mut cursor) {
|
||||
Err(TransportError::Eof) => {}
|
||||
other => panic!("expected Eof, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn truncated_length_prefix_returns_eof() {
|
||||
// Two bytes of a four-byte length prefix.
|
||||
let mut cursor = Cursor::new(vec![0x00, 0x10]);
|
||||
match read_message::<Hello>(&mut cursor) {
|
||||
Err(TransportError::Eof) => {}
|
||||
other => panic!("expected Eof, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn truncated_payload_returns_eof() {
|
||||
// Length advertises 100 bytes; only 5 follow.
|
||||
let mut bytes = 100u32.to_be_bytes().to_vec();
|
||||
bytes.extend_from_slice(&[0x01, 0x02, 0x03, 0x04, 0x05]);
|
||||
let mut cursor = Cursor::new(bytes);
|
||||
match read_message::<Hello>(&mut cursor) {
|
||||
Err(TransportError::Eof) => {}
|
||||
other => panic!("expected Eof, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn frame_larger_than_max_rejected_without_allocating() {
|
||||
// Advertise MAX_FRAME_BYTES + 1; we expect rejection before any
|
||||
// body bytes are read.
|
||||
let len = u32::try_from(MAX_FRAME_BYTES + 1).expect("fits in u32");
|
||||
let bytes = len.to_be_bytes().to_vec();
|
||||
let mut cursor = Cursor::new(bytes);
|
||||
match read_message::<Hello>(&mut cursor) {
|
||||
Err(TransportError::FrameTooLarge { len: l }) => {
|
||||
assert_eq!(l, MAX_FRAME_BYTES + 1);
|
||||
}
|
||||
other => panic!("expected FrameTooLarge, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn frame_at_exact_max_size_passes_length_check() {
|
||||
// Advertise exactly MAX_FRAME_BYTES — the boundary case must
|
||||
// not be rejected by the length check. We don't actually have
|
||||
// a payload this large; we expect Eof from the body fetch,
|
||||
// which proves the length check passed.
|
||||
let len = u32::try_from(MAX_FRAME_BYTES).expect("fits in u32");
|
||||
let bytes = len.to_be_bytes().to_vec();
|
||||
let mut cursor = Cursor::new(bytes);
|
||||
match read_message::<Hello>(&mut cursor) {
|
||||
Err(TransportError::Eof) => {}
|
||||
other => panic!("expected Eof at MAX_FRAME_BYTES boundary, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bad_postcard_bytes_return_decode_error() {
|
||||
// Length prefix says 8, payload is garbage bytes that do not
|
||||
// decode as a Hello.
|
||||
let payload = vec![0xFFu8; 8];
|
||||
let mut bytes = u32::try_from(payload.len()).unwrap().to_be_bytes().to_vec();
|
||||
bytes.extend_from_slice(&payload);
|
||||
let mut cursor = Cursor::new(bytes);
|
||||
match read_message::<Hello>(&mut cursor) {
|
||||
Err(TransportError::Decode(_)) => {}
|
||||
other => panic!("expected Decode, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn multiple_messages_back_to_back() {
|
||||
// Two messages share one buffer; framing must not leak state
|
||||
// between them.
|
||||
let h1 = FrontendEvent::Detach(FrontendId(1));
|
||||
let h2 = FrontendEvent::Detach(FrontendId(2));
|
||||
let mut buf = Vec::new();
|
||||
write_message(&mut buf, &h1).expect("write 1");
|
||||
write_message(&mut buf, &h2).expect("write 2");
|
||||
let mut cursor = Cursor::new(buf);
|
||||
let d1: FrontendEvent = read_message(&mut cursor).expect("read 1");
|
||||
let d2: FrontendEvent = read_message(&mut cursor).expect("read 2");
|
||||
match d1 {
|
||||
FrontendEvent::Detach(id) => assert_eq!(id, FrontendId(1)),
|
||||
other => panic!("expected Detach(1), got {other:?}"),
|
||||
}
|
||||
match d2 {
|
||||
FrontendEvent::Detach(id) => assert_eq!(id, FrontendId(2)),
|
||||
other => panic!("expected Detach(2), got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn read_after_consuming_only_message_returns_eof() {
|
||||
let h = FrontendEvent::Detach(FrontendId(7));
|
||||
let mut buf = Vec::new();
|
||||
write_message(&mut buf, &h).expect("write");
|
||||
let mut cursor = Cursor::new(buf);
|
||||
let _: FrontendEvent = read_message(&mut cursor).expect("read");
|
||||
match read_message::<FrontendEvent>(&mut cursor) {
|
||||
Err(TransportError::Eof) => {}
|
||||
other => panic!("expected Eof after consuming the only message, got {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
pub use pmacs_protocol::transport::*;
|
||||
|
|
|
|||
Loading…
Reference in New Issue