refactor(guest-exec): CBOR wire protocol, drop hand-rolled framing

Replace length-prefixed JSON framing with self-delimiting CBOR in both
directions. Request: client CBOR.encodes a map and sends raw bytes;
agent reads one value via ciborium::from_reader (no EOF needed).
Response: agent ciborium::into_writer's then closes; client reads to EOF
and CBOR.decodes the accumulated buffer. stdout/stderr are CBOR byte
strings (serde_bytes on Rust, %CBOR.Tag{tag: :bytes} unwrap on Elixir).
Removes serde_json from the guest-agent crate. Cross-language contract
is pinned by a Rust fixture (decodes the exact Elixir-produced anchor
bytes) and a matching Elixir assertion on the same hex anchor.

Author: markovejnovic

lib/hyper/node/fire_vmm/exec.ex

@@ -13,12 +13,14 @@ defmodule Hyper.Node.FireVMM.Exec do
       → "CONNECT 1024\\n"
       ← "OK <port>\\n"     (anything not starting with "OK " is an error)
 
-  Request frame: `<<byte_size(json)::32-unsigned-big>>` followed by UTF-8
-  JSON `{argv, env?, cwd?, timeout_ms?}`. Keys with `nil` values are omitted.
+  **Request**: a CBOR-encoded map `{argv, env?, cwd?, timeout_ms?}`. The
+  client sends the bytes directly with no framing — CBOR is self-delimiting
+  and the agent reads exactly one value via `ciborium::from_reader`.
 
-  Response frame: `<<exit_code::32-signed-big>>`, then two length-prefixed
-  byte fields — `<<stdout_len::32>>` + stdout bytes, `<<stderr_len::32>>`
-  + stderr bytes.
+  **Response**: a CBOR-encoded map `{exit_code, stdout, stderr}`. `stdout`
+  and `stderr` are CBOR byte strings that unwrap from `%CBOR.Tag{tag: :bytes}`
+  to raw binaries. The agent writes the value then closes the connection; the
+  client reads to EOF and decodes the accumulated bytes.
 
   ## Readiness retry
 
@@ -120,62 +122,52 @@ defmodule Hyper.Node.FireVMM.Exec do
 
   @spec send_request(:gen_tcp.socket(), [String.t()], keyword()) :: :ok | {:error, term()}
   defp send_request(sock, argv, opts) do
-    json = build_request_json(argv, opts)
-    :gen_tcp.send(sock, [<<byte_size(json)::32>>, json])
+    body =
+      %{"argv" => argv}
+      |> maybe_put("env", Keyword.get(opts, :env))
+      |> maybe_put("cwd", Keyword.get(opts, :cwd))
+      |> maybe_put("timeout_ms", Keyword.get(opts, :timeout_ms))
+      |> CBOR.encode()
+
+    :gen_tcp.send(sock, body)
   end
 
-  @spec build_request_json([String.t()], keyword()) :: binary()
-  defp build_request_json(argv, opts) do
-    %{"argv" => argv}
-    |> maybe_put("env", Keyword.get(opts, :env))
-    |> maybe_put("cwd", Keyword.get(opts, :cwd))
-    |> maybe_put("timeout_ms", Keyword.get(opts, :timeout_ms))
-    |> Jason.encode!()
-  end
-
-  @spec maybe_put(map(), String.t(), term()) :: map()
-  defp maybe_put(map, _key, nil), do: map
-  defp maybe_put(map, key, value), do: Map.put(map, key, value)
-
   @spec recv_response(:gen_tcp.socket(), non_neg_integer()) ::
           {:ok, exec_result()} | {:error, term()}
   defp recv_response(sock, timeout_ms) do
-    with {:ok, <<exit_code::32-signed>>} <- recv_exactly(sock, 4, timeout_ms),
-         {:ok, <<stdout_len::32>>} <- recv_exactly(sock, 4, timeout_ms),
-         {:ok, stdout} <- recv_exactly(sock, stdout_len, timeout_ms),
-         {:ok, <<stderr_len::32>>} <- recv_exactly(sock, 4, timeout_ms),
-         {:ok, stderr} <- recv_exactly(sock, stderr_len, timeout_ms) do
-      {:ok, %{exit_code: exit_code, stdout: stdout, stderr: stderr}}
+    with {:ok, buffer} <- recv_until_closed(sock, timeout_ms, <<>>) do
+      case CBOR.decode(buffer) do
+        {:ok, %{"exit_code" => code, "stdout" => out, "stderr" => err}, _rest} ->
+          {:ok, %{exit_code: code, stdout: unwrap_bytes(out), stderr: unwrap_bytes(err)}}
+
+        {:ok, _other, _rest} ->
+          {:error, {:cbor_decode, :unexpected_shape}}
+
+        {:error, reason} ->
+          {:error, {:cbor_decode, reason}}
+      end
     end
   end
 
-  # Read exactly `n` bytes, looping until all arrive. `:gen_tcp.recv/3` with a
-  # positive length waits until that many bytes are buffered by the OTP TCP
-  # driver, so the recursive branch is a defensive backstop rather than the
-  # common path — but it makes the accumulation contract explicit and protects
-  # against any future refactor to a lower-level read.
-  @spec recv_exactly(:gen_tcp.socket(), non_neg_integer(), non_neg_integer()) ::
+  @spec recv_until_closed(:gen_tcp.socket(), non_neg_integer(), binary()) ::
           {:ok, binary()} | {:error, term()}
-  defp recv_exactly(_sock, 0, _timeout_ms), do: {:ok, <<>>}
-
-  defp recv_exactly(sock, n, timeout_ms) do
-    case :gen_tcp.recv(sock, n, timeout_ms) do
-      {:ok, data} when byte_size(data) == n ->
-        {:ok, data}
-
-      {:ok, partial} ->
-        case recv_exactly(sock, n - byte_size(partial), timeout_ms) do
-          {:ok, rest} -> {:ok, partial <> rest}
-          err -> err
-        end
-
-      {:error, reason} ->
-        {:error, reason}
+  defp recv_until_closed(sock, timeout_ms, acc) do
+    case :gen_tcp.recv(sock, 0, timeout_ms) do
+      {:ok, data} -> recv_until_closed(sock, timeout_ms, acc <> data)
+      {:error, :closed} -> {:ok, acc}
+      {:error, reason} -> {:error, reason}
     end
   end
 
-  # Read bytes one at a time until `\n`; returns the line without the trailing
-  # newline. Byte-by-byte is correct for the short (<20 byte) handshake line.
+  @spec unwrap_bytes(term()) :: binary()
+  defp unwrap_bytes(%CBOR.Tag{tag: :bytes, value: bin}), do: bin
+  defp unwrap_bytes(nil), do: ""
+  defp unwrap_bytes(bin) when is_binary(bin), do: bin
+
+  @spec maybe_put(map(), String.t(), term()) :: map()
+  defp maybe_put(map, _key, nil), do: map
+  defp maybe_put(map, key, value), do: Map.put(map, key, value)
+
   @spec recv_line(:gen_tcp.socket(), non_neg_integer()) :: {:ok, String.t()} | {:error, term()}
   defp recv_line(sock, timeout_ms), do: recv_line(sock, timeout_ms, <<>>)
 

mix.exs

@@ -84,6 +84,7 @@ defmodule Hyper.MixProject do
       {:grpc, "~> 1.0"},
       {:grpc_server, "~> 1.0"},
       {:horde, "~> 0.9"},
+      {:cbor, "~> 1.0"},
       {:jason, "~> 1.4"},
       {:libcluster, "~> 3.3"},
       {:muontrap, "~> 1.5"},

mix.lock

@@ -2,6 +2,7 @@
   "acceptor_pool": {:hex, :acceptor_pool, "1.0.1", "d88c2e8a0be9216cf513fbcd3e5a4beb36bee3ff4168e85d6152c6f899359cdb", [:rebar3], [], "hexpm", "f172f3d74513e8edd445c257d596fc84dbdd56d2c6fa287434269648ae5a421e"},
   "bunt": {:hex, :bunt, "1.0.0", "081c2c665f086849e6d57900292b3a161727ab40431219529f13c4ddcf3e7a44", [:mix], [], "hexpm", "dc5f86aa08a5f6fa6b8096f0735c4e76d54ae5c9fa2c143e5a1fc7c1cd9bb6b5"},
   "castore": {:hex, :castore, "1.0.19", "6903cabdfd9d1af46454126e7c8385186659dd33ecfb74a885cae52221ad6109", [:mix], [], "hexpm", "3669e6cab13f54c2df26b3e6833745d647f35b6e30d8ddd5975df0d5c842ca98"},
+  "cbor": {:hex, :cbor, "1.0.2", "9b0af85af291a556e10a0ffd48ba9a21a75e711828fafd3af193d56d95f0907f", [:mix], [], "hexpm", "edbc9b4a16eb93a582437b9b249c340a75af03958e338fb43d8c1be9fc65b864"},
   "chatterbox": {:hex, :ts_chatterbox, "0.15.1", "5cac4d15dd7ad61fc3c4415ce4826fc563d4643dee897a558ec4ea0b1c835c9c", [:rebar3], [{:hpack, "~> 0.3.0", [hex: :hpack_erl, repo: "hexpm", optional: false]}], "hexpm", "4f75b91451338bc0da5f52f3480fa6ef6e3a2aeecfc33686d6b3d0a0948f31aa"},
   "cowboy": {:hex, :cowboy, "2.16.1", "fa04080b602ff25c40a7700f2dc0152dbc1ba26b42093ae0fa9bb7a337d5a242", [:make, :rebar3], [{:cowlib, ">= 2.16.0 and < 3.0.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, ">= 1.8.0 and < 3.0.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "b8ea4dd317a043e3177ec840cfa3bcb47cfb41035d3abb24d954dc7d51def399"},
   "cowlib": {:hex, :cowlib, "2.17.1", "3e6053016d1ab245730f0af688755476dcedb1c25ed8fb5751f59a2bfdc0c9af", [:make, :rebar3], [], "hexpm", "ff08bd17e6dd931445b18af77315b9b5fe052407110964ad2588c686b57b5e3f"},

native/guest-agent/Cargo.lock

@@ -13,7 +13,8 @@ path = "src/main.rs"
 
 [dependencies]
 serde = { version = "1", features = ["derive"] }
-serde_json = "1"
+ciborium = "0.2"
+serde_bytes = "0.11"
 nix = { version = "0.29", features = ["mount", "process", "socket", "signal", "fs"] }
 
 [dev-dependencies]

native/guest-agent/Cargo.toml

@@ -1,3 +1,18 @@
+//! Wire protocol between the Elixir guest-exec client and this agent.
+//!
+//! **Request** (Elixir client → agent): the client `CBOR.encode`s a map and
+//! sends the bytes directly — no length prefix, no delimiter. The agent reads
+//! exactly one self-delimiting CBOR value via `ciborium::from_reader`, which
+//! stops at the end of the definite-length map without needing EOF.
+//!
+//! **Response** (agent → client): the agent `ciborium::into_writer`s a
+//! [`Response`] map and then **closes the connection**. The client reads the
+//! socket to EOF, then calls `CBOR.decode` on the accumulated bytes.
+//!
+//! `stdout` and `stderr` are CBOR **byte strings** (major type 2), encoded via
+//! `serde_bytes` on the Rust side and unwrapped from `%CBOR.Tag{tag: :bytes}`
+//! on the Elixir side, so raw binary data round-trips without base64.
+
 use std::collections::BTreeMap;
 use std::io::{self, Read, Write};
 
@@ -14,56 +29,29 @@ pub struct Request {
     pub timeout_ms: Option<u64>,
 }
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct Response {
     pub exit_code: i32,
+    #[serde(with = "serde_bytes")]
     pub stdout: Vec<u8>,
+    #[serde(with = "serde_bytes")]
     pub stderr: Vec<u8>,
 }
 
-fn read_u32(r: &mut impl Read) -> io::Result<u32> {
-    let mut b = [0u8; 4];
-    r.read_exact(&mut b)?;
-    Ok(u32::from_be_bytes(b))
+pub fn read_request(r: impl Read) -> io::Result<Request> {
+    ciborium::from_reader(r).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))
 }
 
-fn read_frame(r: &mut impl Read) -> io::Result<Vec<u8>> {
-    let len = read_u32(r)? as usize;
-    let mut buf = vec![0u8; len];
-    r.read_exact(&mut buf)?;
-    Ok(buf)
-}
-
-fn write_frame(w: &mut impl Write, bytes: &[u8]) -> io::Result<()> {
-    w.write_all(&(bytes.len() as u32).to_be_bytes())?;
-    w.write_all(bytes)
-}
-
-pub fn read_request(r: &mut impl Read) -> io::Result<Request> {
-    let frame = read_frame(r)?;
-    serde_json::from_slice(&frame).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))
-}
-
-pub fn write_request(w: &mut impl Write, req: &Request) -> io::Result<()> {
-    let json = serde_json::to_vec(req)?;
-    write_frame(w, &json)
+pub fn write_request(mut w: impl Write, req: &Request) -> io::Result<()> {
+    ciborium::into_writer(req, &mut w).map_err(|e| io::Error::other(e.to_string()))?;
+    w.flush()
 }
 
-pub fn write_response(w: &mut impl Write, resp: &Response) -> io::Result<()> {
-    w.write_all(&resp.exit_code.to_be_bytes())?;
-    write_frame(w, &resp.stdout)?;
-    write_frame(w, &resp.stderr)?;
+pub fn write_response(mut w: impl Write, resp: &Response) -> io::Result<()> {
+    ciborium::into_writer(resp, &mut w).map_err(|e| io::Error::other(e.to_string()))?;
     w.flush()
 }
 
-pub fn read_response(r: &mut impl Read) -> io::Result<Response> {
-    let mut code = [0u8; 4];
-    r.read_exact(&mut code)?;
-    let stdout = read_frame(r)?;
-    let stderr = read_frame(r)?;
-    Ok(Response {
-        exit_code: i32::from_be_bytes(code),
-        stdout,
-        stderr,
-    })
+pub fn read_response(r: impl Read) -> io::Result<Response> {
+    ciborium::from_reader(r).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))
 }