feat(hyperhdr): add the protocol codec and the network thread

proto encodes registration, image and clear requests with the
[u32 big-endian length][flatbuffer] framing and decodes the replies. client
drives them from a thread: it connects while frames flow, applies the pacer,
sends keepalives and reconnects with backoff.

src/hyperhdr/client.rs

@@ -0,0 +1,508 @@
+//! Network thread: connects to HyperHDR while frames are available, paces them, keeps the
+//! connection alive and reconnects with backoff. Blocking `std::net` I/O with timeouts.
+
+use std::io::{self, Write};
+use std::net::{Shutdown, SocketAddr, TcpStream, ToSocketAddrs};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+use std::thread::{self, JoinHandle};
+use std::time::{Duration, Instant};
+
+use tracing::{debug, info, info_span, warn};
+
+use super::proto::{self, Encoder};
+use crate::config::{Config, HYPERHDR_MAX_MESSAGE};
+use crate::pacing::{Action, Backoff, Pacer, PacingConfig, RateLimit};
+use crate::slot::{Frame, FrameSlot};
+use crate::stats::{self, Stats};
+
+/// A connection closed this quickly, before any reply, was most likely rejected by HyperHDR.
+const REJECT_WINDOW: Duration = Duration::from_secs(1);
+
+#[derive(Debug, Clone)]
+pub struct ClientConfig {
+    pub host: String,
+    pub port: u16,
+    pub priority: i32,
+    pub origin: String,
+    pub pacing: PacingConfig,
+    pub connect_timeout: Duration,
+    pub write_timeout: Duration,
+    pub backoff_initial: Duration,
+    pub backoff_max: Duration,
+    pub backoff_reset_after: Duration,
+    /// Minimum interval between logged error replies and rejection hints.
+    pub error_log_interval: Duration,
+    pub stats_interval: Option<Duration>,
+    /// Bound for sending the final `Clear` and for waiting until the server closes.
+    pub shutdown_timeout: Duration,
+}
+
+impl ClientConfig {
+    /// Daemon defaults for the given server.
+    pub fn new(host: impl Into<String>, port: u16) -> Self {
+        Self {
+            host: host.into(),
+            port,
+            priority: 150,
+            origin: "gamescope".to_owned(),
+            pacing: PacingConfig {
+                min_interval: Duration::from_secs(1) / 30,
+                keepalive: Duration::from_secs(1),
+                idle_clear: None,
+            },
+            connect_timeout: Duration::from_secs(2),
+            write_timeout: Duration::from_secs(2),
+            backoff_initial: Duration::from_millis(500),
+            backoff_max: Duration::from_secs(10),
+            backoff_reset_after: Duration::from_secs(30),
+            error_log_interval: Duration::from_secs(10),
+            stats_interval: Some(Duration::from_secs(60)),
+            shutdown_timeout: Duration::from_millis(400),
+        }
+    }
+
+    pub fn from_config(cfg: &Config) -> Self {
+        Self {
+            priority: cfg.priority,
+            origin: cfg.origin.clone(),
+            pacing: PacingConfig {
+                min_interval: cfg.min_send_interval(),
+                keepalive: cfg.keepalive(),
+                idle_clear: cfg.idle_clear(),
+            },
+            connect_timeout: cfg.connect_timeout(),
+            write_timeout: cfg.write_timeout(),
+            ..Self::new(cfg.host.clone(), cfg.port)
+        }
+    }
+}
+
+/// Starts the network thread. It runs until [`FrameSlot::shutdown`].
+pub fn spawn(
+    cfg: ClientConfig,
+    slot: Arc<FrameSlot>,
+    stats: Arc<Stats>,
+) -> io::Result<JoinHandle<()>> {
+    thread::Builder::new()
+        .name("hyperhdr".into())
+        .spawn(move || Client::new(cfg, slot, stats).run())
+}
+
+#[derive(Debug, Default)]
+struct ConnShared {
+    closed: AtomicBool,
+    got_reply: AtomicBool,
+    reason: Mutex<Option<String>>,
+}
+
+struct Connection {
+    stream: TcpStream,
+    peer: SocketAddr,
+    connected_at: Instant,
+    shared: Arc<ConnShared>,
+    reader: Option<JoinHandle<()>>,
+}
+
+impl Connection {
+    fn open(cfg: &ClientConfig, slot: &Arc<FrameSlot>, stats: &Arc<Stats>) -> io::Result<Self> {
+        let mut last_err = None;
+        for addr in (cfg.host.as_str(), cfg.port).to_socket_addrs()? {
+            match TcpStream::connect_timeout(&addr, cfg.connect_timeout) {
+                Ok(stream) => return Self::start(stream, addr, cfg, slot, stats),
+                Err(e) => last_err = Some(e),
+            }
+        }
+        Err(last_err.unwrap_or_else(|| {
+            io::Error::new(
+                io::ErrorKind::NotFound,
+                format!("{} resolved to no addresses", cfg.host),
+            )
+        }))
+    }
+
+    fn start(
+        stream: TcpStream,
+        peer: SocketAddr,
+        cfg: &ClientConfig,
+        slot: &Arc<FrameSlot>,
+        stats: &Arc<Stats>,
+    ) -> io::Result<Self> {
+        stream.set_nodelay(true)?;
+        stream.set_write_timeout(Some(cfg.write_timeout))?;
+        let read_half = stream.try_clone()?;
+        let shared = Arc::new(ConnShared::default());
+        let reader = {
+            let (shared, slot, stats) = (shared.clone(), slot.clone(), stats.clone());
+            let (priority, interval) = (cfg.priority, cfg.error_log_interval);
+            let span = tracing::Span::current();
+            thread::Builder::new()
+                .name("hyperhdr-reader".into())
+                .spawn(move || {
+                    let _enter = span.enter();
+                    read_replies(read_half, &shared, &stats, priority, interval);
+                    slot.wake();
+                })?
+        };
+        Ok(Self {
+            stream,
+            peer,
+            connected_at: Instant::now(),
+            shared,
+            reader: Some(reader),
+        })
+    }
+
+    fn is_closed(&self) -> bool {
+        self.shared.closed.load(Ordering::Relaxed)
+    }
+
+    fn close_reason(&self) -> String {
+        let reason = self.shared.reason.lock().unwrap_or_else(|e| e.into_inner());
+        reason
+            .clone()
+            .unwrap_or_else(|| "connection closed".to_owned())
+    }
+
+    /// Closes both directions immediately and reaps the reader.
+    fn abort(mut self) {
+        let _ = self.stream.shutdown(Shutdown::Both);
+        if let Some(reader) = self.reader.take() {
+            let _ = reader.join();
+        }
+    }
+
+    /// Sends FIN and lets the reader drain replies until the server closes or `timeout`.
+    /// Closing with unread replies would send RST, which may discard our last message.
+    fn finish(self, timeout: Duration) {
+        let _ = self.stream.shutdown(Shutdown::Write);
+        let deadline = Instant::now() + timeout;
+        while self.reader.as_ref().is_some_and(|r| !r.is_finished()) && Instant::now() < deadline {
+            thread::sleep(Duration::from_millis(5));
+        }
+        self.abort();
+    }
+}
+
+fn read_replies(
+    mut stream: TcpStream,
+    shared: &ConnShared,
+    stats: &Stats,
+    priority: i32,
+    error_log_interval: Duration,
+) {
+    let mut buf = Vec::with_capacity(64);
+    let mut error_log = RateLimit::new(error_log_interval);
+    let mut registered = false;
+    let reason = loop {
+        match proto::read_message(&mut stream, &mut buf, HYPERHDR_MAX_MESSAGE) {
+            Ok(true) => {}
+            Ok(false) => break "connection closed by HyperHDR".to_owned(),
+            Err(e) => break format!("read failed: {e}"),
+        }
+        shared.got_reply.store(true, Ordering::Relaxed);
+        match proto::decode_reply(&buf) {
+            Ok(reply) => {
+                if let Some(error) = reply.error {
+                    stats::inc(&stats.error_replies);
+                    if let Some(suppressed) = error_log.check(Instant::now()) {
+                        warn!(suppressed, "HyperHDR replied with an error: {error}");
+                    }
+                } else if reply.registered != -1 && !registered {
+                    registered = true;
+                    if reply.registered == priority {
+                        info!("registered with priority {priority}");
+                    } else {
+                        warn!(
+                            "HyperHDR reports priority {} instead of {priority}",
+                            reply.registered
+                        );
+                    }
+                }
+            }
+            Err(e) => {
+                if let Some(suppressed) = error_log.check(Instant::now()) {
+                    warn!(suppressed, "undecodable reply from HyperHDR: {e}");
+                }
+            }
+        }
+    };
+    *shared.reason.lock().unwrap_or_else(|e| e.into_inner()) = Some(reason);
+    shared.closed.store(true, Ordering::Relaxed);
+}
+
+struct Client {
+    cfg: ClientConfig,
+    slot: Arc<FrameSlot>,
+    stats: Arc<Stats>,
+    pacer: Pacer,
+    backoff: Backoff,
+    encoder: Encoder,
+    frame: Frame,
+    conn: Option<Connection>,
+    retry_at: Option<Instant>,
+    /// The next successful connection is a reconnect (after a failure).
+    reconnecting: bool,
+    failed_attempts: u32,
+    last_connect_error: Option<String>,
+    reject_log: RateLimit,
+    next_stats: Option<Instant>,
+}
+
+impl Client {
+    fn new(cfg: ClientConfig, slot: Arc<FrameSlot>, stats: Arc<Stats>) -> Self {
+        Self {
+            pacer: Pacer::new(cfg.pacing),
+            backoff: Backoff::new(
+                cfg.backoff_initial,
+                cfg.backoff_max,
+                cfg.backoff_reset_after,
+            ),
+            encoder: Encoder::new(),
+            frame: Frame::default(),
+            conn: None,
+            retry_at: None,
+            reconnecting: false,
+            failed_attempts: 0,
+            last_connect_error: None,
+            reject_log: RateLimit::new(cfg.error_log_interval.max(Duration::from_secs(60))),
+            next_stats: cfg.stats_interval.map(|i| Instant::now() + i),
+            cfg,
+            slot,
+            stats,
+        }
+    }
+
+    fn run(mut self) {
+        let server = format!("{}:{}", self.cfg.host, self.cfg.port);
+        let span = info_span!("hyperhdr", server = %server);
+        let _enter = span.enter();
+
+        loop {
+            let deadline = self.next_deadline(Instant::now());
+            let update = self.slot.wait(&mut self.frame, deadline);
+            let now = Instant::now();
+
+            if update.shutdown {
+                if let Some(conn) = self.conn.take() {
+                    self.close_gracefully(conn, "shutting down");
+                }
+                return;
+            }
+            if update.source_gone {
+                self.on_source_gone();
+            }
+            if update.new_frame {
+                let was_cleared = self.pacer.is_cleared();
+                if self.pacer.on_new_frame(now) {
+                    stats::inc(&self.stats.superseded);
+                }
+                if was_cleared {
+                    info!("new frames after idle, resuming");
+                }
+            }
+            if let Some(conn) = self.conn.as_ref().filter(|c| c.is_closed()) {
+                let reason = conn.close_reason();
+                self.on_connection_lost(reason, now);
+            }
+            self.maybe_connect(now);
+            self.pump(now);
+            self.maybe_log_stats(now);
+        }
+    }
+
+    fn next_deadline(&self, now: Instant) -> Option<Instant> {
+        let deadline = match self.conn {
+            Some(_) => match self.pacer.poll(now) {
+                Action::Wait(at) => at,
+                _ => Some(now),
+            },
+            None => self.retry_at,
+        };
+        match (deadline, self.next_stats) {
+            (Some(a), Some(b)) => Some(a.min(b)),
+            (a, b) => a.or(b),
+        }
+    }
+
+    fn on_source_gone(&mut self) {
+        self.pacer.on_source_gone();
+        if let Some(conn) = self.conn.take() {
+            self.close_gracefully(conn, "source gone");
+        }
+        self.retry_at = None;
+        self.reconnecting = false;
+        self.failed_attempts = 0;
+        self.last_connect_error = None;
+        self.backoff.reset();
+    }
+
+    /// Clears our priority and closes the connection cleanly.
+    fn close_gracefully(&mut self, mut conn: Connection, reason: &str) {
+        let timeout = self.cfg.shutdown_timeout;
+        let _ = conn.stream.set_write_timeout(Some(timeout));
+        let msg = self.encoder.clear(self.cfg.priority);
+        match conn
+            .stream
+            .write_all(msg)
+            .and_then(|()| conn.stream.flush())
+        {
+            Ok(()) => {
+                stats::add(&self.stats.bytes_sent, msg.len() as u64);
+                info!(
+                    priority = self.cfg.priority,
+                    "{reason}: sent Clear, disconnecting"
+                );
+            }
+            Err(e) => warn!("{reason}: could not send Clear ({e}), disconnecting"),
+        }
+        conn.finish(timeout);
+    }
+
+    fn maybe_connect(&mut self, now: Instant) {
+        if self.conn.is_some() || !self.pacer.wants_connection(now) {
+            return;
+        }
+        if self.retry_at.is_some_and(|at| now < at) {
+            return;
+        }
+        match Connection::open(&self.cfg, &self.slot, &self.stats) {
+            Ok(mut conn) => {
+                if self.failed_attempts > 0 {
+                    info!(peer = %conn.peer, "connected after {} failed attempts", self.failed_attempts);
+                } else {
+                    info!(peer = %conn.peer, "connected");
+                }
+                if self.reconnecting {
+                    stats::inc(&self.stats.reconnects);
+                }
+                self.reconnecting = false;
+                self.failed_attempts = 0;
+                self.last_connect_error = None;
+                self.retry_at = None;
+
+                let msg = self.encoder.register(&self.cfg.origin, self.cfg.priority);
+                let result = conn.stream.write_all(msg);
+                let sent = msg.len() as u64;
+                self.conn = Some(conn);
+                match result {
+                    Ok(()) => {
+                        stats::add(&self.stats.bytes_sent, sent);
+                        self.pacer.on_connected();
+                    }
+                    Err(e) => self.on_connection_lost(
+                        format!("sending Register failed: {e}"),
+                        Instant::now(),
+                    ),
+                }
+            }
+            Err(e) => {
+                self.failed_attempts += 1;
+                self.reconnecting = true;
+                let delay = self.backoff.on_failure(None);
+                self.retry_at = Some(now + delay);
+                let error = e.to_string();
+                if self.last_connect_error.as_deref() == Some(error.as_str()) {
+                    debug!(
+                        attempt = self.failed_attempts,
+                        "cannot connect: {error}; next attempt in {delay:?}"
+                    );
+                } else {
+                    warn!(
+                        "cannot connect: {error}; retrying with backoff, next attempt in {delay:?}"
+                    );
+                    self.last_connect_error = Some(error);
+                }
+            }
+        }
+    }
+
+    fn on_connection_lost(&mut self, reason: String, now: Instant) {
+        let Some(conn) = self.conn.take() else {
+            return;
+        };
+        let session = now.saturating_duration_since(conn.connected_at);
+        let got_reply = conn.shared.got_reply.load(Ordering::Relaxed);
+        conn.abort();
+
+        let delay = self.backoff.on_failure(Some(session));
+        self.retry_at = Some(now + delay);
+        self.reconnecting = true;
+
+        if session < REJECT_WINDOW && !got_reply {
+            match self.reject_log.check(now) {
+                Some(suppressed) => warn!(
+                    suppressed,
+                    "HyperHDR closed the connection right after connecting ({reason}). HyperHDR only \
+                     accepts clients from its own /24 subnet unless their IP is whitelisted \
+                     (Network Services) or internet access is enabled; see the HyperHDR log for \
+                     \"not whitelisted, access denied\". Retrying in {delay:?}"
+                ),
+                None => debug!("connection closed immediately again ({reason}); retrying in {delay:?}"),
+            }
+        } else {
+            warn!("disconnected after {session:.1?}: {reason}; reconnecting in {delay:?}");
+        }
+    }
+
+    fn pump(&mut self, now: Instant) {
+        loop {
+            let Some(conn) = self.conn.as_mut() else {
+                return;
+            };
+            let action = self.pacer.poll(now);
+            let msg = match action {
+                Action::Wait(_) => return,
+                Action::SendFrame | Action::SendKeepalive => {
+                    self.encoder
+                        .image(self.frame.width, self.frame.height, &self.frame.rgb)
+                }
+                Action::Clear => self.encoder.clear(self.cfg.priority),
+                Action::Heartbeat => self.encoder.register(&self.cfg.origin, self.cfg.priority),
+            };
+            match conn.stream.write_all(msg) {
+                Ok(()) => {
+                    stats::add(&self.stats.bytes_sent, msg.len() as u64);
+                    match action {
+                        Action::SendFrame => stats::inc(&self.stats.frames_sent),
+                        Action::SendKeepalive => stats::inc(&self.stats.keepalives),
+                        Action::Clear => info!(
+                            priority = self.cfg.priority,
+                            "no new frames for {:?}, cleared priority",
+                            self.cfg.pacing.idle_clear.unwrap_or_default()
+                        ),
+                        _ => {}
+                    }
+                    self.pacer.on_sent(action, now);
+                }
+                Err(e) => {
+                    let reason = if matches!(
+                        e.kind(),
+                        io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
+                    ) {
+                        format!(
+                            "write timed out after {:?} (HyperHDR is not reading)",
+                            self.cfg.write_timeout
+                        )
+                    } else {
+                        format!("write failed: {e}")
+                    };
+                    self.on_connection_lost(reason, Instant::now());
+                    return;
+                }
+            }
+        }
+    }
+
+    fn maybe_log_stats(&mut self, now: Instant) {
+        let (Some(interval), Some(next)) = (self.cfg.stats_interval, self.next_stats) else {
+            return;
+        };
+        if now < next {
+            return;
+        }
+        self.next_stats = Some(now + interval);
+        let snapshot = self.stats.take();
+        info!("stats for the last {interval:?}: {snapshot}");
+    }
+}

src/hyperhdr/mod.rs

@@ -1,3 +1,7 @@
 //! Client for the HyperHDR flatbuffers server.
 
+pub mod client;
 pub mod generated;
+pub mod proto;
+
+pub use client::ClientConfig;

src/hyperhdr/proto.rs

@@ -0,0 +1,374 @@
+//! HyperHDR flatbuffers messages and their `[u32 big-endian length][flatbuffer]` framing.
+
+use std::io::{self, Read};
+
+use flatbuffers::{FlatBufferBuilder, InvalidFlatbuffer, UnionWIPOffset, WIPOffset};
+
+use super::generated::{reply, request};
+
+/// Builds framed requests, reusing one builder.
+pub struct Encoder {
+    fbb: FlatBufferBuilder<'static>,
+}
+
+impl Default for Encoder {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Encoder {
+    pub fn new() -> Self {
+        Self {
+            fbb: FlatBufferBuilder::with_capacity(1024),
+        }
+    }
+
+    pub fn register(&mut self, origin: &str, priority: i32) -> &[u8] {
+        self.fbb.reset();
+        let origin = self.fbb.create_string(origin);
+        let register = request::Register::create(
+            &mut self.fbb,
+            &request::RegisterArgs {
+                origin: Some(origin),
+                priority,
+            },
+        );
+        self.finish(request::Command::Register, register.as_union_value())
+    }
+
+    /// `Image{RawImage}` with `duration = -1` (shown until cleared or disconnected).
+    pub fn image(&mut self, width: u32, height: u32, rgb: &[u8]) -> &[u8] {
+        self.fbb.reset();
+        let data = self.fbb.create_vector(rgb);
+        let raw = request::RawImage::create(
+            &mut self.fbb,
+            &request::RawImageArgs {
+                data: Some(data),
+                width: i32::try_from(width).unwrap_or(i32::MAX),
+                height: i32::try_from(height).unwrap_or(i32::MAX),
+            },
+        );
+        let image = request::Image::create(
+            &mut self.fbb,
+            &request::ImageArgs {
+                data_type: request::ImageType::RawImage,
+                data: Some(raw.as_union_value()),
+                duration: -1,
+            },
+        );
+        self.finish(request::Command::Image, image.as_union_value())
+    }
+
+    pub fn clear(&mut self, priority: i32) -> &[u8] {
+        self.fbb.reset();
+        let clear = request::Clear::create(&mut self.fbb, &request::ClearArgs { priority });
+        self.finish(request::Command::Clear, clear.as_union_value())
+    }
+
+    fn finish(
+        &mut self,
+        command_type: request::Command,
+        command: WIPOffset<UnionWIPOffset>,
+    ) -> &[u8] {
+        let req = request::Request::create(
+            &mut self.fbb,
+            &request::RequestArgs {
+                command_type,
+                command: Some(command),
+            },
+        );
+        self.fbb.finish_size_prefixed(req, None);
+        // flatbuffers writes a little-endian length prefix; HyperHDR expects big-endian.
+        let (buf, start) = self.fbb.mut_finished_buffer();
+        let message = &mut buf[start..];
+        let (prefix, _) = message.split_at_mut(4);
+        let len = u32::from_le_bytes([prefix[0], prefix[1], prefix[2], prefix[3]]);
+        prefix.copy_from_slice(&len.to_be_bytes());
+        message
+    }
+}
+
+/// A decoded `Request` (used by the mock server and tests).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Command<'a> {
+    Register {
+        origin: &'a str,
+        priority: i32,
+    },
+    RawImage {
+        width: i32,
+        height: i32,
+        duration: i32,
+        data: &'a [u8],
+    },
+    Nv12Image,
+    ImageWithoutData,
+    Clear {
+        priority: i32,
+    },
+    Color {
+        data: i32,
+        duration: i32,
+    },
+    Unknown,
+}
+
+/// Verifies and decodes a request body (without the length prefix).
+pub fn decode_request(body: &[u8]) -> Result<Command<'_>, InvalidFlatbuffer> {
+    let req = request::root_as_request(body)?;
+    let command = if let Some(r) = req.command_as_register() {
+        Command::Register {
+            origin: r.origin(),
+            priority: r.priority(),
+        }
+    } else if let Some(image) = req.command_as_image() {
+        if let Some(raw) = image.data_as_raw_image() {
+            Command::RawImage {
+                width: raw.width(),
+                height: raw.height(),
+                duration: image.duration(),
+                data: raw.data().map_or(&[][..], |d| d.bytes()),
+            }
+        } else if image.data_as_nv12_image().is_some() {
+            Command::Nv12Image
+        } else {
+            Command::ImageWithoutData
+        }
+    } else if let Some(c) = req.command_as_clear() {
+        Command::Clear {
+            priority: c.priority(),
+        }
+    } else if let Some(c) = req.command_as_color() {
+        Command::Color {
+            data: c.data(),
+            duration: c.duration(),
+        }
+    } else {
+        Command::Unknown
+    };
+    Ok(command)
+}
+
+/// A decoded `Reply`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Reply<'a> {
+    pub error: Option<&'a str>,
+    pub video: i32,
+    pub registered: i32,
+}
+
+/// Verifies and decodes a reply body (without the length prefix).
+pub fn decode_reply(body: &[u8]) -> Result<Reply<'_>, InvalidFlatbuffer> {
+    let r = reply::root_as_reply(body)?;
+    Ok(Reply {
+        error: r.error(),
+        video: r.video(),
+        registered: r.registered(),
+    })
+}
+
+/// Encodes a framed `Reply` into `out`, the way `FlatBuffersParser` does.
+pub fn encode_reply(
+    fbb: &mut FlatBufferBuilder<'_>,
+    error: Option<&str>,
+    registered: i32,
+    out: &mut Vec<u8>,
+) {
+    fbb.reset();
+    let error = error.map(|e| fbb.create_string(e));
+    let r = reply::Reply::create(
+        fbb,
+        &reply::ReplyArgs {
+            error,
+            video: -1,
+            registered,
+        },
+    );
+    fbb.finish(r, None);
+    let data = fbb.finished_data();
+    out.clear();
+    out.extend_from_slice(&(data.len() as u32).to_be_bytes());
+    out.extend_from_slice(data);
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum FrameError {
+    #[error("invalid message length {0}")]
+    InvalidLength(u32),
+    #[error(transparent)]
+    Io(#[from] io::Error),
+}
+
+/// Reads one framed message body into `buf`. Returns `Ok(false)` on a clean EOF between messages.
+pub fn read_message(
+    r: &mut impl Read,
+    buf: &mut Vec<u8>,
+    max_len: usize,
+) -> Result<bool, FrameError> {
+    let mut header = [0u8; 4];
+    let mut filled = 0;
+    while filled < header.len() {
+        match r.read(&mut header[filled..]) {
+            Ok(0) if filled == 0 => return Ok(false),
+            Ok(0) => return Err(io::Error::from(io::ErrorKind::UnexpectedEof).into()),
+            Ok(n) => filled += n,
+            Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
+            Err(e) => return Err(e.into()),
+        }
+    }
+    let len = u32::from_be_bytes(header);
+    if len == 0 || len as usize > max_len {
+        return Err(FrameError::InvalidLength(len));
+    }
+    buf.resize(len as usize, 0);
+    r.read_exact(buf)?;
+    Ok(true)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn body(message: &[u8]) -> &[u8] {
+        let len = u32::from_be_bytes(message[..4].try_into().unwrap()) as usize;
+        assert_eq!(len, message.len() - 4, "big-endian length prefix");
+        &message[4..]
+    }
+
+    #[test]
+    fn register_roundtrip() {
+        let mut enc = Encoder::new();
+        let msg = enc.register("gamescope", 150).to_vec();
+        let req = request::root_as_request(body(&msg)).unwrap();
+        assert_eq!(req.command_type(), request::Command::Register);
+        assert_eq!(
+            decode_request(body(&msg)).unwrap(),
+            Command::Register {
+                origin: "gamescope",
+                priority: 150
+            }
+        );
+    }
+
+    #[test]
+    fn image_roundtrip() {
+        let mut enc = Encoder::new();
+        let rgb: Vec<u8> = (0..3 * 5 * 3).map(|i| i as u8).collect();
+        let msg = enc.image(5, 3, &rgb).to_vec();
+        let req = request::root_as_request(body(&msg)).unwrap();
+        assert_eq!(req.command_type(), request::Command::Image);
+        let image = req.command_as_image().unwrap();
+        assert_eq!(image.data_type(), request::ImageType::RawImage);
+        assert_eq!(
+            decode_request(body(&msg)).unwrap(),
+            Command::RawImage {
+                width: 5,
+                height: 3,
+                duration: -1,
+                data: &rgb
+            }
+        );
+    }
+
+    #[test]
+    fn clear_roundtrip() {
+        let mut enc = Encoder::new();
+        let msg = enc.clear(150).to_vec();
+        assert_eq!(
+            decode_request(body(&msg)).unwrap(),
+            Command::Clear { priority: 150 }
+        );
+    }
+
+    #[test]
+    fn encoder_reuse() {
+        let mut enc = Encoder::new();
+        let big = vec![7u8; 320 * 180 * 3];
+        let _ = enc.image(320, 180, &big);
+        let msg = enc.register("x", 99).to_vec();
+        assert_eq!(
+            decode_request(body(&msg)).unwrap(),
+            Command::Register {
+                origin: "x",
+                priority: 99
+            }
+        );
+        let msg = enc.image(320, 180, &big).to_vec();
+        match decode_request(body(&msg)).unwrap() {
+            Command::RawImage { data, .. } => assert_eq!(data, &big[..]),
+            other => panic!("unexpected {other:?}"),
+        }
+        // 172800 bytes of pixels plus a small overhead.
+        assert!(msg.len() < 320 * 180 * 3 + 128, "{}", msg.len());
+    }
+
+    #[test]
+    fn reply_roundtrip() {
+        let mut fbb = FlatBufferBuilder::new();
+        let mut out = Vec::new();
+        encode_reply(&mut fbb, None, 150, &mut out);
+        let r = decode_reply(body(&out)).unwrap();
+        assert_eq!(
+            r,
+            Reply {
+                error: None,
+                video: -1,
+                registered: 150
+            }
+        );
+
+        encode_reply(&mut fbb, Some("Unable to parse message"), -1, &mut out);
+        let r = decode_reply(body(&out)).unwrap();
+        assert_eq!(r.error, Some("Unable to parse message"));
+        assert_eq!(r.registered, -1);
+    }
+
+    #[test]
+    fn garbage_is_rejected() {
+        assert!(decode_request(&[1, 2, 3]).is_err());
+        assert!(decode_request(&[0xFF; 64]).is_err());
+        assert!(decode_reply(&[0xFF; 16]).is_err());
+    }
+
+    #[test]
+    fn framing() {
+        let mut enc = Encoder::new();
+        let mut stream = enc.register("a", 150).to_vec();
+        stream.extend_from_slice(enc.clear(150));
+        let mut cursor = io::Cursor::new(stream);
+        let mut buf = Vec::new();
+        assert!(read_message(&mut cursor, &mut buf, 1000).unwrap());
+        assert!(matches!(
+            decode_request(&buf).unwrap(),
+            Command::Register { .. }
+        ));
+        assert!(read_message(&mut cursor, &mut buf, 1000).unwrap());
+        assert!(matches!(
+            decode_request(&buf).unwrap(),
+            Command::Clear { .. }
+        ));
+        assert!(!read_message(&mut cursor, &mut buf, 1000).unwrap());
+
+        let mut zero = io::Cursor::new(vec![0, 0, 0, 0]);
+        assert!(matches!(
+            read_message(&mut zero, &mut buf, 1000),
+            Err(FrameError::InvalidLength(0))
+        ));
+        let mut long = io::Cursor::new(vec![0, 0, 3, 233, 0]);
+        assert!(matches!(
+            read_message(&mut long, &mut buf, 1000),
+            Err(FrameError::InvalidLength(1001))
+        ));
+        let mut truncated = io::Cursor::new(vec![0, 0, 0, 8, 1, 2]);
+        assert!(matches!(
+            read_message(&mut truncated, &mut buf, 1000),
+            Err(FrameError::Io(_))
+        ));
+        let mut half_header = io::Cursor::new(vec![0, 0]);
+        assert!(matches!(
+            read_message(&mut half_header, &mut buf, 1000),
+            Err(FrameError::Io(_))
+        ));
+    }
+}