From 39dcd417235a81ecea15f5eb9dc4455205cc35c8 Mon Sep 17 00:00:00 2001 From: Kevin McConnell Date: Wed, 23 Sep 2026 17:44:04 -0500 Subject: [PATCH 1/3] Encode the protocol as CBOR with skippable frames gliff hasn't shipped, so settle a wire format that later releases can extend without breaking each other. postcard is positional, so any new field or message broke every older peer. Messages are now CBOR via minicbor, with every variant and field numbered and encoded as arrays: unknown fields are ignored, absent optional ones decode as None, and names never reach the wire. The frame header carries the payload length too, so a reader skips a message it does not know, payload and all, and the writer refuses frames the reader would reject. The handshake exchanges feature lists; anything new goes only to a peer that listed its feature. The rules for future changes are at the top of gliff-proto's msg.rs. The protocol version starts over at 1. The server reads Hello's version before the rest, so a mismatch is always reported, and it refuses a client with no codec or chroma mode in common before setting anything up. Error codes 1 to 255 mean the ends cannot work together: the client says which end is too old and stops reconnecting. --- Cargo.lock | 135 +----- Cargo.toml | 2 +- crates/gliff-probe/src/main.rs | 8 +- crates/gliff-proto/Cargo.toml | 3 +- crates/gliff-proto/src/cbor.rs | 67 +++ crates/gliff-proto/src/clipboard.rs | 19 +- crates/gliff-proto/src/frame.rs | 25 +- crates/gliff-proto/src/lib.rs | 1 + crates/gliff-proto/src/msg.rs | 646 ++++++++++++++++++++++++++-- crates/gliff-server/src/session.rs | 69 +-- crates/gliff-server/src/writer.rs | 2 +- crates/gliff-transport/Cargo.toml | 3 +- crates/gliff-transport/src/lib.rs | 388 +++++++++++++++-- crates/gliff/Cargo.toml | 1 + crates/gliff/src/main.rs | 15 +- crates/gliff/src/net.rs | 42 +- docs/architecture.md | 14 +- 17 files changed, 1191 insertions(+), 249 deletions(-) create mode 100644 crates/gliff-proto/src/cbor.rs diff --git a/Cargo.lock b/Cargo.lock index 5574c4b0..04e4e37f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -82,15 +82,6 @@ dependencies = [ "libloading", ] -[[package]] -name = "atomic-polyfill" -version = "1.0.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8cf2bce30dfe09ef0bfaef228b9d414faaf7e563035494d7fe092dba54b300f4" -dependencies = [ - "critical-section", -] - [[package]] name = "autocfg" version = "1.5.1" @@ -144,12 +135,6 @@ dependencies = [ "syn 3.0.5", ] -[[package]] -name = "byteorder" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" - [[package]] name = "bytes" version = "1.12.1" @@ -278,15 +263,6 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" -[[package]] -name = "cobs" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fa961b519f0b462e3a3b4a34b64d119eeaca1d59af726fe450bbba07a9fc0a1" -dependencies = [ - "thiserror", -] - [[package]] name = "colorchoice" version = "1.0.5" @@ -311,12 +287,6 @@ dependencies = [ "cfg-if", ] -[[package]] -name = "critical-section" -version = "1.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" - [[package]] name = "crossbeam-deque" version = "0.8.8" @@ -413,18 +383,6 @@ version = "1.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34" -[[package]] -name = "embedded-io" -version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef1a6892d9eef45c8fa6b9e0086428a2cca8491aca8f787c534a3d6d0bcb3ced" - -[[package]] -name = "embedded-io" -version = "0.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" - [[package]] name = "equivalent" version = "1.0.2" @@ -746,6 +704,7 @@ dependencies = [ "hypr-ipc", "libadwaita", "serde", + "thiserror", "tokio", "toml", "tracing", @@ -779,10 +738,9 @@ dependencies = [ name = "gliff-proto" version = "0.1.0" dependencies = [ - "postcard", + "minicbor", "proptest", "rayon", - "serde", "thiserror", ] @@ -824,8 +782,7 @@ version = "0.1.0" dependencies = [ "bytes", "gliff-proto", - "postcard", - "serde", + "minicbor", "tempfile", "thiserror", "tokio", @@ -958,35 +915,12 @@ dependencies = [ "system-deps 7.0.8", ] -[[package]] -name = "hash32" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b0c35f58762feb77d74ebe43bdbc3210f09be9fe6742234d573bacc26ed92b67" -dependencies = [ - "byteorder", -] - [[package]] name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" -[[package]] -name = "heapless" -version = "0.7.17" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cdc6457c0eb62c71aac4bc17216026d8410337c4126773b9c5daba343f17964f" -dependencies = [ - "atomic-polyfill", - "hash32", - "rustc_version", - "serde", - "spin", - "stable_deref_trait", -] - [[package]] name = "heck" version = "0.5.0" @@ -1167,15 +1101,6 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" -[[package]] -name = "lock_api" -version = "0.4.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" -dependencies = [ - "scopeguard", -] - [[package]] name = "log" version = "0.4.34" @@ -1215,6 +1140,26 @@ dependencies = [ "autocfg", ] +[[package]] +name = "minicbor" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c12b4033ffaa92fbf9df03df38d19324f52bad130dd223f811734a8006dd2d69" +dependencies = [ + "minicbor-derive", +] + +[[package]] +name = "minicbor-derive" +version = "0.19.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "84f5ad8dfe10176465fe33f19ecfcf08783ba66630fdb40b2e4542547c1046f0" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "miniz_oxide" version = "0.8.9" @@ -1381,19 +1326,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "postcard" -version = "1.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6764c3b5dd454e283a30e6dfe78e9b31096d9e32036b5d1eaac7a6119ccb9a24" -dependencies = [ - "cobs", - "embedded-io 0.4.0", - "embedded-io 0.6.1", - "heapless", - "serde", -] - [[package]] name = "ppv-lite86" version = "0.2.21" @@ -1622,12 +1554,6 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e1cf6437eb19a8f4a6cc0f7dca544973b0b78843adbfeb3683d1a94a0024a294" -[[package]] -name = "scopeguard" -version = "1.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" - [[package]] name = "semver" version = "1.0.28" @@ -1739,21 +1665,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "spin" -version = "0.9.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" -dependencies = [ - "lock_api", -] - -[[package]] -name = "stable_deref_trait" -version = "1.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" - [[package]] name = "strsim" version = "0.11.1" diff --git a/Cargo.toml b/Cargo.toml index 62902cea..00eb8345 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -43,7 +43,7 @@ tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } serde = { version = "1", features = ["derive"] } serde_json = "1" -postcard = { version = "1", features = ["use-std"] } +minicbor = { version = "2", features = ["derive", "alloc"] } tokio = { version = "1", features = ["rt-multi-thread", "macros", "io-util", "net", "process", "sync", "time", "io-std", "fs"] } bytes = "1" clap = { version = "4", features = ["derive"] } diff --git a/crates/gliff-probe/src/main.rs b/crates/gliff-probe/src/main.rs index 3455ac50..23e1ae98 100644 --- a/crates/gliff-probe/src/main.rs +++ b/crates/gliff-probe/src/main.rs @@ -509,7 +509,7 @@ fn serve_test(node: &std::path::Path, addr: &str, frames: usize, video: VideoMod let (rd, wr) = tokio::io::split(stream); let mut reader = Framed::new(rd); let mut writer = Framed::new(wr); - let caps = ClientCaps { codecs: vec![Codec::H264], max_width: 1280, max_height: 720, chroma: vec![ChromaMode::Dual420, ChromaMode::Single420] }; + let caps = ClientCaps { codecs: vec![Codec::H264], max_width: 1280, max_height: 720, chroma: vec![ChromaMode::Dual420, ChromaMode::Single420], features: gliff_proto::features() }; writer.write_msg(&ClientMsg::Hello { version: gliff_proto::PROTOCOL_VERSION, keymap: String::new(), caps }).await?; let ack = reader.read_msg::().await?; let ServerMsg::HelloAck { session, outputs, .. } = ack else { bail!("expected HelloAck, got {ack:?}") }; @@ -783,7 +783,7 @@ fn stream_bench( let (rd, wr) = tokio::io::split(stream); let mut reader = Framed::new(rd); let mut writer = Framed::new(wr); - let caps = ClientCaps { codecs: vec![Codec::H264], max_width: 3840, max_height: 2160, chroma: vec![ChromaMode::Dual420, ChromaMode::Single420] }; + let caps = ClientCaps { codecs: vec![Codec::H264], max_width: 3840, max_height: 2160, chroma: vec![ChromaMode::Dual420, ChromaMode::Single420], features: gliff_proto::features() }; writer.write_msg(&ClientMsg::Hello { version: gliff_proto::PROTOCOL_VERSION, keymap: String::new(), caps }).await?; let ack = reader.read_msg::().await?; let ServerMsg::HelloAck { session, .. } = ack else { bail!("expected HelloAck, got {ack:?}") }; @@ -810,8 +810,8 @@ fn stream_bench( let mut csv_out = match csv { Some(p) => Some(std::io::BufWriter::new(std::fs::File::create(p)?)), None => None }; if let Some(c) = csv_out.as_mut() { writeln!(c, "t_ms,frame_id,key,bytes,latency_ms,decode_ms")?; } - // Reads happen on their own task: `read_msg` is not cancel-safe in - // a `select!`, and pings must go out while no frames arrive. + // Reads happen on their own task, so pings go out while no frames + // arrive. let (msg_tx, mut msg_rx) = tokio::sync::mpsc::unbounded_channel(); tokio::task::spawn_local(async move { loop { diff --git a/crates/gliff-proto/Cargo.toml b/crates/gliff-proto/Cargo.toml index 86ce8863..28a7327e 100644 --- a/crates/gliff-proto/Cargo.toml +++ b/crates/gliff-proto/Cargo.toml @@ -9,8 +9,7 @@ rust-version.workspace = true workspace = true [dependencies] -serde.workspace = true -postcard.workspace = true +minicbor.workspace = true thiserror.workspace = true rayon.workspace = true diff --git a/crates/gliff-proto/src/cbor.rs b/crates/gliff-proto/src/cbor.rs new file mode 100644 index 00000000..0045a136 --- /dev/null +++ b/crates/gliff-proto/src/cbor.rs @@ -0,0 +1,67 @@ +//! Decoding helpers for values a newer peer may extend. + +use minicbor::data::Type; +use minicbor::decode::{Decoder, Error}; +use minicbor::Decode; + +/// Decode an optional enum field, reading a value this build does not know +/// as `None`. minicbor's own `Option` decoding mishandles an unknown +/// `index_only` value and misreads the fields after it, so every optional +/// enum field uses this with `nil = "crate::cbor::none"`. +pub fn optional<'b, Ctx, T: Decode<'b, Ctx>>( + d: &mut Decoder<'b>, + ctx: &mut Ctx, +) -> Result, Error> { + if d.datatype()? == Type::Null { + d.skip()?; + return Ok(None); + } + let start = d.position(); + match T::decode(d, ctx) { + Ok(v) => Ok(Some(v)), + Err(e) if e.is_unknown_variant() => { + d.set_position(start); + d.skip()?; + Ok(None) + } + Err(e) => Err(e), + } +} + +/// The value of an absent [`optional`] field. +pub fn none() -> Option> { + Some(None) +} + +/// Decode a list, dropping entries whose enum variant this build does not +/// know. Used where a peer describes itself before any features are agreed, +/// so a newer peer's additions cannot make the whole message fail. +pub fn known<'b, Ctx, T: Decode<'b, Ctx>>( + d: &mut Decoder<'b>, + ctx: &mut Ctx, +) -> Result, Error> { + let len = d.array()?; + let mut out = Vec::new(); + let mut seen = 0u64; + loop { + match len { + Some(n) if seen == n => break, + None if d.datatype()? == Type::Break => { + d.skip()?; + break; + } + _ => {} + } + let start = d.position(); + match T::decode(d, ctx) { + Ok(v) => out.push(v), + Err(e) if e.is_unknown_variant() => { + d.set_position(start); + d.skip()?; + } + Err(e) => return Err(e), + } + seen += 1; + } + Ok(out) +} diff --git a/crates/gliff-proto/src/clipboard.rs b/crates/gliff-proto/src/clipboard.rs index 38a1f045..9d348dcb 100644 --- a/crates/gliff-proto/src/clipboard.rs +++ b/crates/gliff-proto/src/clipboard.rs @@ -19,7 +19,7 @@ use std::collections::VecDeque; use std::path::{Component, Path, PathBuf}; -use serde::{Deserialize, Serialize}; +use minicbor::{Decode, Encode}; /// Largest `data_len` of one `Data` message. pub const CHUNK: usize = 256 * 1024; @@ -59,23 +59,30 @@ const NON_DATA_TARGETS: &[&str] = &[ "INSERT_SELECTION", ]; -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)] pub struct ClipboardFile { /// Path relative to the copied item, `/`-separated; a top-level item is /// its own name. + #[n(0)] pub path: String, + #[n(1)] pub size: u64, + #[n(2)] pub dir: bool, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)] pub enum ClipboardItem { - Mime(String), + #[n(0)] + Mime(#[n(0)] String), /// Index into the offer's `files`. - File(u32), + #[n(1)] + File(#[n(0)] u32), } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +/// The clipboard messages both peers share. Never encoded itself: each end +/// carries these as its own `ClientMsg`/`ServerMsg` variants. +#[derive(Debug, Clone, PartialEq, Eq)] pub enum ClipboardMsg { /// The sender's selection changed to something it can serve. Empty lists /// mean the selection was cleared or holds nothing forwardable. `serial` diff --git a/crates/gliff-proto/src/frame.rs b/crates/gliff-proto/src/frame.rs index 2ee3d027..b5c6f2e7 100644 --- a/crates/gliff-proto/src/frame.rs +++ b/crates/gliff-proto/src/frame.rs @@ -1,14 +1,25 @@ -//! Wire framing: a length-prefixed postcard message, optionally followed by raw -//! payload bytes whose lengths the message declares. +//! Wire framing: a header, a CBOR message, then raw payload bytes whose +//! lengths the message declares. //! //! Layout on the wire, little-endian: //! ```text -//! u32 body_len | body_len bytes of postcard message | payload bytes (concatenated) +//! u32 body_len | u32 payload_len | body_len bytes of CBOR message | payload_len bytes //! ``` //! The payload bytes are the video `data`/`aux`, cursor `argb`, or clipboard -//! `data` chunk. They are kept out of the postcard body so the encoder's -//! output goes straight to the socket with `write_vectored`, and the reader -//! hands the slices to the decoder without an extra copy. +//! `data` chunk. They are kept out of the CBOR body so the encoder's output +//! goes straight to the socket with `write_vectored`, and the reader hands +//! the slices to the decoder without an extra copy. The header carries the +//! payload length too, so a reader can skip a message it does not know. +//! +//! This layout never changes, whatever the protocol version. + +pub const HEADER_LEN: usize = 8; -/// Largest postcard body we will read; guards against a bad length prefix. +/// Largest CBOR body either end sends or accepts. pub const MAX_FRAME_BODY: usize = 1 << 20; + +/// Largest single payload a message may declare (a keyframe). +pub const MAX_PAYLOAD: u32 = 64 * 1024 * 1024; + +/// Largest total payload of one frame: a keyframe's main and aux streams. +pub const MAX_FRAME_PAYLOAD: u32 = 2 * MAX_PAYLOAD; diff --git a/crates/gliff-proto/src/lib.rs b/crates/gliff-proto/src/lib.rs index 1e8db315..6d8a07d9 100644 --- a/crates/gliff-proto/src/lib.rs +++ b/crates/gliff-proto/src/lib.rs @@ -4,6 +4,7 @@ //! //! No I/O here. `gliff-transport` moves bytes; this crate defines their shape. +pub mod cbor; pub mod chroma; pub mod clipboard; pub mod color; diff --git a/crates/gliff-proto/src/msg.rs b/crates/gliff-proto/src/msg.rs index 4daf28ea..1b57f56c 100644 --- a/crates/gliff-proto/src/msg.rs +++ b/crates/gliff-proto/src/msg.rs @@ -1,25 +1,87 @@ -//! Client and server messages. Keep variants additive; never renumber. +//! Client and server messages, encoded as CBOR; see [`crate::frame`] for +//! how they travel. Every variant and field is numbered with `#[n]`, and a +//! message goes on the wire as `[variant, [field 0, field 1, ...]]`, so the +//! numbers are positions and names never appear. +//! +//! Peers from different builds stay compatible when changes follow these +//! rules: +//! - Never renumber or reuse a variant or field number. +//! - A new field takes the next number and is an `Option`, so an older peer +//! ignores it and a peer that lacks it decodes `None`. An optional enum +//! field decodes with [`crate::cbor::optional`]. +//! - A required field that has shipped must be sent forever: removing it, or +//! making it optional and sending `None`, breaks older peers. +//! - A new message, or a new value of an existing enum, goes only to a peer +//! that listed the feature introducing it in the handshake (see +//! [`FEATURES`]). A peer skips a message it does not know, but an unknown +//! enum value inside a known message fails that message. +//! - `Hello` arrives before any feature is agreed, so its enum lists drop +//! values this build does not know. +//! - The frame header, the head of `Hello` (`[0, [version, ...]]`) and the +//! first two fields of `ServerMsg::Error` never change, so a version +//! mismatch can always be reported; see [`Greeting`]. `Error` may gain +//! more optional fields. +//! +//! A breaking change that these rules cannot express bumps +//! [`PROTOCOL_VERSION`]. -use serde::{Deserialize, Serialize}; +use minicbor::decode::{Decoder, Error}; +use minicbor::{Decode, Encode}; pub use crate::clipboard::{ClipboardFile, ClipboardItem, ClipboardMsg}; -pub const PROTOCOL_VERSION: u16 = 5; +pub const PROTOCOL_VERSION: u16 = 1; -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +/// Optional behaviours this build supports, exchanged in the handshake. A +/// behaviour is used only when both ends list it. +pub const FEATURES: &[&str] = &[]; + +pub fn features() -> Vec { + FEATURES.iter().map(|f| f.to_string()).collect() +} + +/// `ServerMsg::Error` codes. Codes 1 to 255 mean the two ends cannot work +/// together, so reconnecting would fail the same way; a new code in that +/// range is fatal even to clients that predate it. +pub const ERROR_VERSION: u16 = 1; +pub const ERROR_NO_CHROMA: u16 = 2; +pub const ERROR_NO_CODEC: u16 = 3; + +pub fn is_incompatible(code: u16) -> bool { + (1..256).contains(&code) +} + +/// Which end of a protocol version mismatch needs updating. +pub fn version_mismatch(client: u16, server: u16) -> &'static str { + if client < server { + "Client version is too old" + } else { + "Server version is too old" + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Encode, Decode)] +#[cbor(index_only)] pub enum Codec { /// Uncompressed BGRA, phase 1 / localhost only. + #[n(0)] RawBgra, + #[n(1)] H264, + #[n(2)] H265, + #[n(3)] Av1, } /// Which pipeline encodes or decodes on one end, reported for the stats /// display. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Encode, Decode)] +#[cbor(index_only)] pub enum VideoPipeline { + #[n(0)] Gpu, + #[n(1)] Cpu, } @@ -32,221 +94,410 @@ impl VideoPipeline { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Encode, Decode)] +#[cbor(index_only)] pub enum ChromaMode { /// Single stream, encoder produced 4:4:4. + #[n(0)] Native444, /// AVC444-style: main 4:2:0 stream plus an auxiliary 4:2:0 stream carrying /// the chroma the main stream dropped. + #[n(1)] Dual420, /// Low-bandwidth fallback only. + #[n(2)] Single420, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Encode, Decode)] +#[cbor(index_only)] pub enum Axis { + #[n(0)] Vertical, + #[n(1)] Horizontal, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Encode, Decode)] pub struct Rect { + #[n(0)] pub x: i32, + #[n(1)] pub y: i32, + #[n(2)] pub width: i32, + #[n(3)] pub height: i32, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)] pub struct ClientCaps { + #[n(0)] + #[cbor(decode_with = "crate::cbor::known")] pub codecs: Vec, + #[n(1)] pub max_width: u32, + #[n(2)] pub max_height: u32, /// Chroma modes the client can decode, preferred first. + #[n(3)] + #[cbor(decode_with = "crate::cbor::known")] pub chroma: Vec, + /// The client's [`FEATURES`]. + #[n(4)] + pub features: Vec, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)] pub struct OutputInfo { + #[n(0)] pub name: String, + #[n(1)] pub width: u32, + #[n(2)] pub height: u32, + #[n(3)] pub scale_milli: u32, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)] pub struct SessionInfo { + #[n(0)] pub headless: bool, + #[n(1)] pub output: String, } -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Encode, Decode)] pub enum ClientMsg { + #[n(0)] Hello { + #[n(0)] version: u16, + /// XKB keymap text, or empty for the server's default. A client + /// that cannot produce XKB text leaves it empty; a later optional + /// field can carry layout names, which older servers ignore. + #[n(1)] keymap: String, + #[n(2)] caps: ClientCaps, }, + #[n(1)] Resize { + #[n(0)] width: u32, + #[n(1)] height: u32, + #[n(2)] scale: f32, }, + #[n(2)] Key { + #[n(0)] keycode: u32, + #[n(1)] pressed: bool, }, + #[n(3)] PointerMotion { + #[n(0)] x: f64, + #[n(1)] y: f64, }, + #[n(4)] PointerButton { + #[n(0)] button: u32, + #[n(1)] pressed: bool, }, + #[n(5)] PointerAxis { + #[n(0)] axis: Axis, + #[n(1)] value: f64, + #[n(2)] discrete: Option, + #[n(3)] stop: bool, }, + #[n(6)] FrameAck { + #[n(0)] frame_id: u64, + #[n(1)] decoded_at_ms: u64, }, + #[n(7)] RequestKeyframe, /// The local selection changed; see [`ClipboardMsg::Offer`]. + #[n(8)] ClipboardOffer { + #[n(0)] serial: u32, + #[n(1)] mime_types: Vec, + #[n(2)] files: Vec, }, + #[n(9)] ClipboardRequest { + #[n(0)] id: u32, + #[n(1)] serial: u32, + #[n(2)] item: ClipboardItem, }, /// `data_len` payload bytes follow the header. + #[n(10)] ClipboardData { + #[n(0)] id: u32, + #[n(1)] offset: u64, + #[n(2)] data_len: u32, + #[n(3)] done: bool, }, + #[n(11)] Ping { + #[n(0)] t: u64, }, + #[n(12)] Bye, + #[n(13)] ClipboardAck { + #[n(0)] id: u32, + #[n(1)] received: u64, }, + #[n(14)] ClipboardAbort { + #[n(0)] id: u32, }, - /// The local keymap changed since `Hello`; same format as its `keymap`. + /// The local keymap changed since `Hello`; XKB text as in its `keymap`. + #[n(15)] Keymap { + #[n(0)] keymap: String, }, /// Answer to `ServerMsg::Ping`, sent before anything else so the server /// can seed its round-trip estimate ahead of the first frame ack. + #[n(16)] Pong { + #[n(0)] t: u64, }, } -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Encode, Decode)] pub enum ServerMsg { + #[n(0)] HelloAck { + #[n(0)] version: u16, + #[n(1)] session: SessionInfo, + #[n(2)] outputs: Vec, + /// The server's [`FEATURES`]. + #[n(3)] + features: Vec, }, + #[n(1)] StreamConfig { + #[n(0)] codec: Codec, + #[n(1)] chroma: ChromaMode, /// Which pipeline encodes on the server, for the client's stats. - pipeline: VideoPipeline, + #[n(2)] + #[cbor(decode_with = "crate::cbor::optional", nil = "crate::cbor::none")] + pipeline: Option, /// Stream size in physical pixels. + #[n(3)] width: u32, + #[n(4)] height: u32, /// Effective stream scale x1000: the remote's logical size is /// `width / scale`, and pointer coordinates are sent in that logical /// space. It folds in any downscale of the stream, so it stays the /// number stream pixels are divided by. + #[n(5)] scale_milli: u32, + #[n(6)] + #[cbor(with = "minicbor::bytes")] extradata: Vec, + #[n(7)] + #[cbor(with = "minicbor::bytes")] aux_extradata: Option>, /// The full-quality fit size in physical pixels: the area the client /// should draw the stream into. Equal to the stream size unless the /// server sends a reduced-resolution stream. + #[n(8)] view_width: u32, + #[n(9)] view_height: u32, /// The frame-rate ceiling in force, for the client's stats display. + #[n(10)] fps_cap: u32, }, + #[n(2)] VideoFrame { + #[n(0)] frame_id: u64, + #[n(1)] pts_us: u64, + #[n(2)] keyframe: bool, + #[n(3)] damage: Vec, + #[n(4)] data_len: u32, /// `aux_len > 0` iff `chroma == Dual420`; bytes follow the header. + #[n(5)] aux_len: u32, }, + #[n(3)] CursorShape { + #[n(0)] id: u32, + #[n(1)] width: u32, + #[n(2)] height: u32, + #[n(3)] hot_x: i32, + #[n(4)] hot_y: i32, + #[n(5)] argb_len: u32, }, + #[n(4)] CursorPos { + #[n(0)] x: f64, + #[n(1)] y: f64, + #[n(2)] shape_id: u32, + #[n(3)] visible: bool, }, + #[n(5)] ClipboardOffer { + #[n(0)] serial: u32, + #[n(1)] mime_types: Vec, + #[n(2)] files: Vec, }, + #[n(6)] ClipboardRequest { + #[n(0)] id: u32, + #[n(1)] serial: u32, + #[n(2)] item: ClipboardItem, }, + #[n(7)] ClipboardData { + #[n(0)] id: u32, + #[n(1)] offset: u64, + #[n(2)] data_len: u32, + #[n(3)] done: bool, }, + #[n(8)] Pong { + #[n(0)] t: u64, + #[n(1)] server_now_ms: u64, }, + #[n(9)] Error { + #[n(0)] code: u16, + #[n(1)] message: String, + /// The server's [`PROTOCOL_VERSION`]. + #[n(2)] + version: Option, }, + #[n(10)] ClipboardAck { + #[n(0)] id: u32, + #[n(1)] received: u64, }, + #[n(11)] ClipboardAbort { + #[n(0)] id: u32, }, /// Sent right after HelloAck; the client answers with `ClientMsg::Pong` /// ahead of anything else, seeding the server's round-trip estimate /// before the first frame is acked. + #[n(12)] Ping { + #[n(0)] t: u64, }, } +impl ServerMsg { + pub fn error(code: u16, message: impl Into) -> Self { + Self::Error { + code, + message: message.into(), + version: Some(PROTOCOL_VERSION), + } + } +} + +/// The client's first message as the server reads it. The version is read +/// before anything else, so a `Hello` from another protocol version is +/// recognised even when the rest of it has changed shape. +#[derive(Debug, Clone, PartialEq)] +pub enum Greeting { + Hello { keymap: String, caps: ClientCaps }, + OtherVersion(u16), +} + +impl<'b, C> Decode<'b, C> for Greeting { + fn decode(d: &mut Decoder<'b>, ctx: &mut C) -> Result { + let start = d.position(); + d.array()?; + if d.u32()? != 0 { + return Err(Error::message("expected Hello")); + } + d.array()?; + let version = d.u16()?; + if version != PROTOCOL_VERSION { + return Ok(Self::OtherVersion(version)); + } + d.set_position(start); + match ClientMsg::decode(d, ctx)? { + ClientMsg::Hello { keymap, caps, .. } => Ok(Self::Hello { keymap, caps }), + _ => Err(Error::message("expected Hello")), + } + } +} + /// The clipboard variants have the same shape in both directions so the /// transfer logic can be shared; these convert to and from [`ClipboardMsg`]. macro_rules! clipboard_bridge { @@ -330,13 +581,14 @@ mod tests { fn round_trips_messages() { let msgs = vec![ ClientMsg::Hello { - version: 1, + version: PROTOCOL_VERSION, keymap: "xkb".into(), caps: ClientCaps { codecs: vec![Codec::H264], max_width: 3840, max_height: 2160, chroma: vec![ChromaMode::Dual420, ChromaMode::Single420], + features: vec!["x".into()], }, }, ClientMsg::Key { @@ -358,8 +610,8 @@ mod tests { }, ]; for m in msgs { - let bytes = postcard::to_stdvec(&m).unwrap(); - let back: ClientMsg = postcard::from_bytes(&bytes).unwrap(); + let bytes = minicbor::to_vec(&m).unwrap(); + let back: ClientMsg = minicbor::decode(&bytes).unwrap(); assert_eq!(m, back); } let s = ServerMsg::VideoFrame { @@ -375,8 +627,8 @@ mod tests { data_len: 100, aux_len: 50, }; - let bytes = postcard::to_stdvec(&s).unwrap(); - assert_eq!(postcard::from_bytes::(&bytes).unwrap(), s); + let bytes = minicbor::to_vec(&s).unwrap(); + assert_eq!(minicbor::decode::(&bytes).unwrap(), s); } #[test] @@ -410,14 +662,360 @@ mod tests { ]; for m in msgs { let c = ClientMsg::from(m.clone()); - let bytes = postcard::to_stdvec(&c).unwrap(); - let back: ClientMsg = postcard::from_bytes(&bytes).unwrap(); + let bytes = minicbor::to_vec(&c).unwrap(); + let back: ClientMsg = minicbor::decode(&bytes).unwrap(); assert_eq!(back.into_clipboard(), Ok(m.clone())); let s = ServerMsg::from(m.clone()); - let bytes = postcard::to_stdvec(&s).unwrap(); - let back: ServerMsg = postcard::from_bytes(&bytes).unwrap(); + let bytes = minicbor::to_vec(&s).unwrap(); + let back: ServerMsg = minicbor::decode(&bytes).unwrap(); assert_eq!(back.into_clipboard(), Ok(m)); } assert_eq!(ClientMsg::Bye.into_clipboard(), Err(ClientMsg::Bye)); } + + fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() + } + + /// These bytes are the protocol: a change here breaks peers from other + /// builds, so it must follow the rules at the top of this file. + #[test] + fn encodings_are_stable() { + let client = [ + ( + ClientMsg::Hello { + version: 1, + keymap: "k".into(), + caps: ClientCaps { + codecs: vec![Codec::H264], + max_width: 3840, + max_height: 2160, + chroma: vec![ChromaMode::Dual420, ChromaMode::Single420], + features: vec!["f".into()], + }, + }, + "82008301616b858101190f00190870820102816166", + ), + ( + ClientMsg::Key { + keycode: 30, + pressed: true, + }, + "820282181ef5", + ), + ( + ClientMsg::PointerMotion { x: 12.5, y: 7.25 }, + "820382fb4029000000000000fb401d000000000000", + ), + ( + ClientMsg::FrameAck { + frame_id: 42, + decoded_at_ms: 1000, + }, + "820682182a1903e8", + ), + (ClientMsg::RequestKeyframe, "820780"), + ]; + for (msg, bytes) in client { + assert_eq!(hex(&minicbor::to_vec(&msg).unwrap()), bytes, "{msg:?}"); + } + let server = [ + ( + ServerMsg::HelloAck { + version: 1, + session: SessionInfo { + headless: true, + output: "o".into(), + }, + outputs: vec![OutputInfo { + name: "o".into(), + width: 1920, + height: 1080, + scale_milli: 1000, + }], + features: vec![], + }, + "8200840182f5616f8184616f1907801904381903e880", + ), + ( + ServerMsg::StreamConfig { + codec: Codec::H264, + chroma: ChromaMode::Dual420, + pipeline: Some(VideoPipeline::Gpu), + width: 1920, + height: 1080, + scale_milli: 1000, + extradata: vec![], + aux_extradata: None, + view_width: 1920, + view_height: 1080, + fps_cap: 60, + }, + "82018b0101001907801904381903e840f6190780190438183c", + ), + ( + ServerMsg::VideoFrame { + frame_id: 1, + pts_us: 2, + keyframe: true, + damage: vec![Rect { + x: 0, + y: 0, + width: 10, + height: 10, + }], + data_len: 100, + aux_len: 50, + }, + "8202860102f5818400000a0a18641832", + ), + (ServerMsg::error(ERROR_VERSION, "m"), "82098301616d01"), + ]; + for (msg, bytes) in server { + assert_eq!(hex(&minicbor::to_vec(&msg).unwrap()), bytes, "{msg:?}"); + } + } + + /// A newer client's `Hello`: an extra field in the message and in its + /// caps, and a codec and chroma mode this build does not know. + #[derive(Encode)] + enum NewerClientMsg { + #[n(0)] + Hello { + #[n(0)] + version: u16, + #[n(1)] + keymap: String, + #[n(2)] + caps: NewerCaps, + #[n(3)] + layout: String, + }, + } + + #[derive(Encode)] + struct NewerCaps { + #[n(0)] + codecs: Vec, + #[n(1)] + max_width: u32, + #[n(2)] + max_height: u32, + #[n(3)] + chroma: Vec, + #[n(4)] + features: Vec, + #[n(5)] + max_fps: u32, + } + + #[derive(Encode)] + #[cbor(index_only)] + enum NewerCodec { + #[n(1)] + H264OrDual420, + #[n(9)] + Future, + } + + #[test] + fn hello_from_a_newer_client_decodes() { + let newer = NewerClientMsg::Hello { + version: 1, + keymap: "k".into(), + caps: NewerCaps { + codecs: vec![NewerCodec::Future, NewerCodec::H264OrDual420], + max_width: 3840, + max_height: 2160, + chroma: vec![NewerCodec::H264OrDual420, NewerCodec::Future], + features: vec!["future".into()], + max_fps: 120, + }, + layout: "us".into(), + }; + let bytes = minicbor::to_vec(&newer).unwrap(); + assert_eq!( + minicbor::decode::(&bytes).unwrap(), + ClientMsg::Hello { + version: 1, + keymap: "k".into(), + caps: ClientCaps { + codecs: vec![Codec::H264], + max_width: 3840, + max_height: 2160, + chroma: vec![ChromaMode::Dual420], + features: vec!["future".into()], + }, + } + ); + } + + #[test] + fn greeting_reads_the_version_of_any_hello() { + let hello = ClientMsg::Hello { + version: PROTOCOL_VERSION, + keymap: "k".into(), + caps: ClientCaps { + codecs: vec![Codec::H264], + max_width: 1, + max_height: 1, + chroma: vec![ChromaMode::Dual420], + features: vec![], + }, + }; + let ClientMsg::Hello { keymap, caps, .. } = hello.clone() else { + unreachable!() + }; + assert_eq!( + minicbor::decode::(&minicbor::to_vec(&hello).unwrap()).unwrap(), + Greeting::Hello { keymap, caps } + ); + + /// A future protocol version whose `Hello` has a different shape. + #[derive(Encode)] + enum Future { + #[n(0)] + Hello { + #[n(0)] + version: u16, + #[n(1)] + caps: u64, + }, + } + let future = minicbor::to_vec(Future::Hello { + version: PROTOCOL_VERSION + 1, + caps: 7, + }) + .unwrap(); + assert_eq!( + minicbor::decode::(&future).unwrap(), + Greeting::OtherVersion(PROTOCOL_VERSION + 1) + ); + } + + /// A newer server's `StreamConfig`: a pipeline this build does not know, + /// and an extra field after the ones it does. + #[derive(Encode)] + enum NewerServerMsg { + #[n(1)] + StreamConfig { + #[n(0)] + codec: Codec, + #[n(1)] + chroma: ChromaMode, + #[n(2)] + pipeline: NewerPipeline, + #[n(3)] + width: u32, + #[n(4)] + height: u32, + #[n(5)] + scale_milli: u32, + #[n(6)] + #[cbor(with = "minicbor::bytes")] + extradata: Vec, + #[n(7)] + #[cbor(with = "minicbor::bytes")] + aux_extradata: Option>, + #[n(8)] + view_width: u32, + #[n(9)] + view_height: u32, + #[n(10)] + fps_cap: u32, + #[n(11)] + hdr: bool, + }, + } + + #[derive(Encode)] + #[cbor(index_only)] + enum NewerPipeline { + #[n(9)] + Npu, + } + + #[test] + fn stream_config_from_a_newer_server_decodes() { + let newer = NewerServerMsg::StreamConfig { + codec: Codec::H264, + chroma: ChromaMode::Dual420, + pipeline: NewerPipeline::Npu, + width: 1920, + height: 1080, + scale_milli: 1000, + extradata: vec![], + aux_extradata: None, + view_width: 1920, + view_height: 1080, + fps_cap: 60, + hdr: true, + }; + assert_eq!( + minicbor::decode::(&minicbor::to_vec(&newer).unwrap()).unwrap(), + ServerMsg::StreamConfig { + codec: Codec::H264, + chroma: ChromaMode::Dual420, + pipeline: None, + width: 1920, + height: 1080, + scale_milli: 1000, + extradata: vec![], + aux_extradata: None, + view_width: 1920, + view_height: 1080, + fps_cap: 60, + } + ); + } + + #[test] + fn known_reads_indefinite_lists_and_keeps_other_errors() { + let mut d = minicbor::Decoder::new(&[0x9f, 0x01, 0x09, 0x02, 0xff]); + assert_eq!( + crate::cbor::known::<(), Codec>(&mut d, &mut ()).unwrap(), + vec![Codec::H264, Codec::H265] + ); + let mut d = minicbor::Decoder::new(&[0x81, 0x61, 0x78]); + assert!(crate::cbor::known::<(), Codec>(&mut d, &mut ()).is_err()); + } + + /// The shape every new field takes: an optional enum appended last. + #[derive(Debug, PartialEq, Encode, Decode)] + struct Grown { + #[n(0)] + width: u32, + #[n(1)] + #[cbor(decode_with = "crate::cbor::optional", nil = "crate::cbor::none")] + pipeline: Option, + } + + #[derive(Debug, PartialEq, Encode, Decode)] + struct Original { + #[n(0)] + width: u32, + } + + #[test] + fn an_appended_optional_enum_is_invisible_when_absent() { + let grown = Grown { + width: 7, + pipeline: None, + }; + let bytes = minicbor::to_vec(&grown).unwrap(); + assert_eq!(bytes, minicbor::to_vec(Original { width: 7 }).unwrap()); + assert_eq!(minicbor::decode::(&bytes).unwrap(), grown); + let with_null = [0x82, 0x07, 0xf6]; + assert_eq!(minicbor::decode::(&with_null).unwrap(), grown); + let set = Grown { + width: 7, + pipeline: Some(VideoPipeline::Cpu), + }; + let bytes = minicbor::to_vec(&set).unwrap(); + assert_eq!(minicbor::decode::(&bytes).unwrap(), set); + assert_eq!( + minicbor::decode::(&bytes).unwrap(), + Original { width: 7 } + ); + } } diff --git a/crates/gliff-server/src/session.rs b/crates/gliff-server/src/session.rs index 681fe331..698f2c89 100644 --- a/crates/gliff-server/src/session.rs +++ b/crates/gliff-server/src/session.rs @@ -13,8 +13,9 @@ use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender}; use bytes::Bytes; use gliff_proto::clipboard::CHUNK; use gliff_proto::{ - ChromaMode, ClientCaps, ClientMsg, ClipboardMsg, Codec, OutputInfo as ProtoOutput, Rect, - ServerMsg, SessionInfo, VideoPipeline, PROTOCOL_VERSION, + features, version_mismatch, ChromaMode, ClientCaps, ClientMsg, ClipboardMsg, Codec, Greeting, + OutputInfo as ProtoOutput, Rect, ServerMsg, SessionInfo, VideoPipeline, ERROR_NO_CHROMA, + ERROR_NO_CODEC, ERROR_VERSION, PROTOCOL_VERSION, }; use gliff_sw::VideoMode; use gliff_transport::clipboard::progress::Jobs; @@ -77,22 +78,31 @@ where let mut writer = Framed::new(wr); let (keymap, caps) = handshake(&mut reader, &mut writer).await?; - - let instance = cfg.target.instance().context("find Hyprland instance")?; - let output = setup_output(&instance, &cfg, &caps)?; - tracing::info!(output = %output.name, output.width, output.height, headless = output.is_headless(), "session output ready"); - - let video = VideoTier::open(cfg.video, &cfg.render_node); + if !caps.codecs.contains(&Codec::H264) { + writer + .write_msg(&ServerMsg::error( + ERROR_NO_CODEC, + "No video codec in common", + )) + .await?; + anyhow::bail!("client advertises no codec this server speaks"); + } if !caps.chroma.contains(&ChromaMode::Dual420) && !caps.chroma.contains(&ChromaMode::Single420) { writer - .write_msg(&ServerMsg::Error { - code: 2, - message: "no common chroma mode".into(), - }) + .write_msg(&ServerMsg::error( + ERROR_NO_CHROMA, + "No chroma mode in common", + )) .await?; anyhow::bail!("client advertises no chroma mode this server speaks"); } + + let instance = cfg.target.instance().context("find Hyprland instance")?; + let output = setup_output(&instance, &cfg, &caps)?; + tracing::info!(output = %output.name, output.width, output.height, headless = output.is_headless(), "session output ready"); + + let video = VideoTier::open(cfg.video, &cfg.render_node); let codec = Codec::H264; writer .write_msg(&ServerMsg::HelloAck { @@ -107,6 +117,7 @@ where height: output.height, scale_milli: (output.scale * 1000.0).round() as u32, }], + features: features(), }) .await?; let ladder = Ladder::new(START_LEVEL); @@ -323,24 +334,18 @@ where R: AsyncRead + Unpin, W: AsyncWrite + Unpin, { - match reader.read_msg::().await.context("read Hello")? { - ClientMsg::Hello { - version, - keymap, - caps, - } => { - if version != PROTOCOL_VERSION { - writer - .write_msg(&ServerMsg::Error { - code: 1, - message: format!("version {version} unsupported"), - }) - .await?; - anyhow::bail!("client version {version} != {PROTOCOL_VERSION}"); - } + match reader.read_msg::().await.context("read Hello")? { + Greeting::Hello { keymap, caps } => { + tracing::debug!(features = ?caps.features, "client features"); Ok((keymap, caps)) } - other => anyhow::bail!("expected Hello, got {other:?}"), + Greeting::OtherVersion(version) => { + let message = version_mismatch(version, PROTOCOL_VERSION); + writer + .write_msg(&ServerMsg::error(ERROR_VERSION, message)) + .await?; + anyhow::bail!("{message}: client protocol {version}, server {PROTOCOL_VERSION}"); + } } } @@ -363,7 +368,11 @@ where loop { let msg = match reader.read_msg::().await { Ok(m) => m, - Err(_) => break, + Err(gliff_transport::Error::Closed) => break, + Err(e) => { + tracing::warn!(error = %e, "cannot read from the client"); + break; + } }; let inbound = match msg.into_clipboard() { Err(msg) => Inbound::Msg(msg), @@ -1347,7 +1356,7 @@ fn stream_config( ServerMsg::StreamConfig { codec, chroma, - pipeline, + pipeline: Some(pipeline), width: stream.0, height: stream.1, scale_milli: (effective_scale * 1000.0).round() as u32, diff --git a/crates/gliff-server/src/writer.rs b/crates/gliff-server/src/writer.rs index 14f3d280..905ff864 100644 --- a/crates/gliff-server/src/writer.rs +++ b/crates/gliff-server/src/writer.rs @@ -166,7 +166,7 @@ mod tests { ServerMsg::StreamConfig { codec: Codec::H264, chroma: ChromaMode::Single420, - pipeline: VideoPipeline::Gpu, + pipeline: Some(VideoPipeline::Gpu), width, height: 480, scale_milli: 1000, diff --git a/crates/gliff-transport/Cargo.toml b/crates/gliff-transport/Cargo.toml index dbadd4bd..8c4ba235 100644 --- a/crates/gliff-transport/Cargo.toml +++ b/crates/gliff-transport/Cargo.toml @@ -12,8 +12,7 @@ workspace = true gliff-proto.workspace = true tokio.workspace = true bytes.workspace = true -postcard.workspace = true -serde.workspace = true +minicbor.workspace = true thiserror.workspace = true tracing.workspace = true diff --git a/crates/gliff-transport/src/lib.rs b/crates/gliff-transport/src/lib.rs index d4df5db4..2aa40c53 100644 --- a/crates/gliff-transport/src/lib.rs +++ b/crates/gliff-transport/src/lib.rs @@ -1,19 +1,22 @@ //! Length-prefixed framing over any tokio `AsyncRead + AsyncWrite`, the //! clipboard transfer engine, plus ssh process spawning for the client. //! -//! Each frame is a postcard-encoded message (`ClientMsg`/`ServerMsg`) preceded -//! by a `u32` little-endian length, optionally followed by raw payload bytes -//! the message declares (`data_len`, `aux_len`, `argb_len`). Payloads never go -//! through postcard; the writer sends them with `write_vectored` and the reader -//! returns them as a `Bytes` slice for the decoder. +//! Each frame is a CBOR-encoded message (`ClientMsg`/`ServerMsg`) behind a +//! header that gives its length and the total length of the raw payload bytes +//! after it (the `data_len`, `aux_len`, `argb_len` the message declares). +//! Payloads never go through the codec; the writer sends them with +//! `write_vectored` and the reader returns them as a `Bytes` slice for the +//! decoder. Because the header covers the payload, a message this build does +//! not know is skipped whole. pub mod clipboard; mod ssh; -use bytes::{Bytes, BytesMut}; -use gliff_proto::frame::MAX_FRAME_BODY; -use serde::de::DeserializeOwned; -use serde::Serialize; +use bytes::{Buf, Bytes, BytesMut}; +pub use gliff_proto::frame::MAX_PAYLOAD; +use gliff_proto::frame::{HEADER_LEN, MAX_FRAME_BODY, MAX_FRAME_PAYLOAD}; +use minicbor::decode::Decoder; +use minicbor::{Decode, Encode}; use std::io::IoSlice; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; @@ -23,28 +26,40 @@ pub use ssh::{spawn_ssh, SshTarget}; pub enum Error { #[error("io: {0}")] Io(#[from] std::io::Error), - #[error("postcard: {0}")] - Postcard(#[from] postcard::Error), + #[error("encode: {0}")] + Encode(#[from] minicbor::encode::Error), + #[error("decode: {0}")] + Decode(#[from] minicbor::decode::Error), #[error("frame body of {0} bytes exceeds the limit")] BodyTooLarge(u32), #[error("payload of {0} bytes exceeds the limit")] PayloadTooLarge(u32), + #[error("read {asked} payload bytes, but the frame has {left} left")] + PayloadOverrun { asked: u32, left: u32 }, #[error("connection closed")] Closed, } pub type Result = std::result::Result; -/// Largest single payload (a keyframe) we will read. -pub const MAX_PAYLOAD: u32 = 64 * 1024 * 1024; +#[derive(Clone, Copy)] +struct Header { + body_len: u32, + payload_len: u32, +} /// Framed reader/writer over a split or duplex stream. pub struct Framed { stream: S, read_buf: BytesMut, - /// Body length of a frame whose prefix is consumed but whose body has - /// not arrived, so a cancelled `read_msg` resumes where it stopped. - pending_len: Option, + /// Header of a frame whose body has not arrived, so a cancelled + /// `read_msg` resumes where it stopped. + pending: Option
, + /// Payload bytes of the last message the caller has not read yet. + payload_left: u32, + /// Bytes to discard before the next header: a skipped message's payload, + /// or payload the caller left unread. + skip_left: u64, } impl Framed { @@ -52,7 +67,9 @@ impl Framed { Self { stream, read_buf: BytesMut::with_capacity(64 * 1024), - pending_len: None, + pending: None, + payload_left: 0, + skip_left: 0, } } @@ -63,22 +80,34 @@ impl Framed { impl Framed { /// Write a message with no trailing payload. - pub async fn write_msg(&mut self, msg: &M) -> Result<()> { + pub async fn write_msg>(&mut self, msg: &M) -> Result<()> { self.write_msg_with_payloads(msg, &[]).await } /// Write a message then the given payload slices, all in order. Uses /// `write_vectored` so a large encoder payload is not copied into a /// staging buffer first. - pub async fn write_msg_with_payloads( + pub async fn write_msg_with_payloads>( &mut self, msg: &M, payloads: &[&[u8]], ) -> Result<()> { - let body = postcard::to_stdvec(msg)?; - let len = (body.len() as u32).to_le_bytes(); + let body = minicbor::to_vec(msg)?; + if body.len() > MAX_FRAME_BODY { + return Err(Error::BodyTooLarge(saturating_u32(body.len()))); + } + if let Some(p) = payloads.iter().find(|p| p.len() > MAX_PAYLOAD as usize) { + return Err(Error::PayloadTooLarge(saturating_u32(p.len()))); + } + let payload_len: usize = payloads.iter().map(|p| p.len()).sum(); + if payload_len > MAX_FRAME_PAYLOAD as usize { + return Err(Error::PayloadTooLarge(saturating_u32(payload_len))); + } + let mut header = [0u8; HEADER_LEN]; + header[..4].copy_from_slice(&(body.len() as u32).to_le_bytes()); + header[4..].copy_from_slice(&(payload_len as u32).to_le_bytes()); let mut parts: Vec<&[u8]> = Vec::with_capacity(2 + payloads.len()); - parts.push(&len); + parts.push(&header); parts.push(&body); parts.extend_from_slice(payloads); write_all_vectored(&mut self.stream, &parts).await?; @@ -88,27 +117,50 @@ impl Framed { } impl Framed { - /// Read one message body. Payloads (if any) must then be read with - /// [`read_payload`] according to the message's declared lengths. + /// Read one message body. Its payload (if any) must then be read with + /// [`read_payload`] according to the message's declared lengths; payload + /// left unread is discarded before the next message. A message this + /// build does not know is skipped, payload and all. /// - /// Cancel-safe: a caller may race this in a `select!`. A parsed length - /// prefix is kept across a cancellation, and buffered stream bytes are - /// only consumed once the full item is present. - pub async fn read_msg(&mut self) -> Result { - let len = match self.pending_len { - Some(len) => len, - None => { - let len = self.read_u32().await?; - if len as usize > MAX_FRAME_BODY { - return Err(Error::BodyTooLarge(len)); + /// Cancel-safe: a caller may race this in a `select!`. A parsed header + /// and a partly discarded payload are kept across a cancellation, and + /// buffered stream bytes are only consumed once the full item is present. + pub async fn read_msg Decode<'b, ()>>(&mut self) -> Result { + loop { + self.skip_left += u64::from(std::mem::take(&mut self.payload_left)); + self.discard_skipped().await?; + let header = match self.pending { + Some(h) => h, + None => { + let b = self.read_exact_bytes(HEADER_LEN).await?; + let h = Header { + body_len: u32::from_le_bytes([b[0], b[1], b[2], b[3]]), + payload_len: u32::from_le_bytes([b[4], b[5], b[6], b[7]]), + }; + if h.body_len as usize > MAX_FRAME_BODY { + return Err(Error::BodyTooLarge(h.body_len)); + } + if h.payload_len > MAX_FRAME_PAYLOAD { + return Err(Error::PayloadTooLarge(h.payload_len)); + } + self.pending = Some(h); + h + } + }; + let body = self.read_exact_bytes(header.body_len as usize).await?; + self.pending = None; + match minicbor::decode::(&body) { + Ok(msg) => { + self.payload_left = header.payload_len; + return Ok(msg); } - self.pending_len = Some(len); - len + Err(e) if is_unknown_message(&body, &e) => { + tracing::debug!(error = %e, "skipping a message this build does not know"); + self.skip_left += u64::from(header.payload_len); + } + Err(e) => return Err(e.into()), } - }; - let body = self.read_exact_bytes(len as usize).await?; - self.pending_len = None; - Ok(postcard::from_bytes(&body)?) + } } /// Read exactly `len` payload bytes that follow a message. @@ -116,12 +168,27 @@ impl Framed { if len > MAX_PAYLOAD { return Err(Error::PayloadTooLarge(len)); } - Ok(self.read_exact_bytes(len as usize).await?.freeze()) + if len > self.payload_left { + return Err(Error::PayloadOverrun { + asked: len, + left: self.payload_left, + }); + } + let bytes = self.read_exact_bytes(len as usize).await?.freeze(); + self.payload_left -= len; + Ok(bytes) } - async fn read_u32(&mut self) -> Result { - let b = self.read_exact_bytes(4).await?; - Ok(u32::from_le_bytes([b[0], b[1], b[2], b[3]])) + async fn discard_skipped(&mut self) -> Result<()> { + while self.skip_left > 0 { + if self.read_buf.is_empty() && self.stream.read_buf(&mut self.read_buf).await? == 0 { + return Err(Error::Closed); + } + let n = self.read_buf.len().min(self.skip_left as usize); + self.read_buf.advance(n); + self.skip_left -= n as u64; + } + Ok(()) } async fn read_exact_bytes(&mut self, n: usize) -> Result { @@ -136,6 +203,18 @@ impl Framed { } } +fn saturating_u32(n: usize) -> u32 { + u32::try_from(n).unwrap_or(u32::MAX) +} + +/// Whether `e` is the message's own variant being unknown, as opposed to an +/// unknown value inside a message this build does know. Messages encode as +/// `[variant, fields]`, so the variant index sits right after the array head. +fn is_unknown_message(body: &[u8], e: &minicbor::decode::Error) -> bool { + let mut d = Decoder::new(body); + e.is_unknown_variant() && d.array().is_ok() && e.position() == Some(d.position()) +} + /// Write every byte of `parts` in order using vectored writes, handling /// partial writes by tracking a global offset and rebuilding the `IoSlice` /// list for the unwritten remainder each iteration. @@ -166,7 +245,8 @@ async fn write_all_vectored(stream: &mut S, parts: &[&[u8 #[cfg(test)] mod tests { use super::*; - use gliff_proto::{ChromaMode, ClientCaps, ClientMsg, Codec}; + use gliff_proto::frame::MAX_FRAME_BODY; + use gliff_proto::{ChromaMode, ClientCaps, ClientMsg, Codec, PROTOCOL_VERSION}; #[tokio::test] async fn round_trips_message_and_payload() { @@ -174,13 +254,14 @@ mod tests { let mut wr = Framed::new(a); let mut rd = Framed::new(b); let msg = ClientMsg::Hello { - version: 1, + version: PROTOCOL_VERSION, keymap: "k".into(), caps: ClientCaps { codecs: vec![Codec::H264], max_width: 1920, max_height: 1080, chroma: vec![ChromaMode::Dual420], + features: vec![], }, }; let payload = vec![7u8; 5000]; @@ -201,4 +282,219 @@ mod tests { assert_eq!(&gp2[..], &p2[..]); task.await.unwrap(); } + + /// Messages a newer peer might send: one this build has no variant for, + /// and a known one carrying an enum value this build does not know. + #[derive(Encode)] + enum NewerClientMsg { + #[n(2)] + Key { + #[n(0)] + keycode: u32, + #[n(1)] + pressed: bool, + }, + #[n(5)] + PointerAxis { + #[n(0)] + axis: NewerAxis, + #[n(1)] + value: f64, + #[n(2)] + discrete: Option, + #[n(3)] + stop: bool, + }, + #[n(99)] + Future { + #[n(0)] + data_len: u32, + }, + } + + #[derive(Encode)] + #[cbor(index_only)] + enum NewerAxis { + #[n(7)] + Diagonal, + } + + const KEY: NewerClientMsg = NewerClientMsg::Key { + keycode: 30, + pressed: true, + }; + + #[tokio::test] + async fn skips_an_unknown_message_and_its_payload() { + let (a, b) = tokio::io::duplex(64 * 1024); + let mut wr = Framed::new(a); + let mut rd = Framed::new(b); + let payload = vec![1u8; 40_000]; + let task = tokio::spawn(async move { + wr.write_msg_with_payloads(&NewerClientMsg::Future { data_len: 40_000 }, &[&payload]) + .await + .unwrap(); + wr.write_msg(&KEY).await.unwrap(); + }); + let got: ClientMsg = rd.read_msg().await.unwrap(); + assert_eq!( + got, + ClientMsg::Key { + keycode: 30, + pressed: true + } + ); + task.await.unwrap(); + } + + #[tokio::test] + async fn discards_payload_the_caller_leaves_unread() { + let (a, b) = tokio::io::duplex(64 * 1024); + let mut wr = Framed::new(a); + let mut rd = Framed::new(b); + let chunk = ClientMsg::ClipboardData { + id: 1, + offset: 0, + data_len: 3, + done: true, + }; + let sent = chunk.clone(); + let task = tokio::spawn(async move { + wr.write_msg_with_payloads(&sent, &[b"abc"]).await.unwrap(); + wr.write_msg(&KEY).await.unwrap(); + }); + assert_eq!(rd.read_msg::().await.unwrap(), chunk); + assert!(matches!( + rd.read_payload(4).await, + Err(Error::PayloadOverrun { asked: 4, left: 3 }) + )); + assert!(matches!( + rd.read_msg::().await.unwrap(), + ClientMsg::Key { .. } + )); + task.await.unwrap(); + } + + #[tokio::test] + async fn fails_on_an_unknown_value_inside_a_known_message() { + let (a, b) = tokio::io::duplex(4096); + let mut wr = Framed::new(a); + let mut rd = Framed::new(b); + let task = tokio::spawn(async move { + let axis = NewerClientMsg::PointerAxis { + axis: NewerAxis::Diagonal, + value: 1.0, + discrete: None, + stop: false, + }; + wr.write_msg(&axis).await.unwrap(); + }); + assert!(matches!( + rd.read_msg::().await, + Err(Error::Decode(e)) if e.is_unknown_variant() + )); + task.await.unwrap(); + } + + #[tokio::test] + async fn frame_layout_is_stable() { + let mut wr = Framed::new(Vec::new()); + wr.write_msg_with_payloads(&KEY, &[b"ab"]).await.unwrap(); + let hex: String = wr.into_inner().iter().map(|b| format!("{b:02x}")).collect(); + assert_eq!(hex, "0600000002000000820282181ef56162"); + } + + #[tokio::test] + async fn writer_refuses_frames_the_reader_would_reject() { + let mut wr = Framed::new(Vec::new()); + let huge = ClientMsg::Keymap { + keymap: "x".repeat(MAX_FRAME_BODY), + }; + assert!(matches!( + wr.write_msg(&huge).await, + Err(Error::BodyTooLarge(_)) + )); + assert!(wr.into_inner().is_empty()); + let mut wr = Framed::new(Vec::new()); + let big = vec![0u8; MAX_PAYLOAD as usize + 1]; + assert!(matches!( + wr.write_msg_with_payloads(&KEY, &[&big]).await, + Err(Error::PayloadTooLarge(_)) + )); + assert!(wr.into_inner().is_empty()); + } + + #[tokio::test] + async fn reader_refuses_oversized_headers() { + let mut header = Vec::new(); + header.extend_from_slice(&6u32.to_le_bytes()); + header.extend_from_slice(&(MAX_FRAME_PAYLOAD + 1).to_le_bytes()); + let mut rd = Framed::new(&header[..]); + assert!(matches!( + rd.read_msg::().await, + Err(Error::PayloadTooLarge(_)) + )); + let mut header = Vec::new(); + header.extend_from_slice(&(MAX_FRAME_BODY as u32 + 1).to_le_bytes()); + header.extend_from_slice(&0u32.to_le_bytes()); + let mut rd = Framed::new(&header[..]); + assert!(matches!( + rd.read_msg::().await, + Err(Error::BodyTooLarge(_)) + )); + } + + /// Read a message while another branch keeps winning the race, as a + /// `select!` in the session loop may. + async fn read_racing( + rd: &mut Framed, + cancelled: &mut u32, + ) -> ClientMsg { + loop { + tokio::select! { + biased; + _ = tokio::task::yield_now() => *cancelled += 1, + msg = rd.read_msg::() => return msg.unwrap(), + } + } + } + + /// `read_msg` is raced in `select!`, so it must resume cleanly after + /// being cancelled mid-header, mid-body and mid-skip. + #[tokio::test] + async fn read_msg_survives_cancellation() { + let mut frames = Framed::new(Vec::new()); + frames + .write_msg_with_payloads(&NewerClientMsg::Future { data_len: 300 }, &[&[5u8; 300]]) + .await + .unwrap(); + let chunk = ClientMsg::ClipboardData { + id: 1, + offset: 0, + data_len: 3, + done: true, + }; + frames + .write_msg_with_payloads(&chunk, &[b"abc"]) + .await + .unwrap(); + frames.write_msg(&KEY).await.unwrap(); + let bytes = frames.into_inner(); + + let (mut a, b) = tokio::io::duplex(1); + let task = tokio::spawn(async move { + for byte in bytes { + a.write_all(&[byte]).await.unwrap(); + tokio::task::yield_now().await; + } + }); + let mut rd = Framed::new(b); + let mut cancelled = 0; + assert_eq!(read_racing(&mut rd, &mut cancelled).await, chunk); + assert_eq!(&rd.read_payload(3).await.unwrap()[..], b"abc"); + let key = read_racing(&mut rd, &mut cancelled).await; + assert!(matches!(key, ClientMsg::Key { .. })); + assert!(cancelled > 100, "only {cancelled} cancellations"); + task.await.unwrap(); + } } diff --git a/crates/gliff/Cargo.toml b/crates/gliff/Cargo.toml index 65c9ac64..abd8b094 100644 --- a/crates/gliff/Cargo.toml +++ b/crates/gliff/Cargo.toml @@ -10,6 +10,7 @@ workspace = true [dependencies] anyhow.workspace = true +thiserror.workspace = true clap.workspace = true tracing.workspace = true tracing-subscriber.workspace = true diff --git a/crates/gliff/src/main.rs b/crates/gliff/src/main.rs index 778a863f..f5a8eed0 100644 --- a/crates/gliff/src/main.rs +++ b/crates/gliff/src/main.rs @@ -586,8 +586,7 @@ fn schedule_reconnect(ui: Rc, session: u64) { if n > MAX_RETRIES { ui.status .set_text("Disconnected — press Reconnect to retry"); - ui.connect_btn.set_icon_name(RECONNECT_ICON); - ui.connect_btn.set_tooltip_text(Some("Reconnect")); + offer_reconnect(&ui); return; } let Some(endpoint) = ui.endpoint.borrow().clone() else { @@ -602,6 +601,11 @@ fn schedule_reconnect(ui: Rc, session: u64) { }); } +fn offer_reconnect(ui: &App) { + ui.connect_btn.set_icon_name(RECONNECT_ICON); + ui.connect_btn.set_tooltip_text(Some("Reconnect")); +} + /// Pull decoded frames on the GTK main loop, latest-wins, and paint them. fn poll_frames(ui: Rc, rx: Receiver, session: u64) { glib::timeout_add_local(Duration::from_millis(8), move || { @@ -766,6 +770,13 @@ fn poll_status(ui: Rc, rx: Receiver, session: u64) { schedule_reconnect(ui.clone(), session); return glib::ControlFlow::Break; } + Status::Incompatible(message) => { + tracing::error!(error = %message, "incompatible server"); + ui.status.set_text(&message); + show_disconnected(&ui); + offer_reconnect(&ui); + return glib::ControlFlow::Break; + } Status::Closed => { show_disconnected(&ui); schedule_reconnect(ui.clone(), session); diff --git a/crates/gliff/src/net.rs b/crates/gliff/src/net.rs index d3b45666..ee79dfbf 100644 --- a/crates/gliff/src/net.rs +++ b/crates/gliff/src/net.rs @@ -12,8 +12,8 @@ use std::sync::Arc; use bytes::Bytes; use gliff_proto::clipboard::CHUNK; use gliff_proto::{ - ChromaMode, ClientCaps, ClientMsg, ClipboardFile, ClipboardMsg, Codec, ServerMsg, - PROTOCOL_VERSION, + features, is_incompatible, version_mismatch, ChromaMode, ClientCaps, ClientMsg, ClipboardFile, + ClipboardMsg, Codec, ServerMsg, VideoPipeline, PROTOCOL_VERSION, }; use gliff_sw::VideoMode; use gliff_transport::clipboard::progress::Progress; @@ -89,9 +89,16 @@ pub enum Status { progress: Progress, }, Error(String), + /// The two ends speak different protocol versions, so retrying cannot + /// help. + Incompatible(String), Closed, } +#[derive(Debug, thiserror::Error)] +#[error("{0}")] +struct Incompatible(String); + /// How to reach the server. #[derive(Clone)] pub enum Endpoint { @@ -165,7 +172,10 @@ impl Worker { } }); if let Err(e) = result { - let _ = status.send(Status::Error(e.to_string())); + let _ = status.send(match e.downcast::() { + Ok(Incompatible(message)) => Status::Incompatible(message), + Err(e) => Status::Error(e.to_string()), + }); } else { let _ = status.send(Status::Closed); } @@ -239,6 +249,7 @@ where } else { vec![ChromaMode::Single420] }, + features: features(), }; let _ = tokio::time::timeout(FIRST_KEYMAP_WAIT, keymap.wait_for(|k| !k.is_empty())).await; let first_keymap = keymap.borrow_and_update().clone(); @@ -260,10 +271,15 @@ where let mut got_ack = false; let (mut width, mut height, mut chroma, mut scale_milli, pipeline, mut view, mut fps_cap) = loop { match reader.read_msg::().await? { - ServerMsg::HelloAck { version, .. } => { + ServerMsg::HelloAck { + version, features, .. + } => { if version != PROTOCOL_VERSION { - anyhow::bail!("server version {version} != {PROTOCOL_VERSION}"); + return Err( + Incompatible(version_mismatch(PROTOCOL_VERSION, version).into()).into(), + ); } + tracing::debug!(?features, "server features"); got_ack = true; } ServerMsg::Ping { t } => writer.write_msg(&ClientMsg::Pong { t }).await?, @@ -288,7 +304,10 @@ where fps_cap, ) } - ServerMsg::Error { code, message } => { + ServerMsg::Error { code, message, .. } if is_incompatible(code) => { + return Err(Incompatible(message).into()) + } + ServerMsg::Error { code, message, .. } => { anyhow::bail!("server error {code}: {message}") } other => anyhow::bail!("unexpected message before StreamConfig: {other:?}"), @@ -296,7 +315,7 @@ where }; let mut decoder = new_decoder(&gpu, chroma, width, height)?; let local = if gpu.is_some() { "gpu" } else { "cpu" }; - let mut video_label = format!("{} > {}", pipeline.label(), local); + let mut video_label = format!("{} > {}", pipeline.map_or("?", VideoPipeline::label), local); let _ = status.send(Status::Connected { video: video_label.clone(), view_width: view.0, @@ -552,7 +571,7 @@ where } (width, height, chroma, scale_milli) = (w, h, c, s); (view, fps_cap) = ((view_width, view_height), f); - video_label = format!("{} > {}", pipeline.label(), local); + video_label = format!("{} > {}", pipeline.map_or("?", VideoPipeline::label), local); let _ = status.send(Status::Connected { video: video_label.clone(), view_width, @@ -581,7 +600,12 @@ where let _ = out_tx.send((ClientMsg::Pong { t }, Bytes::new())); } ServerMsg::CursorPos { .. } | ServerMsg::Pong { .. } => {} - ServerMsg::Error { code, message } => anyhow::bail!("server error {code}: {message}"), + ServerMsg::Error { code, message, .. } if is_incompatible(code) => { + return Err(Incompatible(message).into()) + } + ServerMsg::Error { code, message, .. } => { + anyhow::bail!("server error {code}: {message}") + } ServerMsg::HelloAck { .. } => {} other => { if let Ok(clip) = other.into_clipboard() { diff --git a/docs/architecture.md b/docs/architecture.md index e5252a86..e4f74a04 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -58,9 +58,17 @@ round-trip. - **SSH is the only transport.** The server opens no port in production; the client spawns `ssh -T host gliff-server --stdio` and the protocol runs over that pipe. `--listen`/`--connect` exist for local development only and have no - auth. TCP framing is length-prefixed postcard messages; large video and cursor - payloads ride outside the postcard body so the encoder output is sent with - `write_vectored` and read straight into the decoder. + auth. Each frame is a header (body and payload lengths), a CBOR message with + numeric keys, then any payload; large video and cursor payloads ride outside + the CBOR body so the encoder output is sent with `write_vectored` and read + straight into the decoder. + +- **The protocol evolves without version bumps.** Peers ignore fields they do + not know and skip whole messages they do not know, since the frame header + covers the payload too. Anything new is sent only to a peer that listed the + matching feature in the handshake. `PROTOCOL_VERSION` changes only for a + break those rules cannot absorb; the rules are at the top of + `crates/gliff-proto/src/msg.rs`. - **Why not UDP.** Measured on 2026-09-14 between two Wi-Fi machines over Tailscale's direct path (MTU 1280, 330-380 Mbit/s of SSH throughput, RTT From c29066028b6f403e3cd56987e24d5f5155faf7b2 Mon Sep 17 00:00:00 2001 From: Kevin McConnell Date: Wed, 23 Sep 2026 17:44:04 -0500 Subject: [PATCH 2/3] End the client session when a write fails The writer task stopped on its first error without telling anyone, so input and frame acks went nowhere while the session looked alive. --- crates/gliff/src/net.rs | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/crates/gliff/src/net.rs b/crates/gliff/src/net.rs index ee79dfbf..6b33bc5f 100644 --- a/crates/gliff/src/net.rs +++ b/crates/gliff/src/net.rs @@ -9,6 +9,7 @@ use std::rc::Rc; use std::sync::mpsc::{Sender as StdSender, SyncSender}; use std::sync::Arc; +use anyhow::Context; use bytes::Bytes; use gliff_proto::clipboard::CHUNK; use gliff_proto::{ @@ -330,17 +331,15 @@ where // never block on a write and the two peers cannot deadlock. Both the reader // loop (acks, keyframe requests) and the UI thread (input) feed `out_tx`. let (out_tx, mut out_rx) = unbounded_channel::<(ClientMsg, Bytes)>(); - tokio::task::spawn_local(async move { + let mut writes = tokio::task::spawn_local(async move { while let Some((m, payload)) = out_rx.recv().await { - let r = if payload.is_empty() { - writer.write_msg(&m).await + if payload.is_empty() { + writer.write_msg(&m).await?; } else { - writer.write_msg_with_payloads(&m, &[&payload]).await - }; - if r.is_err() { - break; + writer.write_msg_with_payloads(&m, &[&payload]).await?; } } + Ok::<(), gliff_transport::Error>(()) }); // Clipboard transfers write through the same channel, in chunks. let (clip_out_tx, mut clip_out_rx) = outbound_channel(); @@ -421,6 +420,10 @@ where let read = tokio::select! { read = reader.read_msg::() => read, _ = &mut ui_gone => return Ok(()), + written = &mut writes => { + written.context("writer task")?.context("write to server")?; + return Ok(()); + } _ = tokio::time::sleep_until(drain_at), if undelivered.is_some() || matches!(&decoder, VideoDecoder::Cpu(d) if d.has_pending()) => From 9af13ca932f6475d5cb870e170e0f76a97ce1c1c Mon Sep 17 00:00:00 2001 From: Kevin McConnell Date: Wed, 23 Sep 2026 17:44:04 -0500 Subject: [PATCH 3/3] Cap file offers so one fits in a frame An offer of 10,000 long paths could exceed the 1 MiB frame body limit. Cap the summed path length, so the largest offer the other caps allow still fits. --- crates/gliff-proto/src/clipboard.rs | 17 +++++++++++++++++ crates/gliff-proto/src/msg.rs | 21 +++++++++++++++++++++ 2 files changed, 38 insertions(+) diff --git a/crates/gliff-proto/src/clipboard.rs b/crates/gliff-proto/src/clipboard.rs index 9d348dcb..7a79cc0a 100644 --- a/crates/gliff-proto/src/clipboard.rs +++ b/crates/gliff-proto/src/clipboard.rs @@ -31,6 +31,9 @@ pub const MAX_ITEM: u64 = 32 * 1024 * 1024; pub const MAX_FILES_TOTAL: u64 = 4 * 1024 * 1024 * 1024; /// Cap on the number of entries (files and directories) in one offer. pub const MAX_FILE_ENTRIES: usize = 10_000; +/// Cap on the summed path lengths in one offer, which with the per-entry +/// overhead keeps an offer well inside one frame. +pub const MAX_FILE_PATH_BYTES: usize = 512 * 1024; /// Mime type the pasting side uses for text it received as any text flavour. pub const TEXT_MIME: &str = "text/plain;charset=utf-8"; @@ -301,6 +304,10 @@ pub fn validate_files(files: &[ClipboardFile]) -> Result<(), FilesError> { if files.len() > MAX_FILE_ENTRIES { return Err(FilesError::TooMany(files.len())); } + let path_bytes: usize = files.iter().map(|f| f.path.len()).sum(); + if path_bytes > MAX_FILE_PATH_BYTES { + return Err(FilesError::PathsTooLong(path_bytes)); + } let mut total: u64 = 0; let mut seen = std::collections::HashSet::new(); for f in files { @@ -325,6 +332,8 @@ pub enum FilesError { TooMany(usize), #[error("files total {0} bytes, over the limit of {MAX_FILES_TOTAL}")] TooLarge(u64), + #[error("paths total {0} bytes, over the limit of {MAX_FILE_PATH_BYTES}")] + PathsTooLong(usize), #[error("unsafe or duplicate path {0:?}")] BadPath(String), } @@ -539,6 +548,14 @@ mod tests { validate_files(&[f("x", MAX_FILES_TOTAL + 1, false)]), Err(FilesError::TooLarge(_)) )); + let long_name = "x".repeat(200); + let long: Vec<_> = (0..MAX_FILE_PATH_BYTES / 200 + 1) + .map(|i| f(&format!("{i}{long_name}"), 0, false)) + .collect(); + assert!(matches!( + validate_files(&long), + Err(FilesError::PathsTooLong(_)) + )); let many: Vec<_> = (0..=MAX_FILE_ENTRIES) .map(|i| f(&format!("f{i}"), 0, false)) .collect(); diff --git a/crates/gliff-proto/src/msg.rs b/crates/gliff-proto/src/msg.rs index 1b57f56c..d1f6f114 100644 --- a/crates/gliff-proto/src/msg.rs +++ b/crates/gliff-proto/src/msg.rs @@ -980,6 +980,27 @@ mod tests { assert!(crate::cbor::known::<(), Codec>(&mut d, &mut ()).is_err()); } + #[test] + fn the_largest_file_offer_fits_in_a_frame() { + use crate::clipboard::{MAX_FILE_ENTRIES, MAX_FILE_PATH_BYTES}; + let path_len = MAX_FILE_PATH_BYTES / MAX_FILE_ENTRIES; + let offer = ClientMsg::ClipboardOffer { + serial: u32::MAX, + mime_types: crate::clipboard::TEXT_MIMES + .iter() + .map(|m| m.to_string()) + .collect(), + files: (0..MAX_FILE_ENTRIES) + .map(|i| ClipboardFile { + path: format!("{i:0path_len$}"), + size: u64::MAX, + dir: false, + }) + .collect(), + }; + assert!(minicbor::to_vec(&offer).unwrap().len() <= crate::frame::MAX_FRAME_BODY); + } + /// The shape every new field takes: an optional enum appended last. #[derive(Debug, PartialEq, Encode, Decode)] struct Grown {