From 938f5e65447ba1d77b2ad1e869fdde9d5aaf6b8b Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Wed, 26 Aug 2026 18:00:01 +0300 Subject: [PATCH] feat(nesprotocol): open the shared wire types MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First component into this repo. Renamed from `nestri-protocol` — everything else in the family carries the `nes` prefix and this was the odd one out. **Imported as a tree, not as history.** The upstream repo is private, so its commits and commit messages have never been reviewed for what may be published, and squashing avoids the failure this project has already documented once: a repo published wholesale carries private history with it, permanently. Origin is `nestrilabs/nestri-protocol`, and this is its state today rather than its past. The `control` module is deliberately left behind. It carries the host↔guest control channel, and its types are shaped by a payload that has no business being described in a public repo — a box is supposed to be able to run anything. It was already an optional feature that nothing here enables, so leaving it out costs nothing today and stops a boundary from being crossed by accident. What lands is the media protocol: frames, audio, cursor, input and stats. One definition shared by both ends, so no two can drift silently. 9 tests pass. One pre-existing clippy warning (`input.rs`, too many arguments) is left alone on purpose — an import commit should be a faithful copy, and mixing a cleanup into one makes both harder to read. Co-Authored-By: Claude Opus 5 (1M context) --- Cargo.lock | 7 + Cargo.toml | 4 +- crates/nesprotocol/.gitignore | 4 + crates/nesprotocol/Cargo.toml | 7 + crates/nesprotocol/src/datagram.rs | 215 +++++++++++++++++++ crates/nesprotocol/src/input.rs | 317 +++++++++++++++++++++++++++++ crates/nesprotocol/src/lib.rs | 224 ++++++++++++++++++++ crates/nesprotocol/src/reliable.rs | 183 +++++++++++++++++ crates/nesprotocol/src/stats.rs | 116 +++++++++++ 9 files changed, 1076 insertions(+), 1 deletion(-) create mode 100644 Cargo.lock create mode 100644 crates/nesprotocol/.gitignore create mode 100644 crates/nesprotocol/Cargo.toml create mode 100644 crates/nesprotocol/src/datagram.rs create mode 100644 crates/nesprotocol/src/input.rs create mode 100644 crates/nesprotocol/src/lib.rs create mode 100644 crates/nesprotocol/src/reliable.rs create mode 100644 crates/nesprotocol/src/stats.rs diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 00000000..366f0cb9 --- /dev/null +++ b/Cargo.lock @@ -0,0 +1,7 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "nesprotocol" +version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index fc458991..f02547aa 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,7 +12,9 @@ [workspace] resolver = "3" -members = [] +members = [ + "crates/nesprotocol", +] # One pin per dependency, here. A member never writes a version — that is what # stops two crates in one tree from disagreeing about tokio. diff --git a/crates/nesprotocol/.gitignore b/crates/nesprotocol/.gitignore new file mode 100644 index 00000000..89a525a7 --- /dev/null +++ b/crates/nesprotocol/.gitignore @@ -0,0 +1,4 @@ +.idea/ +.vscode/ +.zed/ +/target/ \ No newline at end of file diff --git a/crates/nesprotocol/Cargo.toml b/crates/nesprotocol/Cargo.toml new file mode 100644 index 00000000..057a8f0b --- /dev/null +++ b/crates/nesprotocol/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "nesprotocol" +version = "0.1.0" +description = "Wire types shared by the capture, audio and compositor components" +edition.workspace = true +license.workspace = true +repository.workspace = true diff --git a/crates/nesprotocol/src/datagram.rs b/crates/nesprotocol/src/datagram.rs new file mode 100644 index 00000000..2eb13c2c --- /dev/null +++ b/crates/nesprotocol/src/datagram.rs @@ -0,0 +1,215 @@ +//! Media transport over QUIC datagrams. +//! +//! Video and audio used to have one unidirectional QUIC stream each, and QUIC +//! streams are reliable *and ordered*: a single lost packet stalls every frame +//! queued behind it until the retransmission lands. On a LAN that is invisible. +//! Over the internet, and on mobile especially, frames are never lost but arrive +//! late and bunched, and playback sags with no dropped frame to explain it. +//! +//! Datagrams give up the ordering guarantee on purpose. A lost packet costs the +//! one frame it belonged to and nothing else. The receiver reassembles fragments +//! and decides what to do about the gaps, which is where that decision belongs: +//! for an interactive stream, showing the newer frame and asking for a keyframe +//! beats waiting for the older one every time. +//! +//! Input, cursor and stats stay on reliable streams. They are small, rare, and +//! stateful — a lost key-up is a stuck key. +//! +//! # Wire format +//! +//! ```text +//! [1B kind] [2B frame seq LE] [2B fragment index LE] [2B fragment count LE] [fragment] +//! ``` +//! +//! The fragments carry a frame *body*: exactly the bytes [`encode_frame`] produces +//! minus its four-byte length prefix, which only a stream needs since a datagram +//! already knows its own length. Reassembled, the body is `[1B type][2B seq][payload]` +//! and parses exactly as it did when it arrived on a stream. +//! +//! The frame's sequence number therefore appears twice: once in the datagram +//! header, where reassembly needs it before the body exists, and once inside the +//! body, where the stream-era parser has always read it. The duplication buys an +//! unchanged frame parser on the far side, which is worth two bytes. +//! +//! [`encode_frame`]: crate::encode_frame + +/// kind(1) + frame seq(2) + fragment index(2) + fragment count(2). +pub const DGRAM_HDR_LEN: usize = 7; + +/// Datagram kinds. These share the numbering of the stream types they replace, +/// so a frame's identity does not change with the transport under it. +pub const DGRAM_VIDEO: u8 = 0; +pub const DGRAM_AUDIO: u8 = 1; + +/// Bounds reassembly work. A 4K keyframe runs to a few hundred fragments at a +/// typical MTU; anything past this did not come from a sender of ours. +pub const MAX_FRAGMENTS_PER_FRAME: usize = 4096; + +/// Floor on the usable payload per datagram. If the path cannot carry even this +/// much there is no point fragmenting — a frame would need more pieces than the +/// limit allows long before it arrived. +pub const MIN_DGRAM_PAYLOAD: usize = 64; + +/// Datagram buffer headroom to give a connection carrying media, at each end. +/// +/// The QUIC defaults are sized for streams, where a sender is paced by flow +/// control. A keyframe is not paced: it is a burst of hundreds of datagrams +/// handed over at once. When a datagram buffer fills, QUIC drops the *oldest* +/// queued datagrams — for a keyframe that means dropping its beginning and +/// delivering a tail that reassembles into nothing, and since every later frame +/// predicts from that keyframe, the stream cannot recover until the next one. +/// Both ends therefore get room well past the largest frame we expect. +pub const DGRAM_BUFFER_BYTES: usize = 4 * 1024 * 1024; + +/// How long a receiver holds the pieces of a frame that has not fully arrived. +/// +/// This is not a playout deadline: a frame missing a fragment is never going to +/// be shown, so this only bounds how long its pieces occupy memory. +pub const PARTIAL_TIMEOUT_MS: u64 = 500; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DatagramHeader { + pub kind: u8, + /// Sequence number of the frame this fragment belongs to. + pub seq: u16, + /// This fragment's position within the frame, `0..total`. + pub index: u16, + /// How many fragments the whole frame was split into. Always at least 1. + pub total: u16, +} + +/// Write a datagram header into the first [`DGRAM_HDR_LEN`] bytes of `buf`. +/// +/// # Panics +/// If `buf` is shorter than [`DGRAM_HDR_LEN`]. +pub fn write_datagram_header(buf: &mut [u8], kind: u8, seq: u16, index: u16, total: u16) { + buf[0] = kind; + buf[1..3].copy_from_slice(&seq.to_le_bytes()); + buf[3..5].copy_from_slice(&index.to_le_bytes()); + buf[5..7].copy_from_slice(&total.to_le_bytes()); +} + +/// Split a received datagram into its header and fragment payload. +/// +/// Returns `None` for anything too short or self-inconsistent. A datagram is +/// unauthenticated only in the sense that it may be truncated or corrupt in +/// transit below QUIC's own protection, but a bad `total` would still size an +/// allocation, so it is rejected here rather than trusted downstream. +pub fn decode_datagram(buf: &[u8]) -> Option<(DatagramHeader, &[u8])> { + if buf.len() < DGRAM_HDR_LEN { + return None; + } + let total = u16::from_le_bytes([buf[5], buf[6]]); + let index = u16::from_le_bytes([buf[3], buf[4]]); + if total == 0 || total as usize > MAX_FRAGMENTS_PER_FRAME || index >= total { + return None; + } + let header = DatagramHeader { + kind: buf[0], + seq: u16::from_le_bytes([buf[1], buf[2]]), + index, + total, + }; + Some((header, &buf[DGRAM_HDR_LEN..])) +} + +/// How many datagrams a body of this size needs. +/// +/// An empty body still takes one, so that the receiver sees the frame at all. +pub fn fragment_count(body_len: usize, payload_size: usize) -> usize { + if body_len == 0 || payload_size == 0 { + return 1; + } + body_len.div_ceil(payload_size) +} + +/// Compare 16-bit sequence numbers across wraparound: is `a` older than `b`? +/// +/// Straight comparison breaks at the wrap — 65535 is older than 0, not newer. +/// Serial arithmetic treats the shorter way round the circle as the true order, +/// which is right as long as the two are less than half a lap apart. At 60fps a +/// half-lap is nine minutes of frames, so that holds for anything the reorder +/// window could plausibly be looking at. +pub fn seq_older_than(a: u16, b: u16) -> bool { + a != b && b.wrapping_sub(a) < 0x8000 +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn header_roundtrips() { + let mut buf = [0u8; DGRAM_HDR_LEN + 3]; + write_datagram_header(&mut buf, DGRAM_VIDEO, 0xBEEF, 2, 7); + buf[DGRAM_HDR_LEN..].copy_from_slice(&[1, 2, 3]); + + let (header, payload) = decode_datagram(&buf).expect("valid datagram"); + assert_eq!( + header, + DatagramHeader { + kind: DGRAM_VIDEO, + seq: 0xBEEF, + index: 2, + total: 7 + } + ); + assert_eq!(payload, &[1, 2, 3]); + } + + #[test] + fn rejects_malformed_headers() { + assert!(decode_datagram(&[0u8; DGRAM_HDR_LEN - 1]).is_none()); + + // total == 0 describes a frame with no fragments. + let mut buf = [0u8; DGRAM_HDR_LEN]; + write_datagram_header(&mut buf, DGRAM_VIDEO, 1, 0, 0); + assert!(decode_datagram(&buf).is_none()); + + // index outside the fragment count. + write_datagram_header(&mut buf, DGRAM_VIDEO, 1, 4, 4); + assert!(decode_datagram(&buf).is_none()); + + // A count that would size an implausible allocation. + write_datagram_header(&mut buf, DGRAM_VIDEO, 1, 0, u16::MAX); + assert!(decode_datagram(&buf).is_none()); + } + + #[test] + fn empty_body_still_takes_one_fragment() { + assert_eq!(fragment_count(0, 1000), 1); + assert_eq!(fragment_count(1, 1000), 1); + assert_eq!(fragment_count(1000, 1000), 1); + assert_eq!(fragment_count(1001, 1000), 2); + assert_eq!(fragment_count(2500, 1000), 3); + } + + #[test] + fn a_datagram_body_parses_like_a_stream_frame() { + // The whole point of reusing the body layout: what comes back out of + // reassembly goes through the same parser the stream path always used. + let mut body = Vec::new(); + crate::encode_frame_body(&mut body, crate::MSG_DATA, 42, &[9, 8, 7]); + + let mut framed = Vec::new(); + crate::encode_frame(&mut framed, crate::MSG_DATA, 42, &[9, 8, 7]); + assert_eq!(&framed[4..], &body[..], "body is the frame minus its prefix"); + + let (msg_type, seq, payload) = crate::decode_frame(&body).expect("body parses"); + assert_eq!(msg_type, crate::MSG_DATA); + assert_eq!(seq, 42); + assert_eq!(payload, &[9, 8, 7]); + } + + #[test] + fn sequence_order_survives_wraparound() { + assert!(seq_older_than(1, 2)); + assert!(!seq_older_than(2, 1)); + assert!(!seq_older_than(5, 5)); + + // Across the wrap, the shorter way round wins. + assert!(seq_older_than(65535, 0)); + assert!(seq_older_than(65530, 4)); + assert!(!seq_older_than(0, 65535)); + } +} diff --git a/crates/nesprotocol/src/input.rs b/crates/nesprotocol/src/input.rs new file mode 100644 index 00000000..04c8a119 --- /dev/null +++ b/crates/nesprotocol/src/input.rs @@ -0,0 +1,317 @@ +// Input event types over the QUIC bidi stream (flow ID 2). +// Format: [1B type][fixed-size payload], batched back-to-back. + +pub const INPUT_KEY: u8 = 0; +pub const INPUT_MOUSE_MOVE: u8 = 1; +pub const INPUT_MOUSE_BUTTON: u8 = 2; +pub const INPUT_MOUSE_WHEEL: u8 = 3; + +pub const INPUT_KEY_UP: u8 = 0; +pub const INPUT_KEY_DOWN: u8 = 1; + +pub const INPUT_BTN_LEFT: u8 = 0; +pub const INPUT_BTN_MIDDLE: u8 = 1; +pub const INPUT_BTN_RIGHT: u8 = 2; + +// ── Cursor update (reverse direction: nescope → hub → desktop-app) ── + +/// Cursor update wire type sent from nescope back over the IPC socket. +pub const CURSOR_UPDATE: u8 = 0x80; + +pub const CURSOR_HIDDEN: u8 = 0; +pub const CURSOR_NAMED: u8 = 1; +pub const CURSOR_SURFACE: u8 = 2; +pub const CURSOR_IMAGE: u8 = 0x81; + +/// Parsed cursor image data from a CURSOR_IMAGE wire message. +#[derive(Debug, Clone)] +pub struct CursorImageData { + pub x: f32, + pub y: f32, + pub width: u16, + pub height: u16, + pub hotspot_x: u16, + pub hotspot_y: u16, + pub rgba: Vec, +} + +/// Encode a cursor image + position update. +/// +/// Format: `[0x81][x:f32 LE][y:f32 LE][w:u16 LE][h:u16 LE][hx:u16 LE][hy:u16 LE][rgba_len:u32 LE][rgba...]` +pub fn encode_cursor_image( + buf: &mut Vec, + x: f32, + y: f32, + width: u16, + height: u16, + hotspot_x: u16, + hotspot_y: u16, + rgba: &[u8], +) { + buf.push(CURSOR_IMAGE); + buf.extend_from_slice(&x.to_le_bytes()); + buf.extend_from_slice(&y.to_le_bytes()); + buf.extend_from_slice(&width.to_le_bytes()); + buf.extend_from_slice(&height.to_le_bytes()); + buf.extend_from_slice(&hotspot_x.to_le_bytes()); + buf.extend_from_slice(&hotspot_y.to_le_bytes()); + buf.extend_from_slice(&(rgba.len() as u32).to_le_bytes()); + buf.extend_from_slice(rgba); +} + +/// Decode a CURSOR_IMAGE message from raw wire bytes (including the 0x81 type byte). +/// Returns None if the buffer is invalid or too short. +pub fn decode_cursor_image(data: &[u8]) -> Option { + if data.len() < 21 || data[0] != CURSOR_IMAGE { + return None; + } + let x = f32::from_le_bytes([data[1], data[2], data[3], data[4]]); + let y = f32::from_le_bytes([data[5], data[6], data[7], data[8]]); + let width = u16::from_le_bytes([data[9], data[10]]); + let height = u16::from_le_bytes([data[11], data[12]]); + let hotspot_x = u16::from_le_bytes([data[13], data[14]]); + let hotspot_y = u16::from_le_bytes([data[15], data[16]]); + let rgba_len = u32::from_le_bytes([data[17], data[18], data[19], data[20]]) as usize; + if data.len() < 21 + rgba_len { + return None; + } + let rgba = data[21..21 + rgba_len].to_vec(); + Some(CursorImageData { + x, + y, + width, + height, + hotspot_x, + hotspot_y, + rgba, + }) +} + +/// Encode a cursor position + visibility update. +/// +/// Format: `[0x80][x: f32 LE][y: f32 LE][status: u8]` +/// - status: `CURSOR_HIDDEN`, `CURSOR_NAMED`, or `CURSOR_SURFACE` +/// - x,y: logical coordinates in the output space +/// +/// NOTE: Custom cursor surface (RGBA pixel) transmission is not yet +/// implemented. When it is, an additional variable-length payload containing +/// width, height, stride, and RGBA/BGRA pixels will follow the fixed header. +pub fn encode_cursor_update(buf: &mut Vec, x: f32, y: f32, status: u8) { + buf.push(CURSOR_UPDATE); + buf.extend_from_slice(&x.to_le_bytes()); + buf.extend_from_slice(&y.to_le_bytes()); + buf.push(status); +} + +/// Decoded input event from the wire protocol. +#[derive(Debug, Clone)] +pub enum DecodedInput { + Key { down: bool, keycode: u16 }, + MouseMove { dx: i16, dy: i16 }, + MouseButton { button: u8, down: bool }, + MouseWheel { dx: i16, dy: i16 }, +} + +/// Decode a single input event from raw wire bytes. +/// +/// Format: `[type:1B][payload:N]` where: +/// - type=0 INPUT_KEY: payload=[down/up:1B][keycode:2B LE] +/// - type=1 INPUT_MOUSE_MOVE: payload=[dx:2B LE][dy:2B LE] +/// - type=2 INPUT_MOUSE_BUTTON: payload=[button:1B][down/up:1B] (0=left, 1=middle, 2=right) +/// - type=3 INPUT_MOUSE_WHEEL: payload=[dx:2B LE][dy:2B LE] +/// +/// Returns `None` if the buffer is too short for the given event type or the +/// type byte is unknown. +pub fn decode_input_event(data: &[u8]) -> Option { + if data.is_empty() { + return None; + } + + match data[0] { + INPUT_KEY => { + if data.len() < 4 { + return None; + } + let down = data[1] == INPUT_KEY_DOWN; + let keycode = u16::from_le_bytes([data[2], data[3]]); + Some(DecodedInput::Key { down, keycode }) + } + INPUT_MOUSE_MOVE => { + if data.len() < 5 { + return None; + } + let dx = i16::from_le_bytes([data[1], data[2]]); + let dy = i16::from_le_bytes([data[3], data[4]]); + Some(DecodedInput::MouseMove { dx, dy }) + } + INPUT_MOUSE_BUTTON => { + if data.len() < 3 { + return None; + } + let button = data[1]; + if button > 2 { + return None; + } + let down = data[2] == INPUT_KEY_DOWN; + Some(DecodedInput::MouseButton { button, down }) + } + INPUT_MOUSE_WHEEL => { + if data.len() < 5 { + return None; + } + let dx = i16::from_le_bytes([data[1], data[2]]); + let dy = i16::from_le_bytes([data[3], data[4]]); + Some(DecodedInput::MouseWheel { dx, dy }) + } + _ => None, + } +} + +/// Encode a key event: [0][down/up][keycode u16 LE] +pub fn encode_key_event(buf: &mut Vec, down: bool, linux_keycode: u16) { + buf.push(INPUT_KEY); + buf.push(if down { INPUT_KEY_DOWN } else { INPUT_KEY_UP }); + buf.extend_from_slice(&linux_keycode.to_le_bytes()); +} + +/// Encode a mouse move: [1][dx i16 LE][dy i16 LE] +pub fn encode_mouse_move(buf: &mut Vec, dx: i16, dy: i16) { + buf.push(INPUT_MOUSE_MOVE); + buf.extend_from_slice(&dx.to_le_bytes()); + buf.extend_from_slice(&dy.to_le_bytes()); +} + +/// Encode a mouse button: [2][button][down/up] +pub fn encode_mouse_button(buf: &mut Vec, button: u8, down: bool) { + buf.push(INPUT_MOUSE_BUTTON); + buf.push(button); + buf.push(if down { INPUT_KEY_DOWN } else { INPUT_KEY_UP }); +} + +/// Encode a mouse wheel: [3][dx i16 LE][dy i16 LE] +pub fn encode_mouse_wheel(buf: &mut Vec, dx: i16, dy: i16) { + buf.push(INPUT_MOUSE_WHEEL); + buf.extend_from_slice(&dx.to_le_bytes()); + buf.extend_from_slice(&dy.to_le_bytes()); +} + +/// Mapping from `KeyboardEvent.code` strings to Linux input keycodes. +/// Covers all essential gaming keys. Use `keymap_lookup(code)` to resolve. +pub fn keymap_lookup(code: &str) -> Option { + Some(match code { + // ── Letters ── + "KeyA" => 30, + "KeyB" => 48, + "KeyC" => 46, + "KeyD" => 32, + "KeyE" => 18, + "KeyF" => 33, + "KeyG" => 34, + "KeyH" => 35, + "KeyI" => 23, + "KeyJ" => 36, + "KeyK" => 37, + "KeyL" => 38, + "KeyM" => 50, + "KeyN" => 49, + "KeyO" => 24, + "KeyP" => 25, + "KeyQ" => 16, + "KeyR" => 19, + "KeyS" => 31, + "KeyT" => 20, + "KeyU" => 22, + "KeyV" => 47, + "KeyW" => 17, + "KeyX" => 45, + "KeyY" => 21, + "KeyZ" => 44, + // ── Numbers ── + "Digit0" => 11, + "Digit1" => 2, + "Digit2" => 3, + "Digit3" => 4, + "Digit4" => 5, + "Digit5" => 6, + "Digit6" => 7, + "Digit7" => 8, + "Digit8" => 9, + "Digit9" => 10, + // ── Function keys ── + "F1" => 59, + "F2" => 60, + "F3" => 61, + "F4" => 62, + "F5" => 63, + "F6" => 64, + "F7" => 65, + "F8" => 66, + "F9" => 67, + "F10" => 68, + "F11" => 87, + "F12" => 88, + // ── Navigation ── + "Escape" => 1, + "Backquote" => 41, + "Tab" => 15, + "CapsLock" => 58, + "ShiftLeft" => 42, + "ControlLeft" => 29, + "MetaLeft" => 125, + "AltLeft" => 56, + "Space" => 57, + "AltRight" => 100, + "MetaRight" => 126, + "ControlRight" => 97, + "ShiftRight" => 54, + "Enter" => 28, + "Backspace" => 14, + // ── Arrow keys ── + "ArrowUp" => 103, + "ArrowDown" => 108, + "ArrowLeft" => 105, + "ArrowRight" => 106, + // ── Editing ── + "Insert" => 110, + "Delete" => 111, + "Home" => 102, + "End" => 107, + "PageUp" => 104, + "PageDown" => 109, + // ── Numpad ── + "NumLock" => 69, + "NumpadDivide" => 98, + "NumpadMultiply" => 55, + "NumpadSubtract" => 74, + "NumpadAdd" => 78, + "NumpadEnter" => 96, + "NumpadDecimal" => 83, + "Numpad0" => 82, + "Numpad1" => 79, + "Numpad2" => 80, + "Numpad3" => 81, + "Numpad4" => 75, + "Numpad5" => 76, + "Numpad6" => 77, + "Numpad7" => 71, + "Numpad8" => 72, + "Numpad9" => 73, + // ── Symbols ── + "Minus" => 12, + "Equal" => 13, + "BracketLeft" => 26, + "BracketRight" => 27, + "Semicolon" => 39, + "Quote" => 40, + "Comma" => 51, + "Period" => 52, + "Slash" => 53, + "Backslash" => 43, + "IntlBackslash" => 86, + // ── Media ── + "PrintScreen" => 99, + "ScrollLock" => 70, + "Pause" => 119, + _ => return None, + }) +} diff --git a/crates/nesprotocol/src/lib.rs b/crates/nesprotocol/src/lib.rs new file mode 100644 index 00000000..799d14f3 --- /dev/null +++ b/crates/nesprotocol/src/lib.rs @@ -0,0 +1,224 @@ +// Wire types shared by the components that produce and consume a session's +// media: frames, audio, cursor, input and stats. One definition, so no two ends +// can drift from each other silently. + +pub mod datagram; +pub mod input; +pub mod reliable; +pub mod stats; + +pub const ALPN: &[u8] = b"/nestri/stream/1"; + +pub const IPC_MAGIC: [u8; 4] = [b'N', b'S', b'T', b'R']; + +pub const IPC_HEADER_LEN: usize = 20; + +// ── QUIC stream protocol ──────────────────────────────────────── + +/// Current protocol version; sent as the 2nd byte on every QUIC stream +/// right after the stream-type byte. Increment on breaking frame-format changes. +/// +/// v2 moved video and audio off unidirectional streams and onto QUIC datagrams +/// (see the [`datagram`] module). A v1 peer opens uni streams for media that a +/// v2 peer no longer accepts, so the two cannot interoperate. +/// +/// v3 moved keyframes back onto a unidirectional stream each, keeping delta +/// frames on datagrams (see the [`reliable`] module). A v2 receiver ignores a +/// stream type it does not know, so it would receive no keyframes at all and +/// never decode anything — the two cannot interoperate either, and the mismatch +/// is worth reporting rather than presenting as a frozen picture. +pub const STREAM_VERSION: u8 = 3; + +/// Uniform frame header overhead: [4B u32 LE frame_len] [1B type] [2B u16 LE seq] +pub const FRAME_HDR_LEN: usize = 7; + +/// Types for the type-byte inside a framed message (within a stream). +/// For single-purpose streams (video, audio) the type is redundant but included +/// for uniformity. +pub const MSG_DATA: u8 = 0; // generic data frame (video / audio) +pub const MSG_IDR_REQUEST: u8 = 0x10; // request a keyframe (desktop → hub → hudless) +pub const MSG_ENCODE_SETTINGS: u8 = 0x12; // change encoder settings (desktop → hub → hudless) +pub const MSG_INPUT_BATCH: u8 = 0xFE; // batched input events (desktop → hub) + +/// Build a frame body: `[u8 type] [u16 LE seq] [payload]`. +/// +/// This is what a datagram carries. A stream needs [`encode_frame`] instead, +/// which is the same bytes behind a length prefix — a stream has no message +/// boundaries of its own, a datagram already knows where it ends. +pub fn encode_frame_body(buf: &mut Vec, msg_type: u8, seq: u16, payload: &[u8]) { + buf.reserve(3 + payload.len()); + buf.push(msg_type); + buf.extend_from_slice(&seq.to_le_bytes()); + buf.extend_from_slice(payload); +} + +/// Build a length-prefixed frame: `[u32 LE len = type+seq+payload] [u8 type] [u16 LE seq] [payload]` +pub fn encode_frame(buf: &mut Vec, msg_type: u8, seq: u16, payload: &[u8]) { + let frame_len = 1 + 2 + payload.len(); + buf.reserve(4 + frame_len); + buf.extend_from_slice(&(frame_len as u32).to_le_bytes()); + encode_frame_body(buf, msg_type, seq, payload); +} + +/// Decode a frame from raw bytes. Returns `(msg_type, seq_num, payload_slice)`. +pub fn decode_frame(frame: &[u8]) -> Option<(u8, u16, &[u8])> { + if frame.len() < 3 { + return None; + } + let msg_type = frame[0]; + let seq = u16::from_le_bytes([frame[1], frame[2]]); + Some((msg_type, seq, &frame[3..])) +} + +// ── Stream types ──────────────────────────────────────────────── +// +// These name a media kind, not a transport. `STREAM_VIDEO` and `STREAM_AUDIO` +// still tag IPC frames on the hudless→hub unix sockets, but over QUIC their +// media now travels as datagrams; only cursor and stats still open a uni +// stream and send this as its first byte. + +pub const STREAM_VIDEO: u8 = 0; +pub const STREAM_AUDIO: u8 = 1; +pub const STREAM_CURSOR: u8 = 3; +pub const STREAM_STATS: u8 = 4; +/// One keyframe, on a stream of its own, closed after it. The exception to +/// "media travels as datagrams" — see the [`reliable`] module for why. +pub const STREAM_KEYFRAME: u8 = 5; + +// ── Bidi stream types (desktop ↔ hub over QUIC bidi) ─────────── + +pub const BIDI_INPUT: u8 = 2; // desktop → hub → nescope (input events) + +// Codec IDs +pub const CODEC_H264: u8 = 0; +pub const CODEC_H265: u8 = 1; +pub const CODEC_AV1: u8 = 2; +pub const CODEC_OPUS: u8 = 3; +pub const CODEC_KEEP: u8 = 0xFF; // "keep current" sentinel for dynamic encoder settings + +// Rate control modes (encode settings) +pub const RC_CBR: u8 = 0; +pub const RC_CQP: u8 = 1; + +// Bit-depth (encode settings) +pub const DEPTH_8: u8 = 0; +pub const DEPTH_10: u8 = 1; + +// Video flags (bitfield) +pub const FLAG_KEYFRAME: u8 = 0x01; +pub const FLAG_RECONFIG: u8 = 0x02; // stream reconfiguration (codec/bit-depth change) + +/// Encode an IPC frame into a byte buffer. +/// Format: [4B magic] [1B type] [1B codec] [1B flags] [1B reserved] [4B ts_ms LE] [4B w_h LE] [4B len LE] [N data] +pub fn encode_ipc_frame( + stream_type: u8, + codec: u8, + flags: u8, + timestamp_ms: u32, + width: u16, + height: u16, + data: &[u8], +) -> Vec { + let header_len = IPC_HEADER_LEN; + let mut buf = Vec::with_capacity(header_len + data.len()); + + buf.extend_from_slice(&IPC_MAGIC); + buf.push(stream_type); + buf.push(codec); + buf.push(flags); + buf.push(0); // reserved + buf.extend_from_slice(×tamp_ms.to_le_bytes()); + buf.extend_from_slice(&width.to_le_bytes()); + buf.extend_from_slice(&height.to_le_bytes()); + buf.extend_from_slice(&(data.len() as u32).to_le_bytes()); + buf.extend_from_slice(data); + + buf +} + +/// Decoded IPC frame. +#[derive(Debug)] +pub struct DecodedIpcFrame<'a> { + pub stream_type: u8, + pub codec: u8, + pub flags: u8, + pub timestamp_ms: u32, + pub width: u16, + pub height: u16, + pub data: &'a [u8], +} + +/// Parse a raw IPC datagram. Returns None if magic doesn't match or buffer is too short. +pub fn decode_ipc_frame(buf: &[u8]) -> Option> { + if buf.len() < IPC_HEADER_LEN { + return None; + } + if buf[..4] != IPC_MAGIC { + return None; + } + let stream_type = buf[4]; + let codec = buf[5]; + let flags = buf[6]; + let timestamp_ms = u32::from_le_bytes([buf[8], buf[9], buf[10], buf[11]]); + let width = u16::from_le_bytes([buf[12], buf[13]]); + let height = u16::from_le_bytes([buf[14], buf[15]]); + let data_len = u32::from_le_bytes([buf[16], buf[17], buf[18], buf[19]]) as usize; + + if buf.len() < IPC_HEADER_LEN + data_len { + return None; + } + + Some(DecodedIpcFrame { + stream_type, + codec, + flags, + timestamp_ms, + width, + height, + data: &buf[IPC_HEADER_LEN..IPC_HEADER_LEN + data_len], + }) +} + +/// Codec ID to human-readable name. +pub fn codec_name(codec: u8) -> &'static str { + match codec { + CODEC_H264 => "h264", + CODEC_H265 => "h265", + CODEC_AV1 => "av1", + CODEC_OPUS => "opus", + _ => "unknown", + } +} + +/// Encode an encode-settings change into a frame payload: +/// `[1B codec_id] [1B rate_control_mode] [4B value LE] [1B bit_depth]` +pub fn encode_encode_settings( + buf: &mut Vec, + codec_id: u8, + rate_control: u8, + value: u32, + bit_depth: u8, +) { + buf.reserve(7); + buf.push(codec_id); + buf.push(rate_control); + buf.extend_from_slice(&value.to_le_bytes()); + buf.push(bit_depth); +} + +/// Decode an encode-settings payload. Returns `(codec_id, rate_control_mode, value, bit_depth)`. +/// bit_depth is None for 6-byte (old client) payloads, Some(n) for 7+ byte payloads. +pub fn decode_encode_settings(payload: &[u8]) -> Option<(u8, u8, u32, Option)> { + if payload.len() < 6 { + return None; + } + let codec_id = payload[0]; + let rc = payload[1]; + let value = u32::from_le_bytes([payload[2], payload[3], payload[4], payload[5]]); + let depth = if payload.len() >= 7 { + Some(payload[6]) + } else { + None + }; + Some((codec_id, rc, value, depth)) +} diff --git a/crates/nesprotocol/src/reliable.rs b/crates/nesprotocol/src/reliable.rs new file mode 100644 index 00000000..00768f3e --- /dev/null +++ b/crates/nesprotocol/src/reliable.rs @@ -0,0 +1,183 @@ +//! Keyframes on their own reliable QUIC stream. +//! +//! Media travels as datagrams (see the [`datagram`] module for why), which is +//! right for a frame whose worth expires in 16ms and wrong for the one frame +//! every other frame depends on. +//! +//! # Why keyframes are the exception +//! +//! Fragmentation is what makes the asymmetry brutal. A lost datagram costs the +//! whole frame it belonged to, so a frame's chance of arriving falls off with +//! its size. At 1.4% datagram loss — unremarkable for wifi — and a ~993 byte +//! payload: +//! +//! | | fragments | arrives intact | +//! | --- | --- | --- | +//! | inter frame (7.7 KB) | 8 | 89% | +//! | 1080p keyframe (~150 KB) | 155 | **12%** | +//! +//! The largest and most important object on the wire is the one least likely to +//! survive it. And because every later frame predicts from the keyframe, missing +//! one does not cost one frame: the picture stays frozen until the next keyframe +//! gets lucky. At a two second interval, 12% survival means a viewer can sit +//! frozen for tens of seconds on a link that is otherwise fine. +//! +//! Resending on the same path does not help — the retry is refragmented into the +//! same 155 datagrams with the same 12% chance — and a keyframe alone is not +//! enough to resume anyway, since the deltas that followed it reference frames +//! the receiver never got. So the fix is not a resend protocol but a change of +//! transport for the one frame that warrants it. +//! +//! # The shape of it +//! +//! A keyframe goes out on its own short-lived unidirectional stream: +//! +//! ```text +//! [1B STREAM_KEYFRAME] [1B STREAM_VERSION] [frame body] +//! ``` +//! +//! The body is byte-for-byte what the datagram path would have carried — the +//! same [`encode_frame_body`] output the fragments reassemble into — so the +//! receiver parses one layout no matter how a frame arrived. It needs no length +//! prefix either: the stream carries exactly one frame and is then finished, so +//! the stream boundary is the message boundary, just as a datagram's length is. +//! +//! QUIC retransmits stream data on its own, so there is no FEC scheme and no +//! NACK protocol to design. The head-of-line blocking that makes per-frame +//! streams unworkable is confined to that one keyframe's stream, while deltas +//! keep flowing on datagrams underneath it. Keyframe survival goes from 12% to +//! effectively 100%, and the cost is that a keyframe under loss is *late* by a +//! retransmit round trip rather than *absent*. +//! +//! # What the receiver has to know +//! +//! That lateness is the part worth planning for. A reliable stream is slower +//! than the datagrams around it — it waits for retransmission, and for flow +//! control — so a keyframe routinely arrives *after* deltas that follow it in +//! sequence. A receiver that discards it as stale, the way a late delta is +//! rightly discarded, would throw away the one frame that can end the freeze. +//! A keyframe is a random-access point, so it is never too late to be useful: +//! release it out of order and let the decoder resynchronise on it. +//! +//! [`datagram`]: crate::datagram +//! [`encode_frame_body`]: crate::encode_frame_body + +use crate::{FLAG_KEYFRAME, FLAG_RECONFIG}; + +/// Ceiling on a keyframe body read from a stream. +/// +/// A stream carries no length of its own, so the receiver needs a bound before +/// it starts reading. Generous next to any real keyframe (a 4K one runs to a few +/// hundred KB) and far below anything that would matter as an allocation. +pub const MAX_KEYFRAME_BODY: usize = 8 * 1024 * 1024; + +/// How many keyframe streams one sender may have in flight at once. +/// +/// Deliberately small. At a two second keyframe interval, needing more than a +/// couple at once means the receiver is not keeping up at all, and piling on +/// more data is the wrong response — the excess falls back to datagrams +/// instead. +pub const MAX_KEYFRAME_STREAMS_IN_FLIGHT: usize = 2; + +/// Offset of the flags byte in a video frame payload. +/// +/// The payload the hub broadcasts is `[1B codec] [1B flags] [4B ts_ms LE] +/// [2B width LE] [2B height LE] [data]` — see `encode_ipc_frame`, whose header +/// fields these mirror. +pub const VIDEO_PAYLOAD_FLAGS_OFFSET: usize = 1; + +/// Whether a video frame payload warrants a reliable stream over datagrams. +/// +/// True for a keyframe, and for the reconfiguration frame that follows a codec +/// or bit-depth change: that one carries the parameter sets every frame after it +/// is decoded against, so losing it costs exactly as much as losing a keyframe. +/// +/// Only ask this of a *video* payload. Audio has no keyframes, and at this +/// offset an audio packet holds unrelated bytes, so the answer would be +/// meaningless roughly half the time. +pub fn video_wants_reliable(payload: &[u8]) -> bool { + payload + .get(VIDEO_PAYLOAD_FLAGS_OFFSET) + .is_some_and(|flags| flags & (FLAG_KEYFRAME | FLAG_RECONFIG) != 0) +} + +/// Whether a video frame payload is a keyframe proper: a point the decoder can +/// start from cold. +/// +/// Narrower than [`video_wants_reliable`] on purpose. That one answers a +/// transport question — is this frame worth a reliable stream — and a +/// reconfiguration frame qualifies because losing its parameter sets costs as +/// much as losing a keyframe. This one answers a decoding question, and only a +/// keyframe is an answer to it: a receiver may release a keyframe out of order +/// knowing the decoder will resynchronise, which is not something to assume of +/// a frame merely because it arrived reliably. +pub fn video_is_keyframe(payload: &[u8]) -> bool { + payload + .get(VIDEO_PAYLOAD_FLAGS_OFFSET) + .is_some_and(|flags| flags & FLAG_KEYFRAME != 0) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{CODEC_H264, STREAM_VIDEO, decode_ipc_frame, encode_ipc_frame}; + + /// Build the payload the hub broadcasts out of an IPC frame, the way + /// `ipc_listener` does, so the offset is checked against its real producer + /// rather than against a hand-written copy of the layout. + fn broadcast_payload(flags: u8) -> Vec { + let ipc = encode_ipc_frame( + STREAM_VIDEO, + CODEC_H264, + flags, + 1234, + 1920, + 1080, + &[9, 9, 9], + ); + let decoded = decode_ipc_frame(&ipc).expect("valid ipc frame"); + let mut payload = vec![decoded.codec, decoded.flags]; + payload.extend_from_slice(&decoded.timestamp_ms.to_le_bytes()); + payload.extend_from_slice(&decoded.width.to_le_bytes()); + payload.extend_from_slice(&decoded.height.to_le_bytes()); + payload.extend_from_slice(decoded.data); + payload + } + + #[test] + fn keyframes_and_reconfigs_want_a_stream() { + assert!(video_wants_reliable(&broadcast_payload(FLAG_KEYFRAME))); + assert!(video_wants_reliable(&broadcast_payload(FLAG_RECONFIG))); + assert!(video_wants_reliable(&broadcast_payload( + FLAG_KEYFRAME | FLAG_RECONFIG + ))); + } + + #[test] + fn delta_frames_stay_on_datagrams() { + assert!(!video_wants_reliable(&broadcast_payload(0))); + // An unrelated flag in the same byte must not promote a delta frame. + assert!(!video_wants_reliable(&broadcast_payload(0x80))); + } + + #[test] + fn a_reconfig_alone_is_worth_a_stream_but_is_not_a_resync_point() { + // The transport question and the decoding question have different + // answers for this frame, which is the reason for two predicates. + let reconfig = broadcast_payload(FLAG_RECONFIG); + assert!(video_wants_reliable(&reconfig)); + assert!(!video_is_keyframe(&reconfig)); + + let both = broadcast_payload(FLAG_KEYFRAME | FLAG_RECONFIG); + assert!(video_wants_reliable(&both)); + assert!(video_is_keyframe(&both)); + } + + #[test] + fn a_payload_too_short_to_have_flags_is_not_a_keyframe() { + assert!(!video_wants_reliable(&[])); + assert!(!video_wants_reliable(&[CODEC_H264])); + assert!(!video_is_keyframe(&[])); + assert!(!video_is_keyframe(&[CODEC_H264])); + } +} diff --git a/crates/nesprotocol/src/stats.rs b/crates/nesprotocol/src/stats.rs new file mode 100644 index 00000000..c5a8847a --- /dev/null +++ b/crates/nesprotocol/src/stats.rs @@ -0,0 +1,116 @@ +// Pipeline stats messages sent from each component to the desktop-app +// via the hub. Each source sends its own fixed-size packet. + +pub const STATS_NESCOPE: u8 = 0; +pub const STATS_HUDLESS: u8 = 1; +pub const STATS_HUB: u8 = 2; + +/// Nescope stats: game frame callback rate. +/// [0][game_fps: u8][frame_count: u32 LE] +pub fn encode_nescope_stats(buf: &mut Vec, game_fps: u8, frame_count: u32) { + buf.push(STATS_NESCOPE); + buf.push(game_fps); + buf.extend_from_slice(&frame_count.to_le_bytes()); +} + +/// Hudless stats: capture FPS, encode time, dropped frames, diagnostics, capture latency. +/// [1][capture_fps: u8][encode_avg_ms: f32 LE][dropped: u32 LE][present_attempts: u32 LE][capture_attempts: u32 LE][capture_ms: f32 LE] +pub fn encode_hudless_stats( + buf: &mut Vec, + capture_fps: u8, + encode_avg_ms: f32, + dropped: u32, + present_attempts: u32, + capture_attempts: u32, + capture_ms: f32, +) { + buf.push(STATS_HUDLESS); + buf.push(capture_fps); + buf.extend_from_slice(&encode_avg_ms.to_le_bytes()); + buf.extend_from_slice(&dropped.to_le_bytes()); + buf.extend_from_slice(&present_attempts.to_le_bytes()); + buf.extend_from_slice(&capture_attempts.to_le_bytes()); + buf.extend_from_slice(&capture_ms.to_le_bytes()); +} + +/// Hub stats: client count, video bytes, relay latency, audio. +/// +/// `audio_bitrate_kbps` is *measured* ingest from neswire, not the hub's +/// configured target. It used to be the latter, which made it a constant -- +/// it read the same whether neswire was feeding the socket or had never sent +/// a byte. `audio_channels` stays configuration: a byte count cannot tell you +/// how many channels those bytes describe. +/// +/// The layout is unchanged, so this is not a version bump -- only the meaning +/// of a field that was never trustworthy in the first place. +/// [2][clients: u8][video_bytes_mb: u32 LE][hub_relay_ms: f32 LE][audio_bitrate_kbps: u32 LE][audio_channels: u8] +pub fn encode_hub_stats( + buf: &mut Vec, + clients: u8, + video_bytes_mb: u32, + hub_relay_ms: f32, + audio_bitrate_kbps: u32, + audio_channels: u8, +) { + buf.push(STATS_HUB); + buf.push(clients); + buf.extend_from_slice(&video_bytes_mb.to_le_bytes()); + buf.extend_from_slice(&hub_relay_ms.to_le_bytes()); + buf.extend_from_slice(&audio_bitrate_kbps.to_le_bytes()); + buf.push(audio_channels); +} + +/// Decoded stats from any source. +#[derive(Debug, Clone, Default)] +pub struct PipelineStats { + pub nescope_fps: u8, + pub nescope_frames: u32, + pub hudless_fps: u8, + pub hudless_encode_ms: f32, + pub hudless_capture_ms: f32, + pub hudless_dropped: u32, + pub hub_clients: u8, + pub hub_video_mb: u32, + pub hub_relay_ms: f32, + pub present_attempts: u32, + pub capture_attempts: u32, + pub audio_bitrate_kbps: u32, + pub audio_channels: u8, +} + +/// Try to decode a single stats packet. The `msg_type` is the frame-level +/// type byte (STATS_NESCOPE, STATS_HUDLESS, or STATS_HUB). +pub fn decode_stats(msg_type: u8, data: &[u8], stats: &mut PipelineStats) { + match msg_type { + STATS_NESCOPE if data.len() >= 5 => { + stats.nescope_fps = data[0]; + stats.nescope_frames = u32::from_le_bytes([data[1], data[2], data[3], data[4]]); + } + STATS_HUDLESS if data.len() >= 17 => { + stats.hudless_fps = data[0]; + stats.hudless_encode_ms = f32::from_le_bytes([data[1], data[2], data[3], data[4]]); + stats.hudless_dropped = u32::from_le_bytes([data[5], data[6], data[7], data[8]]); + stats.present_attempts = u32::from_le_bytes([data[9], data[10], data[11], data[12]]); + stats.capture_attempts = u32::from_le_bytes([data[13], data[14], data[15], data[16]]); + if data.len() >= 21 { + stats.hudless_capture_ms = + f32::from_le_bytes([data[17], data[18], data[19], data[20]]); + } + } + STATS_HUB if data.len() >= 5 => { + stats.hub_clients = data[0]; + stats.hub_video_mb = u32::from_le_bytes([data[1], data[2], data[3], data[4]]); + if data.len() >= 9 { + stats.hub_relay_ms = f32::from_le_bytes([data[5], data[6], data[7], data[8]]); + } + if data.len() >= 13 { + stats.audio_bitrate_kbps = + u32::from_le_bytes([data[9], data[10], data[11], data[12]]); + } + if data.len() >= 14 { + stats.audio_channels = data[13]; + } + } + _ => {} + } +}