Reset the protocol format

Cargo.toml

@@ -43,7 +43,7 @@ tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-postcard = { version = "1", features = ["use-std"] }
+minicbor = { version = "2", features = ["derive", "alloc"] }
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "io-util", "net", "process", "sync", "time", "io-std", "fs"] }
 bytes = "1"
 clap = { version = "4", features = ["derive"] }

crates/gliff-probe/src/main.rs

@@ -509,7 +509,7 @@ fn serve_test(node: &std::path::Path, addr: &str, frames: usize, video: VideoMod
         let (rd, wr) = tokio::io::split(stream);
         let mut reader = Framed::new(rd);
         let mut writer = Framed::new(wr);
-        let caps = ClientCaps { codecs: vec![Codec::H264], max_width: 1280, max_height: 720, chroma: vec![ChromaMode::Dual420, ChromaMode::Single420] };
+        let caps = ClientCaps { codecs: vec![Codec::H264], max_width: 1280, max_height: 720, chroma: vec![ChromaMode::Dual420, ChromaMode::Single420], features: gliff_proto::features() };
         writer.write_msg(&ClientMsg::Hello { version: gliff_proto::PROTOCOL_VERSION, keymap: String::new(), caps }).await?;
         let ack = reader.read_msg::<ServerMsg>().await?;
         let ServerMsg::HelloAck { session, outputs, .. } = ack else { bail!("expected HelloAck, got {ack:?}") };
@@ -783,7 +783,7 @@ fn stream_bench(
         let (rd, wr) = tokio::io::split(stream);
         let mut reader = Framed::new(rd);
         let mut writer = Framed::new(wr);
-        let caps = ClientCaps { codecs: vec![Codec::H264], max_width: 3840, max_height: 2160, chroma: vec![ChromaMode::Dual420, ChromaMode::Single420] };
+        let caps = ClientCaps { codecs: vec![Codec::H264], max_width: 3840, max_height: 2160, chroma: vec![ChromaMode::Dual420, ChromaMode::Single420], features: gliff_proto::features() };
         writer.write_msg(&ClientMsg::Hello { version: gliff_proto::PROTOCOL_VERSION, keymap: String::new(), caps }).await?;
         let ack = reader.read_msg::<ServerMsg>().await?;
         let ServerMsg::HelloAck { session, .. } = ack else { bail!("expected HelloAck, got {ack:?}") };
@@ -810,8 +810,8 @@ fn stream_bench(
         let mut csv_out = match csv { Some(p) => Some(std::io::BufWriter::new(std::fs::File::create(p)?)), None => None };
         if let Some(c) = csv_out.as_mut() { writeln!(c, "t_ms,frame_id,key,bytes,latency_ms,decode_ms")?; }
 
-        // Reads happen on their own task: `read_msg` is not cancel-safe in
-        // a `select!`, and pings must go out while no frames arrive.
+        // Reads happen on their own task, so pings go out while no frames
+        // arrive.
         let (msg_tx, mut msg_rx) = tokio::sync::mpsc::unbounded_channel();
         tokio::task::spawn_local(async move {
             loop {

crates/gliff-proto/Cargo.toml

@@ -9,8 +9,7 @@ rust-version.workspace = true
 workspace = true
 
 [dependencies]
-serde.workspace = true
-postcard.workspace = true
+minicbor.workspace = true
 thiserror.workspace = true
 rayon.workspace = true
 

crates/gliff-proto/src/cbor.rs

@@ -0,0 +1,67 @@
+//! Decoding helpers for values a newer peer may extend.
+
+use minicbor::data::Type;
+use minicbor::decode::{Decoder, Error};
+use minicbor::Decode;
+
+/// Decode an optional enum field, reading a value this build does not know
+/// as `None`. minicbor's own `Option` decoding mishandles an unknown
+/// `index_only` value and misreads the fields after it, so every optional
+/// enum field uses this with `nil = "crate::cbor::none"`.
+pub fn optional<'b, Ctx, T: Decode<'b, Ctx>>(
+    d: &mut Decoder<'b>,
+    ctx: &mut Ctx,
+) -> Result<Option<T>, Error> {
+    if d.datatype()? == Type::Null {
+        d.skip()?;
+        return Ok(None);
+    }
+    let start = d.position();
+    match T::decode(d, ctx) {
+        Ok(v) => Ok(Some(v)),
+        Err(e) if e.is_unknown_variant() => {
+            d.set_position(start);
+            d.skip()?;
+            Ok(None)
+        }
+        Err(e) => Err(e),
+    }
+}
+
+/// The value of an absent [`optional`] field.
+pub fn none<T>() -> Option<Option<T>> {
+    Some(None)
+}
+
+/// Decode a list, dropping entries whose enum variant this build does not
+/// know. Used where a peer describes itself before any features are agreed,
+/// so a newer peer's additions cannot make the whole message fail.
+pub fn known<'b, Ctx, T: Decode<'b, Ctx>>(
+    d: &mut Decoder<'b>,
+    ctx: &mut Ctx,
+) -> Result<Vec<T>, Error> {
+    let len = d.array()?;
+    let mut out = Vec::new();
+    let mut seen = 0u64;
+    loop {
+        match len {
+            Some(n) if seen == n => break,
+            None if d.datatype()? == Type::Break => {
+                d.skip()?;
+                break;
+            }
+            _ => {}
+        }
+        let start = d.position();
+        match T::decode(d, ctx) {
+            Ok(v) => out.push(v),
+            Err(e) if e.is_unknown_variant() => {
+                d.set_position(start);
+                d.skip()?;
+            }
+            Err(e) => return Err(e),
+        }
+        seen += 1;
+    }
+    Ok(out)
+}