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..7a79cc0a 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; @@ -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"; @@ -59,23 +62,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` @@ -294,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 { @@ -318,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), } @@ -532,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/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..d1f6f114 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,381 @@ 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()); + } + + #[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 { + #[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..6b33bc5f 100644 --- a/crates/gliff/src/net.rs +++ b/crates/gliff/src/net.rs @@ -9,11 +9,12 @@ 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::{ - 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 +90,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 +173,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 +250,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 +272,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 +305,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 +316,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, @@ -311,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(); @@ -402,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()) => @@ -552,7 +574,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 +603,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