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]; + } + } + _ => {} + } +}