neuwirth
/
pmacs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #41 from levineuwirth/worktree-pmacs-gpu-attach-loop 

Session 3: pmacs-gpu attach loop + rope reconstruction
This commit is contained in:
Levi Neuwirth 2026-05-20 17:12:21 +00:00 committed by GitHub
parent 6a9aa57f69 1506975ddb
commit f59c022d64
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 775 additions and 413 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -2458,6 +2458,7 @@ version = "0.0.1"
dependencies = [
"env_logger",
"glyphon",
"loro",
"pmacs-protocol",
"pollster",
"wgpu",

1
pmacs-gpu/Cargo.toml
View File

@ -44,6 +44,7 @@ env_logger = "0.11.10"
# rather than depending on `cosmic-text` directly so the build never
# ends up with two cosmic-text versions resolving to the same name.
glyphon = "0.11.0"
loro = "=1.12.0"
# Session 1's wire-types crate. Pulled in now so the dep graph is
# settled from session 2 forward; protocol consumption itself lands
# in session 3.

196
pmacs-gpu/src/attach.rs Normal file
View File

@ -0,0 +1,196 @@
//! Attach-mode client: connect to a running pmacs daemon over a Unix
//! socket, negotiate `semantic_render + crdt_replica`, then pump
//! `InstanceMessage` frames onto the winit event loop.
//!
//! Session 3 of the pmacs-gpu arc — see `docs/pmacs-gpu-design.md`.
//! Scope: handshake + decode the message stream. Importing the CRDT
//! snapshot, applying live ops, and reconstructing the rope happen
//! on the main thread, where the `LoroDoc` lives (it isn't trivially
//! `Send`; cross-thread shipping is the *decoded* `InstanceMessage`,
//! not the doc state).
//!
//! The reader thread blocks on a single `read_message` per iteration;
//! every received message becomes an [`AttachEvent`] forwarded
//! through [`winit::event_loop::EventLoopProxy::send_event`], which
//! wakes the main loop so the frame logic can apply the message and
//! redraw.
use std::os::unix::net::UnixStream;
use std::path::Path;
use std::sync::Arc;
use std::thread;
use pmacs_protocol::{
AttachRequest, FrontendCapabilities, Hello, InstanceMessage, PROTOCOL_VERSION,
SUPPORTED_PROTOCOL_VERSIONS, TransportError, is_supported_protocol_version, read_message,
write_message,
};
use winit::event_loop::EventLoopProxy;
use crate::AppEvent;
/// Errors the attach client surfaces. Kept narrow on purpose: the
/// hello-world fallback is the right recovery for any of these in
/// session 3, so the caller's only job is to log + drop back to the
/// inert renderer.
#[derive(Debug)]
pub enum AttachClientError {
/// Couldn't open the Unix socket.
Connect(std::io::Error),
/// Transport framing failed during the handshake.
Handshake(TransportError),
/// Server's `protocol_version` is outside `SUPPORTED_PROTOCOL_VERSIONS`.
VersionMismatch { server: u32, client: u32 },
}
impl std::fmt::Display for AttachClientError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Connect(e) => write!(f, "connect to daemon socket failed: {e}"),
Self::Handshake(e) => write!(f, "attach handshake failed: {e}"),
Self::VersionMismatch { server, client } => write!(
f,
"daemon protocol version {server} not in client's supported set (this client = \
{client}, supports {SUPPORTED_PROTOCOL_VERSIONS:?})"
),
}
}
}
impl std::error::Error for AttachClientError {}
/// One decoded event forwarded from the reader thread to the main
/// loop. `Message` carries the entire `InstanceMessage`; `Disconnected`
/// fires once when the reader thread exits (clean EOF or transport
/// error — both look identical from the main thread's perspective).
#[derive(Debug)]
pub enum AttachEvent {
/// A decoded message frame from the daemon.
Message(Box<InstanceMessage>),
/// The reader thread exited. Includes the disconnect reason for
/// logging on the main thread.
Disconnected(String),
}
/// Connect, handshake, and spawn the reader thread.
///
/// Returns once the handshake has completed and the reader thread is
/// running. The reader thread owns the read half of the stream; the
/// returned [`AttachClient`] retains the write half so the main loop
/// can eventually emit `FrontendEvent`s back to the daemon (session 4
/// will need this — selection / viewport / edits travel that way).
///
/// **Initial window size note** — `AttachRequest::initial_size` is
/// nominally a `CellSize` (rows × cols) anchored to the TUI. The
/// `pmacs-gpu` window isn't a cell grid; we send a placeholder of the
/// approximate cell count for the initial 800×200 window so the
/// daemon's initial render makes plausible space for content. This
/// is a small finding for session 3's audit (the wire-shape detail
/// "`AttachRequest`'s `CellSize` assumes a grid frontend"); resolution
/// classified under rule (iii) as deferred — a structural answer
/// belongs with Q#2's minimap variant or its own protocol thread,
/// not session 3's attach loop.
pub fn connect(
socket_path: &Path,
proxy: EventLoopProxy<AppEvent>,
) -> Result<AttachClient, AttachClientError> {
let stream = UnixStream::connect(socket_path).map_err(AttachClientError::Connect)?;
// Hello round-trip.
let mut handshake_stream = stream.try_clone().map_err(AttachClientError::Connect)?;
let hello: Hello = read_message(&mut handshake_stream).map_err(AttachClientError::Handshake)?;
if !is_supported_protocol_version(hello.protocol_version) {
return Err(AttachClientError::VersionMismatch {
server: hello.protocol_version,
client: PROTOCOL_VERSION,
});
}
eprintln!(
"pmacs-gpu: attached to daemon (protocol v{}, instance pmacs {})",
hello.protocol_version, hello.instance_identity.pmacs_version
);
// AttachRequest — declare the capabilities a semantic frontend
// needs. `multi_frontend` is included because the existing daemon
// gates `crdt_replica` behind it (M10.x dependency).
//
// **Daemon requirement**: the daemon must be built with the
// `crdt` feature (`cargo run --features crdt --bin pmacs --
// --daemon ...`). Without it the daemon's
// `InstanceCapabilities::default` returns `crdt_replica: false`,
// negotiation succeeds but no `BufferSnapshot` ever arrives, and
// the `pmacs-gpu` window sits on `(connecting...)` forever. This
// surfaced as a session-3 finding when manually validating the
// attach loop; classified as small under rule (iii) — recorded
// here so the next person attaching against a non-crdt daemon
// recognizes the symptom immediately.
let req = AttachRequest {
protocol_version: hello.protocol_version,
frontend_capabilities: FrontendCapabilities {
synchronized_output: false,
unicode_smp: true,
true_color: true,
mouse: false,
bracketed_paste: false,
terminal_kind: Some("pmacs-gpu".to_owned()),
multi_frontend: true,
crdt_replica: true,
semantic_render: true,
},
// Placeholder — see the doc comment above. Cell-shaped initial
// size is awkward for a pixel frontend; for session 3 we send
// approximate dimensions so the daemon's initial-render
// ranging is plausible.
initial_size: pmacs_protocol::CellSize::new(24, 80),
};
write_message(&mut handshake_stream, &req).map_err(AttachClientError::Handshake)?;
// Split read/write halves for the reader thread + main-thread
// write path. UnixStream clones share the underlying FD with
// independent buffer state — safe to read on one clone while the
// other writes (the FD is full-duplex).
let mut read_stream = stream.try_clone().map_err(AttachClientError::Connect)?;
let write_stream = stream;
// Reader thread. Each iteration: block on read_message, decode,
// forward via the event-loop proxy. Exits cleanly on EOF / any
// transport error; the main thread receives a single Disconnected
// event and drops back to the inert renderer.
thread::Builder::new()
.name("pmacs-gpu attach reader".into())
.spawn(move || {
loop {
match read_message::<InstanceMessage>(&mut read_stream) {
Ok(msg) => {
if proxy
.send_event(AppEvent::Attach(AttachEvent::Message(Box::new(msg))))
.is_err()
{
// Main loop torn down — quietly exit.
return;
}
}
Err(e) => {
let _ = proxy
.send_event(AppEvent::Attach(AttachEvent::Disconnected(e.to_string())));
return;
}
}
}
})
.expect("spawn attach reader thread");
Ok(AttachClient {
_write_stream: Arc::new(write_stream),
})
}
/// Handle the main loop keeps after `connect` returns. Session 3
/// holds only the write half (unused yet — session 4 wires events
/// back); the read half lives in the spawned reader thread. The
/// `Arc` is so the handle is `Clone` for future per-window cloning,
/// not because we need shared ownership today.
#[allow(dead_code)]
pub struct AttachClient {
_write_stream: Arc<UnixStream>,
}

321
pmacs-gpu/src/main.rs
View File

@ -1,25 +1,34 @@
//! pmacs-gpu — GPU/GUI frontend for pmacs.
//!
//! Session 2 of the pmacs-gpu arc (`docs/pmacs-gpu-design.md`):
//! **hello-world binary**. Opens a window via `winit`, initializes
//! `wgpu` against its surface, sets up `glyphon` text rendering with
//! the bundled `JetBrains` Mono font, and renders "hello, pmacs" once
//! per frame. No protocol consumption yet — that arrives in session
//! 3 (the attach loop). No editor state, no input handling beyond
//! close + Escape.
//! Two run modes:
//!
//! - **Hello-world** (no `--attach` argument; session 2 default).
//! Opens a window and renders "hello, pmacs" in the bundled
//! `JetBrains` Mono. Used to confirm the wgpu/winit/glyphon stack
//! without depending on a daemon.
//! - **Attach** (`--attach <unix-socket-path>`; session 3). Connects
//! to a running pmacs daemon, negotiates `semantic_render +
//! crdt_replica`, imports the daemon's `BufferSnapshot` into a
//! local loro replica, and renders the resulting rope text. Live
//! `CrdtOp` updates from the daemon are applied as they arrive.
//!
//! See `docs/pmacs-gpu-design.md` for the arc framing. Session 3's
//! gate at close is "Daemon ↔ frontend handshake; rope reconstruction
//! matches" — handled by the attach mode below.
//!
//! The bundled font is `JetBrains` Mono Regular, distributed under
//! the SIL Open Font License 1.1 (see `fonts/OFL.txt`). The design
//! note incorrectly recorded the license as Apache 2.0; the actual
//! license has been OFL since the family's open-source release. The
//! design-doc note will be corrected as part of this session.
//! the SIL Open Font License 1.1 (see `fonts/OFL.txt`).
mod attach;
use std::path::PathBuf;
use std::sync::Arc;
use glyphon::{
Attrs, Buffer, Cache, Color, Family, FontSystem, Metrics, Resolution, Shaping, SwashCache,
TextArea, TextAtlas, TextBounds, TextRenderer, Viewport,
};
use pmacs_protocol::InstanceMessage;
use wgpu::MultisampleState;
use winit::application::ApplicationHandler;
use winit::event::{ElementState, KeyEvent, WindowEvent};
@ -27,11 +36,12 @@ use winit::event_loop::{ActiveEventLoop, EventLoop};
use winit::keyboard::{Key, NamedKey};
use winit::window::{Window, WindowId};
use crate::attach::{AttachClient, AttachEvent};
/// Bundled font (SIL Open Font License 1.1 — see `fonts/OFL.txt`).
const JETBRAINS_MONO: &[u8] = include_bytes!("../fonts/JetBrainsMono-Regular.ttf");
/// Initial window size in logical pixels. Session 2 is fixed-size for
/// simplicity; resizes still work, this is just the boot dimension.
/// Initial window size in logical pixels.
const INITIAL_WIDTH: u32 = 800;
const INITIAL_HEIGHT: u32 = 200;
@ -43,60 +53,156 @@ const BG: wgpu::Color = wgpu::Color {
a: 1.0,
};
/// Hello-world payload. Stays inert here — session 3 wires this to
/// the daemon's `BufferSnapshot` instead.
/// Text the hello-world (and attach-pre-snapshot / attach-failed)
/// modes render. Once the daemon's `BufferSnapshot` arrives the
/// rendered text becomes the rope contents instead.
const HELLO_TEXT: &str = "hello, pmacs";
fn main() {
// wgpu emits useful trace output on adapter selection / surface
// configuration. The default `RUST_LOG=info` is fine for
// development.
env_logger::init();
/// Container id the daemon uses on its loro `LoroDoc` for the
/// buffer's text. Must match `pmacs::crdt::CrdtState`'s container
/// name (`"body"`).
const LORO_TEXT_CONTAINER: &str = "body";
let event_loop = EventLoop::new().expect("create winit event loop");
let mut app = App { state: None };
/// Custom events delivered to the winit event loop. The reader thread
/// in `attach.rs` forwards each decoded `InstanceMessage` through the
/// `EventLoopProxy<AppEvent>` it was handed by `connect()`; the main
/// thread dispatches them in `user_event` below.
#[derive(Debug)]
pub enum AppEvent {
/// A message or disconnect notification from the attach reader
/// thread.
Attach(AttachEvent),
}
/// CLI mode derived from argv.
#[derive(Debug, Clone)]
enum Mode {
/// `pmacs-gpu` (no args): inert hello-world.
HelloWorld,
/// `pmacs-gpu --attach <socket>`: connect + render the daemon's
/// rope.
Attach { socket: PathBuf },
}
fn main() {
env_logger::init();
let mode = parse_args(std::env::args().skip(1).collect());
let event_loop = EventLoop::<AppEvent>::with_user_event()
.build()
.expect("create winit event loop");
let proxy = event_loop.create_proxy();
let mut app = App {
mode,
proxy: Some(proxy),
state: None,
attach_client: None,
};
event_loop
.run_app(&mut app)
.expect("winit event loop run_app");
}
/// Top-level application handler. Holds an `Option<State>` because
/// winit 0.30 requires the window + GPU resources to be created
/// *after* `resumed()` fires, not at `main()` start.
struct App {
state: Option<State>,
/// Tiny argv parser. No `clap` because the surface is genuinely two
/// shapes; full CLI parsing arrives when there's more to parse. The
/// `for` ranges over a small set: at most one `--attach <socket>` or
/// `--help` arrives, plus any stray unrecognized flag.
fn parse_args(args: Vec<String>) -> Mode {
let mut iter = args.into_iter();
let Some(first) = iter.next() else {
return Mode::HelloWorld;
};
match first.as_str() {
"--attach" => {
let socket = iter.next().unwrap_or_else(|| {
eprintln!("pmacs-gpu: --attach requires a socket path");
std::process::exit(2);
});
Mode::Attach {
socket: PathBuf::from(socket),
}
}
"--help" | "-h" => {
eprintln!(
"pmacs-gpu — GPU/GUI frontend for pmacs\n\nUSAGE:\n pmacs-gpu \
hello-world (renders \"hello, pmacs\")\n pmacs-gpu --attach <socket> \
connect to a daemon's Unix socket and render its rope\n"
);
std::process::exit(0);
}
other => {
eprintln!("pmacs-gpu: unrecognized argument: {other}");
std::process::exit(2);
}
}
}
/// All resources owned by a single running pmacs-gpu instance: the
/// window, the wgpu device/queue/surface, the glyphon stack, and the
/// shaped text buffer.
/// Top-level application handler. `state` is `Option` because winit
/// 0.30 builds the window in `resumed()`, not at `main()` start;
/// `attach_client` is held so the write half of the Unix stream
/// stays alive for as long as the window does.
struct App {
mode: Mode,
/// The event-loop proxy is taken in `resumed()` and handed to the
/// reader thread. `Option` only because it can't be cloned out of
/// a non-Option in a borrow.
proxy: Option<winit::event_loop::EventLoopProxy<AppEvent>>,
state: Option<State>,
/// Held for lifetime; session 3 doesn't write back yet.
#[allow(dead_code)]
attach_client: Option<AttachClient>,
}
/// All resources owned by one running pmacs-gpu instance.
struct State {
window: Arc<Window>,
// wgpu plumbing.
device: wgpu::Device,
queue: wgpu::Queue,
surface: wgpu::Surface<'static>,
config: wgpu::SurfaceConfiguration,
// glyphon plumbing.
font_system: FontSystem,
swash_cache: SwashCache,
viewport: Viewport,
atlas: TextAtlas,
text_renderer: TextRenderer,
buffer: Buffer,
/// What the buffer is currently shaped to. Held so we can detect
/// no-op updates and skip the re-shape.
current_text: String,
/// Local CRDT replica seeded by `BufferSnapshot`. `None` in
/// hello-world mode or before the first snapshot arrives in
/// attach mode.
loro_doc: Option<loro::LoroDoc>,
}
impl ApplicationHandler for App {
impl ApplicationHandler<AppEvent> for App {
fn resumed(&mut self, event_loop: &ActiveEventLoop) {
if self.state.is_some() {
// `resumed` can fire more than once on platforms that
// suspend/restore (e.g. mobile). The hello-world doesn't
// reinitialize on resume; first call wins.
return;
}
self.state = Some(State::new(event_loop));
let initial_text = match &self.mode {
Mode::HelloWorld => HELLO_TEXT,
Mode::Attach { .. } => "(connecting...)",
};
self.state = Some(State::new(event_loop, initial_text));
// In attach mode, kick off the connection now that the event
// loop is running and a proxy is available. Failure logs and
// leaves the window showing its `(connecting...)` placeholder
// — better UX than killing the window during dev.
if let Mode::Attach { socket } = self.mode.clone() {
let proxy = self.proxy.take().expect("proxy taken twice");
match attach::connect(&socket, proxy) {
Ok(client) => {
self.attach_client = Some(client);
}
Err(e) => {
eprintln!("pmacs-gpu: attach failed: {e}");
if let Some(state) = self.state.as_mut() {
state.set_text("(attach failed; see stderr)");
}
}
}
}
}
fn window_event(&mut self, event_loop: &ActiveEventLoop, _id: WindowId, event: WindowEvent) {
@ -119,15 +225,28 @@ impl ApplicationHandler for App {
_ => {}
}
}
fn user_event(&mut self, _event_loop: &ActiveEventLoop, event: AppEvent) {
let Some(state) = self.state.as_mut() else {
return;
};
match event {
AppEvent::Attach(AttachEvent::Message(msg)) => state.apply_attach_message(*msg),
AppEvent::Attach(AttachEvent::Disconnected(reason)) => {
eprintln!("pmacs-gpu: daemon disconnected ({reason})");
state.set_text("(daemon disconnected)");
}
}
}
}
impl State {
fn new(event_loop: &ActiveEventLoop) -> Self {
fn new(event_loop: &ActiveEventLoop, initial_text: &str) -> Self {
let window = Arc::new(
event_loop
.create_window(
Window::default_attributes()
.with_title("pmacs-gpu hello-world")
.with_title("pmacs-gpu")
.with_inner_size(winit::dpi::LogicalSize::new(
f64::from(INITIAL_WIDTH),
f64::from(INITIAL_HEIGHT),
@ -136,22 +255,16 @@ impl State {
.expect("create window"),
);
// wgpu instance + surface.
let instance = wgpu::Instance::new(wgpu::InstanceDescriptor::new_without_display_handle());
let surface = instance
.create_surface(window.clone())
.expect("create surface");
// Pick an adapter that supports our surface. Power preference
// = LowPower because the hello-world has no GPU appetite;
// saves laptop battery during development.
let adapter = pollster::block_on(instance.request_adapter(&wgpu::RequestAdapterOptions {
power_preference: wgpu::PowerPreference::LowPower,
compatible_surface: Some(&surface),
force_fallback_adapter: false,
}))
.expect("request_adapter");
let (device, queue) = pollster::block_on(adapter.request_device(&wgpu::DeviceDescriptor {
label: Some("pmacs-gpu device"),
required_features: wgpu::Features::empty(),
@ -160,10 +273,6 @@ impl State {
}))
.expect("request_device");
// Configure the surface. Pick the first format the surface
// and adapter both like; glyphon handles colorspace conversion
// internally, so sRGB vs UNORM is the renderer's concern, not
// ours at this layer.
let inner_size = window.inner_size();
let surface_caps = surface.get_capabilities(&adapter);
let surface_format = surface_caps
@ -184,9 +293,6 @@ impl State {
};
surface.configure(&device, &config);
// glyphon plumbing — `FontSystem` owns the font database and
// shaper state; we register the bundled JetBrains Mono before
// anything tries to shape with it.
let mut font_system = FontSystem::new();
font_system.db_mut().load_font_data(JETBRAINS_MONO.to_vec());
let swash_cache = SwashCache::new();
@ -203,12 +309,11 @@ impl State {
let text_renderer =
TextRenderer::new(&mut atlas, &device, MultisampleState::default(), None);
// Shape "hello, pmacs" with JetBrains Mono at a comfortable
// hello-world size. cosmic-text 0.18 (glyphon's pinned
// version) threads `&mut FontSystem` through every Buffer
// mutator that needs to re-shape; the 5th `None` on `set_text`
// is the optional `Align` (we let cosmic-text default it).
let mut buffer = Buffer::new(&mut font_system, Metrics::new(48.0, 56.0));
// Smaller font in attach mode (file contents tend to be more
// than one line); larger only fits "hello, pmacs"-shaped
// strings. Picked metrics that look reasonable for code at
// 800px wide.
let mut buffer = Buffer::new(&mut font_system, Metrics::new(16.0, 22.0));
buffer.set_size(
&mut font_system,
Some(config.width as f32),
@ -216,7 +321,7 @@ impl State {
);
buffer.set_text(
&mut font_system,
HELLO_TEXT,
initial_text,
&Attrs::new().family(Family::Name("JetBrains Mono")),
Shaping::Advanced,
None,
@ -235,12 +340,73 @@ impl State {
atlas,
text_renderer,
buffer,
current_text: initial_text.to_owned(),
loro_doc: None,
}
}
/// Replace the rendered text with `text` and request a redraw.
/// No-op when `text` is byte-identical to the current rendering
/// (avoids the re-shape cost when an unchanged buffer ticks).
fn set_text(&mut self, text: &str) {
if self.current_text == text {
return;
}
self.current_text.clear();
self.current_text.push_str(text);
self.buffer.set_text(
&mut self.font_system,
&self.current_text,
&Attrs::new().family(Family::Name("JetBrains Mono")),
Shaping::Advanced,
None,
);
self.buffer.shape_until_scroll(&mut self.font_system, false);
self.window.request_redraw();
}
/// Apply one `InstanceMessage` to the local replica. Session 3
/// handles the two variants that matter for rope reconstruction:
/// `BufferSnapshot` (bootstrap) and `CrdtOp` (live updates). Every
/// other variant is ignored (logged at debug) — the `SemanticFrame`
/// family will be consumed in later sessions.
fn apply_attach_message(&mut self, msg: InstanceMessage) {
match msg {
InstanceMessage::BufferSnapshot { crdt_snapshot, .. } => {
let doc = loro::LoroDoc::new();
if let Err(e) = doc.import(&crdt_snapshot) {
eprintln!("pmacs-gpu: BufferSnapshot import failed: {e:?}");
return;
}
let text = doc.get_text(LORO_TEXT_CONTAINER).to_string();
self.loro_doc = Some(doc);
self.set_text(&text);
}
InstanceMessage::CrdtOp { op, .. } => {
let Some(doc) = self.loro_doc.as_ref() else {
// No snapshot yet — drop. The daemon's send order
// is snapshot-then-ops, so this case is rare
// (mid-attach race only). When the snapshot lands
// it'll have the ops baked in anyway.
return;
};
if let Err(e) = doc.import(&op.bytes) {
eprintln!("pmacs-gpu: CrdtOp import failed: {e:?}");
return;
}
let text = doc.get_text(LORO_TEXT_CONTAINER).to_string();
self.set_text(&text);
}
_ => {
// Other variants (Cursor, CellDelta, semantic frame
// family, presence, goodbye, etc.) are ignored in
// session 3. Goodbye in particular surfaces via the
// reader thread's clean-EOF path as a Disconnected
// event — not handled here.
}
}
}
/// Reconfigure surface + glyphon viewport on window-size change.
/// The shaped text buffer also gets a new max-size so wrap and
/// scroll align with the new viewport.
fn resize(&mut self, width: u32, height: u32) {
self.config.width = width;
self.config.height = height;
@ -255,16 +421,7 @@ impl State {
self.window.request_redraw();
}
/// One frame: clear the surface to `BG`, render the text buffer,
/// present. Acquisition failures cause a re-configure and skip
/// the frame (a typical recovery for transient surface losses).
fn render(&mut self) {
// wgpu 29 collapses success/error into a single enum
// (`CurrentSurfaceTexture`), not `Result<SurfaceTexture, SurfaceError>`
// as earlier versions did. Lost / Outdated trigger a
// re-configure and skip the frame; Suboptimal is rendered
// through but flagged for the next configure cycle (we don't
// act on it in the hello-world).
let frame = match self.surface.get_current_texture() {
wgpu::CurrentSurfaceTexture::Success(frame)
| wgpu::CurrentSurfaceTexture::Suboptimal(frame) => frame,
@ -272,13 +429,7 @@ impl State {
self.surface.configure(&self.device, &self.config);
return;
}
wgpu::CurrentSurfaceTexture::Timeout | wgpu::CurrentSurfaceTexture::Occluded => {
// `Timeout`: transient acquisition stall — drop this
// frame, try again next redraw.
// `Occluded`: window minimized / behind another
// window — skip the frame, save the GPU work.
return;
}
wgpu::CurrentSurfaceTexture::Timeout | wgpu::CurrentSurfaceTexture::Occluded => return,
wgpu::CurrentSurfaceTexture::Validation => {
eprintln!("surface acquisition raised a validation error");
return;
@ -297,16 +448,12 @@ impl State {
&self.viewport,
[TextArea {
buffer: &self.buffer,
left: 24.0,
top: 60.0,
left: 16.0,
top: 16.0,
scale: 1.0,
bounds: TextBounds {
left: 0,
top: 0,
// Surface dimensions are u32 but `TextBounds`
// is i32; `cast_signed` keeps the bit pattern
// and is correct for typical window sizes well
// below 2^31.
right: self.config.width.cast_signed(),
bottom: self.config.height.cast_signed(),
},
@ -324,7 +471,7 @@ impl State {
});
{
let mut pass = encoder.begin_render_pass(&wgpu::RenderPassDescriptor {
label: Some("pmacs-gpu hello-world pass"),
label: Some("pmacs-gpu pass"),
color_attachments: &[Some(wgpu::RenderPassColorAttachment {
view: &view,
depth_slice: None,

2
pmacs-protocol/src/lib.rs
View File

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

329
pmacs-protocol/src/transport.rs Normal file
View File

@ -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
View File

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