From f02c6554c09e0ac4c7ba251a65a942f96fb762b1 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 20 Oct 2026 09:08:19 -0700 Subject: [PATCH 1/6] tests: serve the control session in the runtime adapter's guest agent Give the runtime adapter its own application framing, length-prefixed messages with a kind byte, and a guest agent that serves the host program with it over the guest's end of the control session, through nvx::guest::nvxs alone. The guest agent acknowledges each reset once it dropped the earlier epoch's partial message, credits each payload back once it holds it, and answers Ping, Echo and Stop. `runtime-adapter-guest serve` runs it on its standard input and output. tests/guest_session.rs checks it against a test broker that plays the broker's end of the guest's session over a Unix socket pair, for the library and for the static binary. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 086f45a6-c39c-4704-85ef-1fa7f8b8b672 --- doc/ci.md | 5 +- doc/project-structure.md | 9 +- tests/runtime-adapter/README.md | 83 ++- .../src/bin/runtime-adapter-guest.rs | 71 +- tests/runtime-adapter/src/frame.rs | 406 +++++++++++ tests/runtime-adapter/src/guest.rs | 123 ++++ tests/runtime-adapter/src/lib.rs | 19 +- tests/runtime-adapter/tests/guest_session.rs | 656 ++++++++++++++++++ 8 files changed, 1342 insertions(+), 30 deletions(-) create mode 100644 tests/runtime-adapter/src/frame.rs create mode 100644 tests/runtime-adapter/src/guest.rs create mode 100644 tests/runtime-adapter/tests/guest_session.rs diff --git a/doc/ci.md b/doc/ci.md index d887d3447..7da2a3cc8 100644 --- a/doc/ci.md +++ b/doc/ci.md @@ -660,7 +660,10 @@ build, the tests, and a check with the minimum Rust version of `nvx`, after rustfmt. The tests check consumer-contract scenario C0: each configuration enables only its side of `nvx`, the guest-only build pulls in neither Tokio nor a dependency that only the host side declares, and the guest binary is a -static executable. +static executable. The guest-only configuration also checks the guest's part +of scenario C1: the adapter's guest half and its binary serve the NVXS session +with the adapter's own framing against a test broker, which sends them +messages that span records, resets, disconnects, and malformed records. The development release job depends on `aci-edge-sandboxes` and `nvx-rs` and requires their combined Linux/Windows results to be successful; failed, diff --git a/doc/project-structure.md b/doc/project-structure.md index ac5a64d47..9b2d944bb 100644 --- a/doc/project-structure.md +++ b/doc/project-structure.md @@ -161,8 +161,8 @@ nvx/ | `-- test_*.py Python tooling unit tests |-- tests/ Tests that build on the repository's crates from outside them | `-- runtime-adapter/ Rust crate `runtime-adapter`, an independent adapter of nvx -| |-- src/ Its library, and its host and guest skeleton binaries -| `-- tests/ Consumer-contract scenario C0, the independent build +| |-- src/ Its framing, its guest half, and its host and guest binaries +| `-- tests/ Consumer-contract scenario C0, and the guest's part of C1 |-- SECURITY.md Microsoft security vulnerability reporting policy |-- .dockerignore Docker build-context exclusions |-- .gitattributes Git path attributes @@ -335,8 +335,9 @@ Tests that build on the repository's crates from outside them. crate with its own `Cargo.lock` that depends on `nvx` with default features off and uses its public API only. Its protocol-only, host-only, and guest-only configurations build in CI, and its tests check that each one builds as an -independent adapter's does. See its -[README](../tests/runtime-adapter/README.md). +independent adapter's does. Its guest half serves the guest's role of the NVXS +session with the adapter's own framing, which its tests check against a test +broker. See its [README](../tests/runtime-adapter/README.md). ## Root files diff --git a/tests/runtime-adapter/README.md b/tests/runtime-adapter/README.md index 578658886..9e7b7af46 100644 --- a/tests/runtime-adapter/README.md +++ b/tests/runtime-adapter/README.md @@ -12,22 +12,67 @@ does: [microsoft/nvx#535](https://github.com/microsoft/nvx/issues/535) sets the contract that it proves: an adapter brings its own application framing over -the NVXS session, its own init, and its own status collector. Today its -binaries are skeletons that report the contract that they were built against. -Later steps fill its guest half -([microsoft/nvx#619](https://github.com/microsoft/nvx/issues/619)) and its host -half ([microsoft/nvx#620](https://github.com/microsoft/nvx/issues/620)), and add -the scenarios that run them on a hypervisor. +the NVXS session, its own init, and its own status collector. Its guest half +serves the guest's role of the NVXS session with the adapter's own framing. +Its host binary is still a skeleton that reports the contract that it was +built against. Later steps give the guest half the rest of the guest API +([microsoft/nvx#619](https://github.com/microsoft/nvx/issues/619)), fill the +host half ([microsoft/nvx#620](https://github.com/microsoft/nvx/issues/620)), +and add the scenarios that run both halves on a hypervisor. ## Configurations Each configuration enables one side of `nvx`: -| Configuration | Cargo options | `nvx` features | Binary | -| --- | --- | --- | --- | -| Protocol-only | none | none | none | -| Host-only | `--features host` | `host` | `runtime-adapter-host` | -| Guest-only | `--features guest --target x86_64-unknown-linux-musl` | `guest` | `runtime-adapter-guest`, a static executable | +| Configuration | Cargo options | `nvx` features | Modules | Binary | +| --- | --- | --- | --- | --- | +| Protocol-only | none | none | `frame` | none | +| Host-only | `--features host` | `host` | `frame` | `runtime-adapter-host` | +| Guest-only | `--features guest --target x86_64-unknown-linux-musl` | `guest` | `frame`, `guest` | `runtime-adapter-guest`, a static executable | + +## Framing + +The `frame` module is the adapter's application framing, which its halves +exchange in the payloads of the session's DATA records in place of NVX's +`NVXC` frames. A message is a 5-byte header, then a body. The header holds the +body's length, a little-endian `u32`, then the message's kind: + +| Kind | Message | Body | +| --- | --- | --- | +| 1 | `Ping`: asks the guest to answer with `Pong` | none | +| 2 | `Pong`: the guest's answer to `Ping` | none | +| 3 | `Echo`: bytes that the guest sends back in an `Echo` of its own | at most 256 KiB | +| 4 | `Stop`: asks the guest to stop serving | none | +| 5 | `Stopped`: the guest's confirmation of `Stop`, its last message | none | + +The session keeps no message boundaries, so a message may span several DATA +records, and one record may end a message and start the next. `Decoder` +reassembles the messages. It checks each header before it takes any byte of +the body, and holds at most one partial message. A malformed message ends the +stream, because the framing cannot find the next message after it. + +## Guest half + +`guest::serve` serves the guest's role of the NVXS session on a stream that +connects to the broker's guest leg, through `nvx::guest::nvxs`. It attaches, +then: + +- acknowledges each RESET with the smallest receive window that NVXS allows, + once it dropped the earlier epoch's partial message, because a new epoch + starts with a new message; +- credits each DATA payload back once it holds the payload, so that a message + longer than the window still arrives whole; +- answers each `Ping` with `Pong`, sends each `Echo` back, and answers `Stop` + with `Stopped`, then returns. + +A failure of the session, a malformed message, and a message that only the +guest sends end serving with an error. + +`runtime-adapter-guest serve` runs the guest half on its standard input and +output, which connect to the broker's guest leg, such as a control console in +raw mode. It exits with 0 once the host stopped it or the broker closed the +stream, and with 1, after a diagnostic on standard error, once serving failed. +Without arguments, it reports the contract that it was built against. ## Tests @@ -46,6 +91,22 @@ configuration: guest binary is a static executable: it has no program interpreter, needs no shared library, and keeps its stack non-executable. +The unit tests of `frame` run in every configuration. They cover each kind's +encoding, messages in pieces of any size and split at every byte, and the +headers that the framing refuses before their body arrives. + +`tests/guest_session.rs` checks the guest's part of consumer-contract scenario +C1, NVXS with independent framing, in the guest-only configuration. A test +broker plays the broker's guest leg over a Unix socket pair with the public +record API of `nvx`, for `guest::serve` and for `runtime-adapter-guest +serve`. It covers messages that span records both ways and fill the receive +window, a record whose bytes arrive one at a time, resets in the middle of a +message, with the next epoch of the instance and with a new instance, streams +that end at and inside a record, and the records and messages that the guest +half refuses: sequence, epoch, and instance errors, stale and identity-less +resets, record lengths, bootstrap identity, ERROR payloads, and malformed +headers. + Run the checks of each configuration in this directory. The guest-only configuration needs Linux and the musl target, which `rustup target add x86_64-unknown-linux-musl` installs: diff --git a/tests/runtime-adapter/src/bin/runtime-adapter-guest.rs b/tests/runtime-adapter/src/bin/runtime-adapter-guest.rs index e157e5bd5..6d486ba27 100644 --- a/tests/runtime-adapter/src/bin/runtime-adapter-guest.rs +++ b/tests/runtime-adapter/src/bin/runtime-adapter-guest.rs @@ -1,17 +1,72 @@ -//! The guest half of the runtime adapter, a static `x86_64-unknown-linux-musl` executable. This -//! skeleton reports the contract that it was built against; later steps make it serve the guest -//! role of the NVXS session with the adapter's own framing. +//! The guest half of the runtime adapter, a static `x86_64-unknown-linux-musl` executable. +//! +//! - `runtime-adapter-guest` reports the contract that it was built against. +//! - `runtime-adapter-guest serve` serves the guest's role of the NVXS session with the adapter's +//! own framing on its standard input and output, which connect to the broker's guest leg, such +//! as a control console in raw mode. It exits with 0 once the host stopped it or the broker +//! closed the stream, and with 1, after a diagnostic on standard error, once serving failed. +//! +//! Anything else is a usage error, with status 2. use std::process::ExitCode; #[cfg(target_os = "linux")] fn main() -> ExitCode { - if std::env::args_os().len() > 1 { - eprintln!("usage: runtime-adapter-guest"); - return ExitCode::from(2); + let arguments = std::env::args_os().skip(1).collect::>(); + match arguments.as_slice() { + [] => { + println!("runtime-adapter-guest: {}", runtime_adapter::contract()); + ExitCode::SUCCESS + } + [command] if command == "serve" => serve(), + _ => { + eprintln!("usage: runtime-adapter-guest [serve]"); + ExitCode::from(2) + } + } +} + +/// Serves the session on standard input and output. +#[cfg(target_os = "linux")] +fn serve() -> ExitCode { + let stdio = Stdio { + input: std::io::stdin().lock(), + output: std::io::stdout().lock(), + }; + match runtime_adapter::guest::serve(stdio) { + Ok(_) => ExitCode::SUCCESS, + Err(error) => { + eprintln!("runtime-adapter-guest: {error}"); + ExitCode::FAILURE + } + } +} + +/// Standard input and output as one stream, which carries nothing but the session's records. +#[cfg(target_os = "linux")] +struct Stdio { + /// Standard input, from the broker. + input: std::io::StdinLock<'static>, + /// Standard output, to the broker. + output: std::io::StdoutLock<'static>, +} + +#[cfg(target_os = "linux")] +impl std::io::Read for Stdio { + fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { + self.input.read(buffer) + } +} + +#[cfg(target_os = "linux")] +impl std::io::Write for Stdio { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.output.write(bytes) + } + + fn flush(&mut self) -> std::io::Result<()> { + self.output.flush() } - println!("runtime-adapter-guest: {}", runtime_adapter::contract()); - ExitCode::SUCCESS } // The guest half runs only in Linux guests. Elsewhere this stub fails, so that every target still diff --git a/tests/runtime-adapter/src/frame.rs b/tests/runtime-adapter/src/frame.rs new file mode 100644 index 000000000..677aaaf95 --- /dev/null +++ b/tests/runtime-adapter/src/frame.rs @@ -0,0 +1,406 @@ +//! The adapter's own application framing, which its halves exchange over the NVXS session where +//! NVX's own adapter exchanges `NVXC` frames. +//! +//! A message is a header of [`HEADER_LEN`] bytes, then a body. The header holds the body's length, +//! a little-endian `u32`, then the message's kind. NVXS carries bytes and keeps no message +//! boundaries, so a message may span several DATA records, and one record may end a message and +//! start the next. [`Decoder`] reassembles the messages: it checks each header before it takes any +//! byte of the body, and holds at most one partial message, whose body has at most [`MAX_BODY`] +//! bytes. +//! +//! The framing cannot find the next message after a malformed one, so a [`FrameError`] ends the +//! stream. + +use std::fmt; + +/// Bytes of a message's header: the body's length, then the message's kind. +pub const HEADER_LEN: usize = 5; + +/// Most bytes of a message's body, which is more than four NVXS DATA records carry. +pub const MAX_BODY: usize = 256 * 1024; + +/// A message of the adapter's framing. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum Message { + /// Asks the guest to answer with [`Message::Pong`]. + Ping, + /// The guest's answer to [`Message::Ping`]. + Pong, + /// Bytes that the guest sends back in an `Echo` of its own. + Echo(Vec), + /// Asks the guest to stop serving, which it confirms with [`Message::Stopped`]. + Stop, + /// The guest's confirmation of [`Message::Stop`], its last message. + Stopped, +} + +impl Message { + /// Returns the message's kind, the last byte of its header. + pub fn kind(&self) -> u8 { + self.parts().0.code() + } + + /// Returns the message's encoding: its header, then its body. An `Echo` body has at most + /// [`MAX_BODY`] bytes. + pub fn encode(&self) -> Result, FrameError> { + let (kind, body) = self.parts(); + if body.len() > kind.max_body() { + return Err(FrameError::Length { + kind: kind.code(), + length: body.len(), + }); + } + // The body has at most `MAX_BODY` bytes, so its length fits the header's field. + let length = body.len() as u32; + let mut encoded = Vec::with_capacity(HEADER_LEN + body.len()); + encoded.extend_from_slice(&length.to_le_bytes()); + encoded.push(kind.code()); + encoded.extend_from_slice(body); + Ok(encoded) + } + + /// Returns the message's kind and body. + fn parts(&self) -> (Kind, &[u8]) { + match self { + Self::Ping => (Kind::Ping, &[]), + Self::Pong => (Kind::Pong, &[]), + Self::Echo(body) => (Kind::Echo, body), + Self::Stop => (Kind::Stop, &[]), + Self::Stopped => (Kind::Stopped, &[]), + } + } +} + +/// Kind of a [`Message`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Kind { + /// [`Message::Ping`]. + Ping, + /// [`Message::Pong`]. + Pong, + /// [`Message::Echo`]. + Echo, + /// [`Message::Stop`]. + Stop, + /// [`Message::Stopped`]. + Stopped, +} + +impl Kind { + /// Returns the kind with `code`, if the framing defines one. + fn from_code(code: u8) -> Option { + match code { + 1 => Some(Self::Ping), + 2 => Some(Self::Pong), + 3 => Some(Self::Echo), + 4 => Some(Self::Stop), + 5 => Some(Self::Stopped), + _ => None, + } + } + + /// Returns the kind's code, the last byte of a header. + fn code(self) -> u8 { + match self { + Self::Ping => 1, + Self::Pong => 2, + Self::Echo => 3, + Self::Stop => 4, + Self::Stopped => 5, + } + } + + /// Returns the most bytes of a body of this kind. + fn max_body(self) -> usize { + match self { + Self::Echo => MAX_BODY, + Self::Ping | Self::Pong | Self::Stop | Self::Stopped => 0, + } + } + + /// Returns the message of this kind with `body`, which has at most + /// [`max_body`](Self::max_body) bytes. + fn message(self, body: Vec) -> Message { + match self { + Self::Ping => Message::Ping, + Self::Pong => Message::Pong, + Self::Echo => Message::Echo(body), + Self::Stop => Message::Stop, + Self::Stopped => Message::Stopped, + } + } +} + +/// A malformed message. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum FrameError { + /// A kind that the framing does not define. + Kind(u8), + /// A body longer than [`MAX_BODY`] bytes, or a body for a kind that has none. + Length { + /// Kind of the message. + kind: u8, + /// Bytes of the body. + length: usize, + }, +} + +impl fmt::Display for FrameError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Kind(kind) => write!(formatter, "unknown message kind {kind}"), + Self::Length { kind, length } => write!( + formatter, + "message of kind {kind} with a body of {length} bytes" + ), + } + } +} + +impl std::error::Error for FrameError {} + +/// Reassembles messages from bytes that arrive in pieces of any size, such as the payloads of an +/// NVXS session's DATA records. +#[derive(Clone, Debug, Default)] +pub struct Decoder { + /// Header of the message in progress. + header: [u8; HEADER_LEN], + /// Bytes of the header that arrived. + header_len: usize, + /// Kind of the message in progress, once its header arrived and checked out. + kind: Option, + /// Bytes of the body that the header announced. + body_len: usize, + /// Bytes of the body that arrived. + body: Vec, +} + +impl Decoder { + /// Creates a decoder that expects the start of a message. + pub fn new() -> Self { + Self::default() + } + + /// Consumes bytes until a message completes or `input` runs out. + /// + /// Returns the number of bytes consumed and the message that the last of them completed, if + /// any. Bytes after that message are not consumed, so the caller passes them again. A + /// malformed header fails before any byte of its body is taken, and leaves the decoder + /// empty. + pub fn push(&mut self, input: &[u8]) -> Result<(usize, Option), FrameError> { + let mut consumed = 0; + let kind = match self.kind { + Some(kind) => kind, + None => { + consumed = (HEADER_LEN - self.header_len).min(input.len()); + self.header[self.header_len..][..consumed].copy_from_slice(&input[..consumed]); + self.header_len += consumed; + if self.header_len < HEADER_LEN { + return Ok((consumed, None)); + } + let [b0, b1, b2, b3, code] = self.header; + // A length that `usize` cannot hold exceeds every limit. + let length = + usize::try_from(u32::from_le_bytes([b0, b1, b2, b3])).unwrap_or(usize::MAX); + let kind = Kind::from_code(code).ok_or(FrameError::Kind(code)); + let checked = kind.and_then(|kind| { + if length > kind.max_body() { + Err(FrameError::Length { kind: code, length }) + } else { + Ok(kind) + } + }); + let kind = checked.inspect_err(|_| self.clear())?; + self.kind = Some(kind); + self.body_len = length; + self.body.reserve_exact(length); + kind + } + }; + let taken = (self.body_len - self.body.len()).min(input.len() - consumed); + self.body + .extend_from_slice(&input[consumed..consumed + taken]); + consumed += taken; + if self.body.len() < self.body_len { + return Ok((consumed, None)); + } + let body = std::mem::take(&mut self.body); + self.clear(); + Ok((consumed, Some(kind.message(body)))) + } + + /// Drops the message in progress, if any, such as when the session resets and its new epoch + /// starts with a new message. + pub fn clear(&mut self) { + *self = Self::default(); + } + + /// Returns whether part of a message arrived. + pub fn is_partial(&self) -> bool { + self.header_len != 0 + } +} + +#[cfg(test)] +mod tests { + use super::{Decoder, FrameError, HEADER_LEN, MAX_BODY, Message}; + + /// Returns one message of each kind, with bodies of no, few, and the most bytes. + fn messages() -> Vec { + vec![ + Message::Ping, + Message::Echo(Vec::new()), + Message::Echo(b"hello".to_vec()), + Message::Pong, + Message::Echo((0..=u8::MAX).cycle().take(MAX_BODY).collect()), + Message::Stop, + Message::Stopped, + ] + } + + /// Returns the encodings of `messages`, one after the other. + fn stream(messages: &[Message]) -> Vec { + messages + .iter() + .flat_map(|message| message.encode().unwrap()) + .collect() + } + + /// Decodes `pieces` in order, and returns the messages that they complete. + fn decode<'a>( + decoder: &mut Decoder, + pieces: impl IntoIterator, + ) -> Vec { + let mut decoded = Vec::new(); + for piece in pieces { + let mut rest = piece; + while !rest.is_empty() { + let (consumed, message) = decoder.push(rest).unwrap(); + assert_ne!(consumed, 0, "a push consumes bytes"); + rest = &rest[consumed..]; + decoded.extend(message); + } + } + decoded + } + + #[test] + fn the_header_holds_the_body_length_then_the_kind() { + assert_eq!(Message::Ping.encode().unwrap(), [0, 0, 0, 0, 1]); + assert_eq!(Message::Pong.encode().unwrap(), [0, 0, 0, 0, 2]); + assert_eq!( + Message::Echo(b"abc".to_vec()).encode().unwrap(), + [3, 0, 0, 0, 3, b'a', b'b', b'c'] + ); + assert_eq!(Message::Stop.encode().unwrap(), [0, 0, 0, 0, 4]); + assert_eq!(Message::Stopped.encode().unwrap(), [0, 0, 0, 0, 5]); + for message in messages() { + assert_eq!(message.encode().unwrap()[HEADER_LEN - 1], message.kind()); + } + } + + #[test] + fn messages_arrive_whole_in_pieces_of_any_size() { + let messages = messages(); + let stream = stream(&messages); + for size in [1, 2, 3, HEADER_LEN, 7, 4096, 64 * 1024, stream.len()] { + let mut decoder = Decoder::new(); + assert_eq!( + decode(&mut decoder, stream.chunks(size)), + messages, + "{size}" + ); + assert!(!decoder.is_partial()); + } + } + + #[test] + fn messages_arrive_whole_across_a_split_at_every_byte() { + let messages = [ + Message::Echo(b"split".to_vec()), + Message::Ping, + Message::Stop, + ]; + let stream = stream(&messages); + for split in 0..=stream.len() { + let mut decoder = Decoder::new(); + let pieces = [&stream[..split], &stream[split..]]; + assert_eq!(decode(&mut decoder, pieces), messages, "{split}"); + assert!(!decoder.is_partial()); + } + } + + #[test] + fn a_push_stops_at_the_end_of_a_message() { + let stream = stream(&[Message::Ping, Message::Echo(b"next".to_vec())]); + let mut decoder = Decoder::new(); + assert_eq!(decoder.push(&stream), Ok((HEADER_LEN, Some(Message::Ping)))); + assert!(!decoder.is_partial()); + assert_eq!( + decoder.push(&stream[HEADER_LEN..]), + Ok((HEADER_LEN + 4, Some(Message::Echo(b"next".to_vec())))) + ); + } + + #[test] + fn malformed_headers_fail_before_their_body_is_taken() { + let too_long = u32::try_from(MAX_BODY + 1).unwrap().to_le_bytes(); + let cases = [ + ([0, 0, 0, 0, 0], FrameError::Kind(0)), + ([0, 0, 0, 0, 6], FrameError::Kind(6)), + ([9, 0, 0, 0, 0xff], FrameError::Kind(0xff)), + ([1, 0, 0, 0, 1], FrameError::Length { kind: 1, length: 1 }), + ([1, 0, 0, 0, 4], FrameError::Length { kind: 4, length: 1 }), + ( + [too_long[0], too_long[1], too_long[2], too_long[3], 3], + FrameError::Length { + kind: 3, + length: MAX_BODY + 1, + }, + ), + ( + [0xff, 0xff, 0xff, 0xff, 3], + FrameError::Length { + kind: 3, + length: u32::MAX as usize, + }, + ), + ]; + for (header, error) in cases { + let mut decoder = Decoder::new(); + assert_eq!(decoder.push(&header[..2]), Ok((2, None))); + assert!(decoder.is_partial()); + let mut rest = header[2..].to_vec(); + rest.extend_from_slice(b"body"); + assert_eq!(decoder.push(&rest), Err(error), "{header:?}"); + assert!(!decoder.is_partial(), "{header:?}"); + assert_eq!(decoder.body.capacity(), 0, "{header:?}"); + } + } + + #[test] + fn clearing_drops_a_partial_message() { + let echo = Message::Echo(vec![1; 100]).encode().unwrap(); + let mut decoder = Decoder::new(); + assert_eq!(decoder.push(&echo[..50]), Ok((50, None))); + assert!(decoder.is_partial()); + decoder.clear(); + assert!(!decoder.is_partial()); + let ping = Message::Ping.encode().unwrap(); + assert_eq!(decoder.push(&ping), Ok((HEADER_LEN, Some(Message::Ping)))); + } + + #[test] + fn encoding_refuses_a_body_beyond_the_limit() { + let longest = Message::Echo(vec![0; MAX_BODY]).encode().unwrap(); + assert_eq!(longest.len(), HEADER_LEN + MAX_BODY); + assert_eq!( + Message::Echo(vec![0; MAX_BODY + 1]).encode(), + Err(FrameError::Length { + kind: 3, + length: MAX_BODY + 1, + }) + ); + } +} diff --git a/tests/runtime-adapter/src/guest.rs b/tests/runtime-adapter/src/guest.rs new file mode 100644 index 000000000..9781e3949 --- /dev/null +++ b/tests/runtime-adapter/src/guest.rs @@ -0,0 +1,123 @@ +//! The adapter's guest half: it serves the guest's role of the NVXS session with the adapter's +//! own [framing](crate::frame), through `nvx::guest::nvxs` alone. + +use std::fmt; +use std::io::{Read, Write}; + +use nvx::guest::nvxs::{GuestConnection, GuestConnectionError}; +use nvx::protocol::OUTER_MIN_RECEIVE_WINDOW; +use nvx::protocol::nvxs::GuestEvent; + +use crate::frame::{Decoder, FrameError, Message}; + +/// The receive window that the guest acknowledges each epoch with: the smallest that NVXS allows, +/// so that the rest of a longer message arrives only as the guest credits what it took. +pub const RECEIVE_WINDOW: u32 = OUTER_MIN_RECEIVE_WINDOW; + +/// How serving ended, when nothing failed. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Served { + /// The host asked the guest to stop, and the guest confirmed. + Stopped, + /// The broker closed the stream at a record boundary. + Closed, +} + +/// A failure that ended serving. +#[derive(Debug)] +#[non_exhaustive] +pub enum ServeError { + /// The NVXS session failed. + Session(GuestConnectionError), + /// The host sent a malformed message. + Frame(FrameError), + /// The host sent a message of a kind that only the guest sends. + Unexpected(u8), +} + +impl fmt::Display for ServeError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Session(error) => error.fmt(formatter), + Self::Frame(error) => error.fmt(formatter), + Self::Unexpected(kind) => { + write!( + formatter, + "the host sent a message of kind {kind}, which only the guest sends" + ) + } + } + } +} + +impl std::error::Error for ServeError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Session(error) => Some(error), + Self::Frame(error) => Some(error), + Self::Unexpected(_) => None, + } + } +} + +impl From for ServeError { + fn from(error: GuestConnectionError) -> Self { + Self::Session(error) + } +} + +impl From for ServeError { + fn from(error: FrameError) -> Self { + Self::Frame(error) + } +} + +/// Serves the guest's role of the NVXS session on `stream`, which connects to the broker's guest +/// leg, until the host stops the guest or the broker closes the stream. +/// +/// The guest attaches, then: +/// +/// - It acknowledges each RESET with [`RECEIVE_WINDOW`], once it dropped the earlier epoch's +/// partial message: a new epoch starts with a new message. +/// - It credits each DATA payload back once it holds the payload, so that a message longer than +/// the window arrives whole. +/// - It answers [`Message::Ping`] with [`Message::Pong`], sends each [`Message::Echo`] back, and +/// answers [`Message::Stop`] with [`Message::Stopped`], then returns without reading further. +/// +/// A failure of the session, a malformed message, and a message that only the guest sends end +/// serving with an error. +pub fn serve(stream: S) -> Result { + let mut connection = GuestConnection::new(stream); + connection.attach()?; + let mut decoder = Decoder::new(); + loop { + let payload = match connection.next_event() { + Ok(GuestEvent::Reset(_)) => { + decoder.clear(); + connection.acknowledge(RECEIVE_WINDOW)?; + continue; + } + Ok(GuestEvent::Data(payload)) => payload, + Err(GuestConnectionError::Closed) => return Ok(Served::Closed), + Err(error) => return Err(error.into()), + }; + // A DATA payload has at most `OUTER_MAX_PAYLOAD` bytes, so its length fits a `u32`. + connection.credit(payload.len() as u32)?; + let mut rest = payload.as_slice(); + while !rest.is_empty() { + let (consumed, message) = decoder.push(rest)?; + rest = &rest[consumed..]; + let reply = match message { + None => continue, + Some(Message::Ping) => Message::Pong, + Some(Message::Echo(body)) => Message::Echo(body), + Some(Message::Stop) => { + connection.send(&Message::Stopped.encode()?)?; + return Ok(Served::Stopped); + } + Some(message) => return Err(ServeError::Unexpected(message.kind())), + }; + connection.send(&reply.encode()?)?; + } + } +} diff --git a/tests/runtime-adapter/src/lib.rs b/tests/runtime-adapter/src/lib.rs index 9bfffe743..ca4a8e25b 100644 --- a/tests/runtime-adapter/src/lib.rs +++ b/tests/runtime-adapter/src/lib.rs @@ -5,15 +5,22 @@ //! Each configuration enables one side of `nvx`: //! //! - protocol-only, without features: the control protocol, the contract versions, and the other -//! items that `nvx` has without features; +//! items that `nvx` has without features, with the adapter's own [`frame`]s; //! - host-only, with the `host` feature: the host API too, and the `runtime-adapter-host` //! binary; -//! - guest-only, with the `guest` feature: the guest API too, and the `runtime-adapter-guest` -//! binary, a static `x86_64-unknown-linux-musl` executable. +//! - guest-only, with the `guest` feature: the guest API too, the adapter's guest half in its +//! `guest` module, and the `runtime-adapter-guest` binary, a static +//! `x86_64-unknown-linux-musl` executable. //! -//! The binaries are skeletons that report the contract they were built against. Later steps give -//! the adapter its own framing over the NVXS session, fill its host and guest halves, and add the -//! scenarios that run them on a hypervisor. +//! The adapter brings its own framing to the NVXS session, in place of NVX's `NVXC` frames. Its +//! guest half serves the guest's role of the session with it, and `runtime-adapter-guest serve` +//! runs the guest half on its standard input and output. The host binary is a skeleton that +//! reports the contract it was built against. Later steps fill the host half, give the guest half +//! the rest of the guest API, and add the scenarios that run both halves on a hypervisor. + +pub mod frame; +#[cfg(all(feature = "guest", target_os = "linux"))] +pub mod guest; /// Describes the runtime contract that this build speaks: the control contract revision, and the /// versions of the microVM ABI, the NVXS records, and the NVXC frames. diff --git a/tests/runtime-adapter/tests/guest_session.rs b/tests/runtime-adapter/tests/guest_session.rs new file mode 100644 index 000000000..005482c8e --- /dev/null +++ b/tests/runtime-adapter/tests/guest_session.rs @@ -0,0 +1,656 @@ +//! Consumer-contract scenario C1, the guest's part: the adapter's guest half serves the guest's +//! role of the NVXS session with the adapter's own framing, through public APIs only. +//! +//! A test broker plays the broker's guest leg over a Unix socket pair, with the public record API +//! of `nvx`: for the guest half in a thread, and for the guest binary, whose standard input and +//! output are the socket. The tests cover messages that span records, partial records, resets in +//! the middle of a message, disconnects, record lengths, bootstrap identity, ERROR payloads, +//! sequence and identity errors, and the bounded buffer of the adapter's decoder. +#![cfg(all(feature = "guest", target_os = "linux"))] + +use std::io::{self, Read, Write}; +use std::net::Shutdown; +use std::os::fd::OwnedFd; +use std::os::unix::net::UnixStream; +use std::process::{Child, Command, Stdio}; +use std::sync::mpsc::{self, Receiver, Sender}; +use std::thread::{self, JoinHandle}; +use std::time::Duration; + +use nvx::guest::nvxs::GuestConnectionError; +use nvx::protocol::OUTER_MAX_PAYLOAD; +use nvx::protocol::nvxs::{ + ErrorCode, Identity, Record, RecordError, RecordParser, RecordType, SessionError, encode_record, +}; +use runtime_adapter::frame::{Decoder, FrameError, HEADER_LEN, MAX_BODY, Message}; +use runtime_adapter::guest::{self, RECEIVE_WINDOW, ServeError, Served}; + +/// Identity of the broker's first epoch. +const FIRST: Identity = Identity { + instance_id: [0x5c; 16], + epoch: 1, +}; + +/// Longest wait for a record from the guest. +const TIMEOUT: Duration = Duration::from_secs(30); + +/// Offset of the protocol version in a record's header. +const VERSION_OFFSET: usize = 4; +/// Offset of the record type in a record's header. +const TYPE_OFFSET: usize = 6; +/// Offset of the flags in a record's header. +const FLAGS_OFFSET: usize = 7; +/// Offset of the instance ID in a record's header. +const INSTANCE_OFFSET: usize = 8; +/// Offset of the payload length in a record's header. +const LENGTH_OFFSET: usize = 40; + +/// The broker's guest leg. It resets the guest once the guest attaches, numbers its records to +/// the guest from 0 in each epoch, and sends DATA only while the guest's receive window has room. +/// It checks the identity and sequence number of each of the guest's records, and panics on any +/// record that the protocol does not allow. +struct Broker { + /// Stream to the guest. + stream: UnixStream, + /// The guest's records, which a thread reads as they arrive, so that the guest never waits to + /// write while the broker writes; then `None`, once the guest closed its stream at a record + /// boundary. + records: Receiver>, + /// Identity of the current epoch. + identity: Identity, + /// Sequence number of the broker's next record. + send_sequence: u64, + /// Sequence number of the guest's next record. + receive_sequence: u64, + /// Bytes for which the guest's receive window has room. + room: usize, + /// Decoder of the guest's messages. + decoder: Decoder, +} + +impl Broker { + /// Creates the broker on its end of a socket pair. + fn new(stream: UnixStream) -> Self { + let reader = stream.try_clone().unwrap(); + let (sender, records) = mpsc::channel(); + thread::spawn(move || read_records(reader, &sender)); + Self { + stream, + records, + identity: FIRST, + send_sequence: 0, + receive_sequence: 0, + room: 0, + decoder: Decoder::new(), + } + } + + /// Takes the guest's attachment, then starts the epoch of `identity`. + fn attach(&mut self, identity: Identity) { + let record = self.read().expect("the guest attaches"); + assert_eq!(record.record_type(), RecordType::GuestAttach); + let header = (record.instance_id(), record.epoch(), record.sequence()); + assert_eq!(header, ([0; 16], 0, 0)); + self.reset(identity); + } + + /// Starts the epoch of `identity` with a RESET, and takes the guest's acknowledgement. + fn reset(&mut self, identity: Identity) { + self.identity = identity; + self.send_sequence = 0; + self.receive_sequence = 0; + self.room = 0; + self.send(RecordType::Reset, &[]); + let ack = self.expect(); + assert_eq!(ack.record_type(), RecordType::Ack); + assert_eq!(ack.window(), Some(RECEIVE_WINDOW)); + self.room = RECEIVE_WINDOW as usize; + } + + /// Sends a record of the current epoch at the broker's next sequence number. + fn send(&mut self, record_type: RecordType, payload: &[u8]) { + let record = record(record_type, self.identity, self.send_sequence, payload); + self.send_sequence += 1; + self.write(&record); + } + + /// Writes bytes to the guest. + fn write(&mut self, bytes: &[u8]) { + self.stream.write_all(bytes).unwrap(); + } + + /// Ends the broker's stream to the guest. + fn close(&self) { + self.stream.shutdown(Shutdown::Write).unwrap(); + } + + /// Returns the guest's next record, or `None` once the guest closed its stream at a record + /// boundary. + fn read(&mut self) -> Option { + self.records + .recv_timeout(TIMEOUT) + .expect("the guest sends a record or ends its stream in time") + } + + /// Reads the guest's next record, and checks that it carries the epoch's identity and the + /// guest's next sequence number. + fn expect(&mut self) -> Record { + let record = self.read().expect("the guest sends a record"); + let header = (record.instance_id(), record.epoch(), record.sequence()); + let expected = ( + self.identity.instance_id, + self.identity.epoch, + self.receive_sequence, + ); + assert_eq!(header, expected, "{record:?}"); + self.receive_sequence += 1; + record + } + + /// Takes the guest's next record: a CREDIT, which makes room in its window, or DATA, whose + /// complete messages it returns. + fn take(&mut self) -> Vec { + let record = self.expect(); + match record.record_type() { + RecordType::Credit => { + self.room += record.credit().unwrap() as usize; + assert!( + self.room <= RECEIVE_WINDOW as usize, + "the guest credited more than it received" + ); + Vec::new() + } + RecordType::Data => decode(&mut self.decoder, record.payload()), + record_type => panic!("the guest sent {record_type} in an active epoch"), + } + } + + /// Sends the host's `bytes` in DATA records of at most `size` bytes, each once the guest's + /// window has room for it, and returns the messages that the guest sent meanwhile. + fn send_data(&mut self, bytes: &[u8], size: usize) -> Vec { + let mut messages = Vec::new(); + for chunk in bytes.chunks(size) { + while self.room < chunk.len() { + messages.extend(self.take()); + } + self.room -= chunk.len(); + self.send(RecordType::Data, chunk); + } + messages + } + + /// Sends `messages` in DATA records of at most `size` bytes, and returns the guest's next + /// `replies` messages. + fn exchange(&mut self, messages: &[Message], size: usize, replies: usize) -> Vec { + let mut received = self.send_data(&encode(messages), size); + while received.len() < replies { + received.extend(self.take()); + } + assert_eq!(received.len(), replies, "{received:?}"); + received + } + + /// Takes the guest's credit for everything that the broker sent, so that no record of the + /// epoch is in flight. + fn settle(&mut self) { + while self.room < RECEIVE_WINDOW as usize { + assert_eq!(self.take(), [], "the guest sent no message"); + } + } +} + +/// Reads the guest's records from `stream` and sends them, then `None` once the guest closed its +/// stream at a record boundary. +fn read_records(mut stream: UnixStream, records: &Sender>) { + let mut parser = RecordParser::new(); + let mut buffer = [0; 4096]; + loop { + let wanted = parser.needed().min(buffer.len()); + let count = match stream.read(&mut buffer[..wanted]) { + Ok(count) => count, + // The guest closed its end before it read everything that the broker wrote, as after + // a record that it refused. Linux returns what the guest wrote first. + Err(error) if error.kind() == io::ErrorKind::ConnectionReset => 0, + Err(error) => panic!("cannot read the guest's stream: {error}"), + }; + if count == 0 { + assert_eq!( + parser.buffered(), + 0, + "the guest ended its stream inside a record" + ); + let _ = records.send(None); + return; + } + let (consumed, record) = parser.push(&buffer[..count]).unwrap(); + assert_eq!(consumed, count); + // A test that ended takes no more records. + if record.is_some_and(|record| records.send(Some(record)).is_err()) { + return; + } + } +} + +/// Encodes a valid record. +fn record(record_type: RecordType, identity: Identity, sequence: u64, payload: &[u8]) -> Vec { + let mut record = Vec::new(); + let (instance, epoch) = (identity.instance_id, identity.epoch); + encode_record( + record_type, + &instance, + epoch, + sequence, + payload, + &mut record, + ) + .unwrap(); + record +} + +/// Returns the encodings of `messages`, one after the other. +fn encode(messages: &[Message]) -> Vec { + messages + .iter() + .flat_map(|message| message.encode().unwrap()) + .collect() +} + +/// Returns the messages that `bytes` complete. +fn decode(decoder: &mut Decoder, mut bytes: &[u8]) -> Vec { + let mut messages = Vec::new(); + while !bytes.is_empty() { + let (consumed, message) = decoder.push(bytes).unwrap(); + bytes = &bytes[consumed..]; + messages.extend(message); + } + messages +} + +/// Returns an echo of the longest body, which spans five records. +fn longest() -> Message { + Message::Echo((0..=u8::MAX).cycle().take(MAX_BODY).collect()) +} + +/// Starts the guest half in a thread, on one end of a socket pair, and returns the broker on the +/// other end with the thread. +fn start() -> (Broker, JoinHandle>) { + let (broker, guest) = UnixStream::pair().unwrap(); + let serving = thread::spawn(move || guest::serve(guest)); + (Broker::new(broker), serving) +} + +/// Starts the guest half, attaches it in the first epoch, sends it `bytes`, and returns the +/// error that ended serving. +fn refused(bytes: &[u8]) -> ServeError { + let (mut broker, serving) = start(); + broker.attach(FIRST); + broker.write(bytes); + let error = serving.join().unwrap().unwrap_err(); + // The guest ended its stream. + while let Some(record) = broker.read() { + assert_eq!(record.record_type(), RecordType::Credit, "{error:?}"); + } + error +} + +#[test] +fn the_guest_half_answers_messages_that_span_records() { + let (mut broker, serving) = start(); + broker.attach(FIRST); + let replies = broker.exchange(&[Message::Ping], OUTER_MAX_PAYLOAD, 1); + assert_eq!(replies, [Message::Pong]); + + // Messages in records of several sizes. The longest spans five records each way, and the + // guest's window holds a quarter of it, so the rest arrives as the guest credits it. + let messages = [ + Message::Ping, + Message::Echo(b"two".to_vec()), + longest(), + Message::Echo(Vec::new()), + Message::Ping, + ]; + let expected = [ + Message::Pong, + Message::Echo(b"two".to_vec()), + longest(), + Message::Echo(Vec::new()), + Message::Pong, + ]; + for size in [7, 4096, OUTER_MAX_PAYLOAD] { + let replies = broker.exchange(&messages, size, expected.len()); + assert_eq!(replies, expected, "records of {size} bytes"); + } + + // A record whose bytes arrive one at a time. + broker.settle(); + let ping = Message::Ping.encode().unwrap(); + let bytes = record(RecordType::Data, FIRST, broker.send_sequence, &ping); + broker.send_sequence += 1; + broker.room -= ping.len(); + for byte in bytes { + broker.write(&[byte]); + thread::sleep(Duration::from_millis(1)); + } + let mut replies = Vec::new(); + while replies.is_empty() { + replies = broker.take(); + } + assert_eq!(replies, [Message::Pong]); + + let replies = broker.exchange(&[Message::Stop], OUTER_MAX_PAYLOAD, 1); + assert_eq!(replies, [Message::Stopped]); + assert_eq!(serving.join().unwrap().unwrap(), Served::Stopped); + assert!(broker.read().is_none()); +} + +#[test] +fn a_reset_starts_the_new_epoch_with_a_new_message() { + let (mut broker, serving) = start(); + broker.attach(FIRST); + let echo = Message::Echo(vec![1; 1000]).encode().unwrap(); + + // Half of a message, then the host's connection drops, which starts the instance's next + // epoch. The guest drops the partial message. + assert_eq!(broker.send_data(&echo[..500], OUTER_MAX_PAYLOAD), []); + broker.settle(); + broker.reset(Identity { epoch: 2, ..FIRST }); + let after = Message::Echo(b"after".to_vec()); + let replies = broker.exchange(std::slice::from_ref(&after), OUTER_MAX_PAYLOAD, 1); + assert_eq!(replies, [after]); + + // A broker restored from a snapshot has a new instance, which starts at epoch 1, here in + // the middle of a header. + assert_eq!( + broker.send_data(&echo[..HEADER_LEN - 1], OUTER_MAX_PAYLOAD), + [] + ); + broker.settle(); + broker.reset(Identity { + instance_id: [0x7e; 16], + epoch: 1, + }); + let replies = broker.exchange(&[Message::Ping, Message::Stop], OUTER_MAX_PAYLOAD, 2); + assert_eq!(replies, [Message::Pong, Message::Stopped]); + assert_eq!(serving.join().unwrap().unwrap(), Served::Stopped); +} + +#[test] +fn the_guest_half_ends_with_its_stream() { + // A stream that ends at a record boundary closes the session. + let (mut broker, serving) = start(); + broker.attach(FIRST); + let replies = broker.exchange(&[Message::Ping], OUTER_MAX_PAYLOAD, 1); + assert_eq!(replies, [Message::Pong]); + broker.close(); + assert_eq!(serving.join().unwrap().unwrap(), Served::Closed); + assert!(broker.read().is_none()); + + // One that ends inside a record fails it. + let (mut broker, serving) = start(); + broker.attach(FIRST); + let ping = record(RecordType::Data, FIRST, 1, &Message::Ping.encode().unwrap()); + broker.write(&ping[..ping.len() - 1]); + broker.close(); + let error = serving.join().unwrap().unwrap_err(); + let truncated = matches!( + &error, + ServeError::Session(GuestConnectionError::Protocol(SessionError::Truncated { buffered })) + if *buffered == ping.len() - 1 + ); + assert!(truncated, "{error:?}"); +} + +#[test] +fn the_guest_half_fails_on_records_that_the_protocol_does_not_allow() { + let ping = Message::Ping.encode().unwrap(); + let data = |identity, sequence| record(RecordType::Data, identity, sequence, &ping); + let next_epoch = Identity { epoch: 2, ..FIRST }; + let other_instance = Identity { + instance_id: [0x7e; 16], + ..FIRST + }; + let patched = |mut bytes: Vec, offset: usize, patch: &[u8]| { + bytes[offset..offset + patch.len()].copy_from_slice(patch); + bytes + }; + // A RESET with a payload, an empty DATA record, and a GUEST_ATTACH with an identity, which + // the record API cannot encode. + let mut reset_with_payload = patched( + record(RecordType::Reset, next_epoch, 1, &[]), + LENGTH_OFFSET, + &4u32.to_le_bytes(), + ); + reset_with_payload.extend_from_slice(&[0; 4]); + let mut empty_data = patched(data(FIRST, 1), LENGTH_OFFSET, &0u32.to_le_bytes()); + empty_data.truncate(empty_data.len() - ping.len()); + let attach_with_identity = patched( + record( + RecordType::GuestAttach, + Identity { + instance_id: [0; 16], + epoch: 0, + }, + 0, + &[], + ), + INSTANCE_OFFSET, + &[1], + ); + let error_record = |code: u32| { + let payload = ErrorCode::Authentication.code().to_le_bytes(); + patched( + record(RecordType::Error, FIRST, 1, &payload), + LENGTH_OFFSET + 4, + &code.to_le_bytes(), + ) + }; + let wrong_magic = patched(data(FIRST, 1), 0, b"NVXT"); + let cases = [ + ( + "a skipped sequence number", + data(FIRST, 2), + SessionError::Sequence { + expected: 1, + actual: 2, + }, + ), + ( + "a repeated sequence number", + data(FIRST, 0), + SessionError::Sequence { + expected: 1, + actual: 0, + }, + ), + ( + "DATA of another epoch", + data(next_epoch, 1), + SessionError::Epoch { + expected: 1, + actual: 2, + }, + ), + ( + "DATA of another instance", + data(other_instance, 1), + SessionError::Instance { + expected: FIRST.instance_id, + actual: other_instance.instance_id, + }, + ), + ( + "a RESET that repeats the epoch", + record(RecordType::Reset, FIRST, 1, &[]), + SessionError::StaleEpoch { + current: 1, + actual: 1, + }, + ), + ( + "a RESET without an identity", + record( + RecordType::Reset, + Identity { + instance_id: [0; 16], + epoch: 0, + }, + 1, + &[], + ), + SessionError::MissingIdentity(RecordType::Reset), + ), + ( + "a RESET with a payload", + reset_with_payload, + SessionError::Record(RecordError::Length { + record_type: RecordType::Reset, + length: 4, + }), + ), + ( + "an empty DATA record", + empty_data, + SessionError::Record(RecordError::Length { + record_type: RecordType::Data, + length: 0, + }), + ), + ( + "a bootstrap record with an identity", + attach_with_identity, + SessionError::Record(RecordError::BootstrapIdentity(RecordType::GuestAttach)), + ), + ( + "an ERROR record with an unknown code", + error_record(2), + SessionError::Record(RecordError::ErrorCode(2)), + ), + ( + "an ERROR record", + error_record(ErrorCode::Authentication.code()), + SessionError::Unexpected(RecordType::Error), + ), + ( + "an ACK record", + record(RecordType::Ack, FIRST, 1, &RECEIVE_WINDOW.to_le_bytes()), + SessionError::Unexpected(RecordType::Ack), + ), + ( + "an unknown record type", + patched(data(FIRST, 1), TYPE_OFFSET, &[0x7f]), + SessionError::Record(RecordError::Type(0x7f)), + ), + ( + "nonzero flags", + patched(data(FIRST, 1), FLAGS_OFFSET, &[1]), + SessionError::Record(RecordError::Flags(1)), + ), + ( + "another protocol version", + patched(data(FIRST, 1), VERSION_OFFSET, &2u16.to_le_bytes()), + SessionError::Record(RecordError::Version(2)), + ), + ( + "a wrong magic", + wrong_magic, + SessionError::Record(RecordError::Magic(*b"NVXT")), + ), + ]; + for (name, bytes, expected) in cases { + let error = refused(&bytes); + let failed = matches!( + &error, + ServeError::Session(GuestConnectionError::Protocol(error)) if *error == expected + ); + assert!(failed, "{name}: {error:?}"); + } +} + +#[test] +fn the_guest_half_refuses_messages_that_its_framing_does_not_allow() { + // A message is refused at its header, before its body arrives: the decoder never buffers a + // body beyond the limit. + let too_long = u32::try_from(MAX_BODY + 1).unwrap().to_le_bytes(); + let header = |length: [u8; 4], kind: u8| { + let mut header = length.to_vec(); + header.push(kind); + header + }; + let cases = [ + (header([0; 4], 0), FrameError::Kind(0)), + (header([0; 4], 9), FrameError::Kind(9)), + ( + header(too_long, 3), + FrameError::Length { + kind: 3, + length: MAX_BODY + 1, + }, + ), + ( + header(1u32.to_le_bytes(), 1), + FrameError::Length { kind: 1, length: 1 }, + ), + ]; + for (message, expected) in cases { + let error = refused(&record(RecordType::Data, FIRST, 1, &message)); + let failed = matches!(&error, ServeError::Frame(error) if *error == expected); + assert!(failed, "{message:?}: {error:?}"); + } + + // The guest half answers requests, and refuses the messages that only it sends. + for message in [Message::Pong, Message::Stopped] { + let bytes = record(RecordType::Data, FIRST, 1, &message.encode().unwrap()); + let error = refused(&bytes); + let failed = matches!(&error, ServeError::Unexpected(kind) if *kind == message.kind()); + assert!(failed, "{message:?}: {error:?}"); + } +} + +/// Starts the guest binary's `serve` with one end of a socket pair as its standard input and +/// output, and returns the broker on the other end with the process. +fn start_binary() -> (Broker, Child) { + let (broker, guest) = UnixStream::pair().unwrap(); + let output = guest.try_clone().unwrap(); + let child = Command::new(env!("CARGO_BIN_EXE_runtime-adapter-guest")) + .arg("serve") + .stdin(Stdio::from(OwnedFd::from(guest))) + .stdout(Stdio::from(OwnedFd::from(output))) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + (Broker::new(broker), child) +} + +#[test] +fn the_guest_binary_serves_the_session_on_its_standard_streams() { + // The host stops the guest. + let (mut broker, child) = start_binary(); + broker.attach(FIRST); + let echo = Message::Echo(vec![0x42; OUTER_MAX_PAYLOAD]); + let messages = [Message::Ping, echo.clone(), Message::Stop]; + let replies = broker.exchange(&messages, OUTER_MAX_PAYLOAD, 3); + assert_eq!(replies, [Message::Pong, echo, Message::Stopped]); + let output = child.wait_with_output().unwrap(); + assert!(output.status.success(), "{output:?}"); + assert!(output.stderr.is_empty(), "{output:?}"); + + // The broker closes the stream at a record boundary. + let (mut broker, child) = start_binary(); + broker.attach(FIRST); + broker.close(); + let output = child.wait_with_output().unwrap(); + assert!(output.status.success(), "{output:?}"); + assert!(output.stderr.is_empty(), "{output:?}"); + + // The host sends a malformed message. + let (mut broker, child) = start_binary(); + broker.attach(FIRST); + broker.send_data(&[0, 0, 0, 0, 9], OUTER_MAX_PAYLOAD); + let output = child.wait_with_output().unwrap(); + assert_eq!(output.status.code(), Some(1), "{output:?}"); + let diagnostic = String::from_utf8_lossy(&output.stderr); + assert_eq!( + diagnostic, + "runtime-adapter-guest: unknown message kind 9\n" + ); +} From d21f93e287f3dd290a17c47aca1e928a9c0743fc Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 20 Oct 2026 17:12:02 -0700 Subject: [PATCH 2/6] tests: describe the runtime adapter without planning terms The runtime adapter's documentation, comments, and test names referred to scenario IDs and later steps that only planning issues define, called its parts halves, and named the control session after its protocol. Describe each part by its role instead: the host program, the guest agent, and the control session between the host and the guest. nvx::guest::nvxs remains as the API path that the guest agent uses, and the contract string keeps its protocol versions. The CI and contribution guides, the check-nvx-rs action, and the microVM workflow describe the adapter the same way. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 086f45a6-c39c-4704-85ef-1fa7f8b8b672 --- .github/actions/check-nvx-rs/action.yml | 6 +- .github/workflows/run-nvx-microvm-tests.yml | 7 +- doc/ci.md | 19 ++--- doc/contribute.md | 4 +- doc/project-structure.md | 10 +-- tests/runtime-adapter/Cargo.toml | 4 +- tests/runtime-adapter/README.md | 85 +++++++++---------- .../src/bin/runtime-adapter-guest.rs | 13 +-- .../src/bin/runtime-adapter-host.rs | 5 +- tests/runtime-adapter/src/frame.rs | 14 +-- tests/runtime-adapter/src/guest.rs | 23 ++--- tests/runtime-adapter/src/lib.rs | 25 +++--- tests/runtime-adapter/tests/guest_session.rs | 39 ++++----- .../tests/independent_build.rs | 2 +- 14 files changed, 125 insertions(+), 131 deletions(-) diff --git a/.github/actions/check-nvx-rs/action.yml b/.github/actions/check-nvx-rs/action.yml index 0cbae299f..cad79c383 100644 --- a/.github/actions/check-nvx-rs/action.yml +++ b/.github/actions/check-nvx-rs/action.yml @@ -143,8 +143,8 @@ runs: python3 ../scripts/nvx.py test-guest-binaries --binaries "${directory}" # The runtime adapter builds on nvx-rs as an independent adapter does: with a lockfile of its - # own, default features off, and the public API only. Its tests check consumer-contract - # scenario C0 in each configuration, which also builds with nvx-rs's minimum Rust version. + # own, default features off, and the public API only. Its tests check that each configuration + # builds that way, and each configuration also builds with nvx-rs's minimum Rust version. - name: Check the runtime adapter's protocol-only configuration shell: bash working-directory: tests/runtime-adapter @@ -168,7 +168,7 @@ runs: cargo "+${{ steps.toolchains.outputs.toolchain }}" test --locked --features host cargo "+${{ steps.toolchains.outputs.msrv }}" check --locked --all-targets --features host - # The guest half runs in Linux guests, as a static executable that a test checks. + # The guest agent runs in Linux guests, as a static executable that a test checks. - name: Check the runtime adapter's guest-only configuration if: runner.os == 'Linux' shell: bash diff --git a/.github/workflows/run-nvx-microvm-tests.yml b/.github/workflows/run-nvx-microvm-tests.yml index 3deb79666..350f129d0 100644 --- a/.github/workflows/run-nvx-microvm-tests.yml +++ b/.github/workflows/run-nvx-microvm-tests.yml @@ -1180,10 +1180,9 @@ jobs: --backend "${{ inputs.backend }}" --output-dir "build/test-results/aci-edge-sandboxes-${{ inputs.backend }}" - # The hook of the runtime adapter's scenarios on this backend: the ignored - # tests of its host-only configuration, which later steps add and which - # read the backend through `nvx::Hypervisor::from_env_or_default`. There - # are none yet, so these steps only build the adapter. + # The runtime adapter's scenarios on this backend: the ignored tests of its + # host-only configuration, which read the backend through + # `nvx::Hypervisor::from_env_or_default`. - name: Run runtime adapter scenarios on Linux if: runner.os != 'Windows' && !inputs.debug-kernel shell: bash diff --git a/doc/ci.md b/doc/ci.md index 7da2a3cc8..d80f4176a 100644 --- a/doc/ci.md +++ b/doc/ci.md @@ -657,13 +657,14 @@ lockfile of its own, default features off, and the public API only. For each of its configurations, protocol-only, host-only, and, on Linux, guest-only for `x86_64-unknown-linux-musl`, the action runs Clippy with warnings denied, the build, the tests, and a check with the minimum Rust version of `nvx`, after -rustfmt. The tests check consumer-contract scenario C0: each configuration -enables only its side of `nvx`, the guest-only build pulls in neither Tokio -nor a dependency that only the host side declares, and the guest binary is a -static executable. The guest-only configuration also checks the guest's part -of scenario C1: the adapter's guest half and its binary serve the NVXS session -with the adapter's own framing against a test broker, which sends them -messages that span records, resets, disconnects, and malformed records. +rustfmt. The tests check that each configuration builds as an independent +adapter's does: it enables only its side of `nvx`, the guest-only build pulls +in neither Tokio nor a dependency that only the host side declares, and the +guest binary is a static executable. In the guest-only configuration, they +also check that the adapter's guest agent serves the control session between +the host and the guest with the adapter's own framing, against a test broker +that sends it messages that span records, resets, disconnects, and malformed +records. The development release job depends on `aci-edge-sandboxes` and `nvx-rs` and requires their combined Linux/Windows results to be successful; failed, @@ -685,9 +686,7 @@ therefore trigger the NVX microVM test matrix. The same jobs, except the debug-kernel ones, run the runtime adapter's scenarios on their backend: the ignored tests of its host-only configuration, -with `NVX_HYPERVISOR` naming the backend. Later steps of -[microsoft/nvx#535](https://github.com/microsoft/nvx/issues/535) add them; until -then, the step only builds the adapter. Changes under `tests/runtime-adapter/` +with `NVX_HYPERVISOR` naming the backend. Changes under `tests/runtime-adapter/` therefore trigger the NVX microVM test matrix too. ## Adversarial campaigns diff --git a/doc/contribute.md b/doc/contribute.md index 08b71fefe..0d62c2f49 100644 --- a/doc/contribute.md +++ b/doc/contribute.md @@ -116,8 +116,8 @@ runs `tests/guest_tools.rs` against the exact guest-profile binaries. Tests of t mechanics, such as the golden fixtures of the control protocol, belong in `nvx`, and tests of the ACI Edge Sandboxes contract in `aci_edge_sandboxes`. Tests of what an independent adapter observes through -the public API of `nvx` alone, such as the consumer-contract scenarios of -[microsoft/nvx#535](https://github.com/microsoft/nvx/issues/535), belong in the +the public API of `nvx` alone, such as its own framing over the control +session between the host and the guest, belong in the [runtime adapter](../tests/runtime-adapter/README.md), whose ignored tests the microVM jobs run on each backend. The `scripts/nvx.py test-aci-edge-sandboxes --backend BACKEND` command runs all ignored `openvmm_e2e` tests on a real hypervisor, diff --git a/doc/project-structure.md b/doc/project-structure.md index 9b2d944bb..70d178b55 100644 --- a/doc/project-structure.md +++ b/doc/project-structure.md @@ -161,8 +161,8 @@ nvx/ | `-- test_*.py Python tooling unit tests |-- tests/ Tests that build on the repository's crates from outside them | `-- runtime-adapter/ Rust crate `runtime-adapter`, an independent adapter of nvx -| |-- src/ Its framing, its guest half, and its host and guest binaries -| `-- tests/ Consumer-contract scenario C0, and the guest's part of C1 +| |-- src/ Its framing, its host program, and its guest agent +| `-- tests/ Its independent build, and its guest agent against a test broker |-- SECURITY.md Microsoft security vulnerability reporting policy |-- .dockerignore Docker build-context exclusions |-- .gitattributes Git path attributes @@ -335,9 +335,9 @@ Tests that build on the repository's crates from outside them. crate with its own `Cargo.lock` that depends on `nvx` with default features off and uses its public API only. Its protocol-only, host-only, and guest-only configurations build in CI, and its tests check that each one builds as an -independent adapter's does. Its guest half serves the guest's role of the NVXS -session with the adapter's own framing, which its tests check against a test -broker. See its [README](../tests/runtime-adapter/README.md). +independent adapter's does. Its guest agent serves the control session between +the host and the guest with the adapter's own framing, which its tests check +against a test broker. See its [README](../tests/runtime-adapter/README.md). ## Root files diff --git a/tests/runtime-adapter/Cargo.toml b/tests/runtime-adapter/Cargo.toml index 3506cfc58..076a8c1d2 100644 --- a/tests/runtime-adapter/Cargo.toml +++ b/tests/runtime-adapter/Cargo.toml @@ -14,9 +14,9 @@ publish = false # No default features: each configuration selects one side of `nvx`, as an independent adapter # does. Without features, the adapter builds against the protocol-only `nvx`. default = [] -# The host half: the host API of `nvx` and the `runtime-adapter-host` binary. +# The host program: the host API of `nvx` and the `runtime-adapter-host` binary. host = ["nvx/host"] -# The guest half: the guest API of `nvx` and the `runtime-adapter-guest` binary, which builds as a +# The guest agent: the guest API of `nvx` and the `runtime-adapter-guest` binary, which builds as a # static `x86_64-unknown-linux-musl` executable. guest = ["nvx/guest"] diff --git a/tests/runtime-adapter/README.md b/tests/runtime-adapter/README.md index 9e7b7af46..a9a38acbd 100644 --- a/tests/runtime-adapter/README.md +++ b/tests/runtime-adapter/README.md @@ -1,24 +1,19 @@ # Runtime adapter -The runtime adapter is an independent adapter of the NVX runtime. It lives in -this repository so that NVX's own CI builds and runs it, and it depends on -[`nvx`](../../nvx-rs/README.md) the way an adapter outside the repository -does: +The runtime adapter is an independent adapter of the NVX runtime, with a host +program and a guest agent. It lives in this repository so that NVX's own CI +builds and runs it, and it depends on [`nvx`](../../nvx-rs/README.md) the way +an adapter outside the repository does: - as a path dependency with default features off; - with a `Cargo.lock` of its own, outside any other workspace; - through the public API alone, with unsafe code forbidden; - with the minimum supported Rust version of `nvx`. -[microsoft/nvx#535](https://github.com/microsoft/nvx/issues/535) sets the -contract that it proves: an adapter brings its own application framing over -the NVXS session, its own init, and its own status collector. Its guest half -serves the guest's role of the NVXS session with the adapter's own framing. -Its host binary is still a skeleton that reports the contract that it was -built against. Later steps give the guest half the rest of the guest API -([microsoft/nvx#619](https://github.com/microsoft/nvx/issues/619)), fill the -host half ([microsoft/nvx#620](https://github.com/microsoft/nvx/issues/620)), -and add the scenarios that run both halves on a hypervisor. +The adapter brings its own application framing to the control session between +the host and the guest, in place of NVX's `NVXC` frames. Its guest agent serves +requests in that framing. Its host program reports the contract that it was +built against. ## Configurations @@ -32,10 +27,10 @@ Each configuration enables one side of `nvx`: ## Framing -The `frame` module is the adapter's application framing, which its halves -exchange in the payloads of the session's DATA records in place of NVX's -`NVXC` frames. A message is a 5-byte header, then a body. The header holds the -body's length, a little-endian `u32`, then the message's kind: +The `frame` module is the adapter's application framing, which its host program +and guest agent exchange over the control session in place of NVX's `NVXC` +frames. A message is a 5-byte header, then a body. The header holds the body's +length, a little-endian `u32`, then the message's kind: | Kind | Message | Body | | --- | --- | --- | @@ -45,22 +40,21 @@ body's length, a little-endian `u32`, then the message's kind: | 4 | `Stop`: asks the guest to stop serving | none | | 5 | `Stopped`: the guest's confirmation of `Stop`, its last message | none | -The session keeps no message boundaries, so a message may span several DATA +The session keeps no message boundaries, so a message may span several of its records, and one record may end a message and start the next. `Decoder` reassembles the messages. It checks each header before it takes any byte of the body, and holds at most one partial message. A malformed message ends the stream, because the framing cannot find the next message after it. -## Guest half +## Guest agent -`guest::serve` serves the guest's role of the NVXS session on a stream that -connects to the broker's guest leg, through `nvx::guest::nvxs`. It attaches, -then: +`guest::serve` serves the host program on a stream to the guest's end of the +control session, through `nvx::guest::nvxs`. It attaches, then: -- acknowledges each RESET with the smallest receive window that NVXS allows, - once it dropped the earlier epoch's partial message, because a new epoch - starts with a new message; -- credits each DATA payload back once it holds the payload, so that a message +- acknowledges each reset of the session with the smallest receive window that + the session allows, once it dropped the earlier epoch's partial message, + because a new epoch starts with a new message; +- credits each payload back once it holds the payload, so that a message longer than the window still arrives whole; - answers each `Ping` with `Pong`, sends each `Echo` back, and answers `Stop` with `Stopped`, then returns. @@ -68,16 +62,17 @@ then: A failure of the session, a malformed message, and a message that only the guest sends end serving with an error. -`runtime-adapter-guest serve` runs the guest half on its standard input and -output, which connect to the broker's guest leg, such as a control console in -raw mode. It exits with 0 once the host stopped it or the broker closed the -stream, and with 1, after a diagnostic on standard error, once serving failed. -Without arguments, it reports the contract that it was built against. +`runtime-adapter-guest serve` runs the guest agent on its standard input and +output, which connect to the guest's end of the session, such as a control +console in raw mode. It exits with 0 once the host stopped it or the broker +closed the stream, and with 1, after a diagnostic on standard error, once +serving failed. Without arguments, it reports the contract that it was built +against. ## Tests -`tests/independent_build.rs` checks consumer-contract scenario C0 in every -configuration: +`tests/independent_build.rs` checks that every configuration builds as an +independent adapter's does: - The adapter is a workspace of its own with its own lockfile, and declares the minimum supported Rust version of `nvx`. @@ -95,17 +90,17 @@ The unit tests of `frame` run in every configuration. They cover each kind's encoding, messages in pieces of any size and split at every byte, and the headers that the framing refuses before their body arrives. -`tests/guest_session.rs` checks the guest's part of consumer-contract scenario -C1, NVXS with independent framing, in the guest-only configuration. A test -broker plays the broker's guest leg over a Unix socket pair with the public -record API of `nvx`, for `guest::serve` and for `runtime-adapter-guest -serve`. It covers messages that span records both ways and fill the receive -window, a record whose bytes arrive one at a time, resets in the middle of a -message, with the next epoch of the instance and with a new instance, streams -that end at and inside a record, and the records and messages that the guest -half refuses: sequence, epoch, and instance errors, stale and identity-less -resets, record lengths, bootstrap identity, ERROR payloads, and malformed -headers. +`tests/guest_session.rs` checks, in the guest-only configuration, that the +guest agent serves the control session with the adapter's own framing. A test +broker plays the broker's end of the guest's session over a Unix socket pair +with the public record API of `nvx`, for `guest::serve` and for +`runtime-adapter-guest serve`. It covers messages that span records both ways +and fill the receive window, a record whose bytes arrive one at a time, resets +in the middle of a message, with the next epoch of the instance and with a new +instance, streams that end at and inside a record, and the records and +messages that the guest agent refuses: sequence, epoch, and instance errors, +stale and identity-less resets, record lengths, bootstrap identity, ERROR +payloads, and malformed headers. Run the checks of each configuration in this directory. The guest-only configuration needs Linux and the musl target, which @@ -133,4 +128,4 @@ NVX microVM tests. Their [workflow](../../.github/workflows/run-nvx-microvm-tests.yml) runs the adapter's hypervisor scenarios on each backend: its ignored tests in the host-only configuration, with `NVX_HYPERVISOR` naming the backend, which -`nvx::Hypervisor::from_env_or_default` reads. There are none yet. +`nvx::Hypervisor::from_env_or_default` reads. diff --git a/tests/runtime-adapter/src/bin/runtime-adapter-guest.rs b/tests/runtime-adapter/src/bin/runtime-adapter-guest.rs index 6d486ba27..7a14a138f 100644 --- a/tests/runtime-adapter/src/bin/runtime-adapter-guest.rs +++ b/tests/runtime-adapter/src/bin/runtime-adapter-guest.rs @@ -1,10 +1,11 @@ -//! The guest half of the runtime adapter, a static `x86_64-unknown-linux-musl` executable. +//! The runtime adapter's guest agent, a static `x86_64-unknown-linux-musl` executable. //! //! - `runtime-adapter-guest` reports the contract that it was built against. -//! - `runtime-adapter-guest serve` serves the guest's role of the NVXS session with the adapter's -//! own framing on its standard input and output, which connect to the broker's guest leg, such -//! as a control console in raw mode. It exits with 0 once the host stopped it or the broker -//! closed the stream, and with 1, after a diagnostic on standard error, once serving failed. +//! - `runtime-adapter-guest serve` serves the host program with the adapter's own framing on its +//! standard input and output, which connect to the guest's end of the control session between +//! the host and the guest, such as a control console in raw mode. It exits with 0 once the host +//! stopped it or the broker closed the stream, and with 1, after a diagnostic on standard error, +//! once serving failed. //! //! Anything else is a usage error, with status 2. @@ -69,7 +70,7 @@ impl std::io::Write for Stdio { } } -// The guest half runs only in Linux guests. Elsewhere this stub fails, so that every target still +// The guest agent runs only in Linux guests. Elsewhere this stub fails, so that every target still // builds with every feature. #[cfg(not(target_os = "linux"))] fn main() -> ExitCode { diff --git a/tests/runtime-adapter/src/bin/runtime-adapter-host.rs b/tests/runtime-adapter/src/bin/runtime-adapter-host.rs index ffaa3c85c..34471149b 100644 --- a/tests/runtime-adapter/src/bin/runtime-adapter-host.rs +++ b/tests/runtime-adapter/src/bin/runtime-adapter-host.rs @@ -1,6 +1,5 @@ -//! The host half of the runtime adapter. This skeleton reports the contract that it was built -//! against and whether `nvx` supports this host; later steps make it start, reattach to, and -//! stop sandboxes through the host API. +//! The runtime adapter's host program. It reports the contract that it was built against and +//! whether `nvx` supports this host. use std::process::ExitCode; diff --git a/tests/runtime-adapter/src/frame.rs b/tests/runtime-adapter/src/frame.rs index 677aaaf95..a91b3a17e 100644 --- a/tests/runtime-adapter/src/frame.rs +++ b/tests/runtime-adapter/src/frame.rs @@ -1,9 +1,9 @@ -//! The adapter's own application framing, which its halves exchange over the NVXS session where -//! NVX's own adapter exchanges `NVXC` frames. +//! The adapter's own application framing, which its host program and guest agent exchange over the +//! control session between the host and the guest, in place of NVX's `NVXC` frames. //! //! A message is a header of [`HEADER_LEN`] bytes, then a body. The header holds the body's length, -//! a little-endian `u32`, then the message's kind. NVXS carries bytes and keeps no message -//! boundaries, so a message may span several DATA records, and one record may end a message and +//! a little-endian `u32`, then the message's kind. The session carries bytes and keeps no message +//! boundaries, so a message may span several of its records, and one record may end a message and //! start the next. [`Decoder`] reassembles the messages: it checks each header before it takes any //! byte of the body, and holds at most one partial message, whose body has at most [`MAX_BODY`] //! bytes. @@ -16,7 +16,7 @@ use std::fmt; /// Bytes of a message's header: the body's length, then the message's kind. pub const HEADER_LEN: usize = 5; -/// Most bytes of a message's body, which is more than four NVXS DATA records carry. +/// Most bytes of a message's body, which is more than four of the session's records carry. pub const MAX_BODY: usize = 256 * 1024; /// A message of the adapter's framing. @@ -160,8 +160,8 @@ impl fmt::Display for FrameError { impl std::error::Error for FrameError {} -/// Reassembles messages from bytes that arrive in pieces of any size, such as the payloads of an -/// NVXS session's DATA records. +/// Reassembles messages from bytes that arrive in pieces of any size, such as the payloads of the +/// control session's records. #[derive(Clone, Debug, Default)] pub struct Decoder { /// Header of the message in progress. diff --git a/tests/runtime-adapter/src/guest.rs b/tests/runtime-adapter/src/guest.rs index 9781e3949..c81470d21 100644 --- a/tests/runtime-adapter/src/guest.rs +++ b/tests/runtime-adapter/src/guest.rs @@ -1,5 +1,6 @@ -//! The adapter's guest half: it serves the guest's role of the NVXS session with the adapter's -//! own [framing](crate::frame), through `nvx::guest::nvxs` alone. +//! The adapter's guest agent: it serves the host program over the guest's end of the control +//! session between the host and the guest, with the adapter's own [framing](crate::frame), through +//! `nvx::guest::nvxs` alone. use std::fmt; use std::io::{Read, Write}; @@ -10,8 +11,8 @@ use nvx::protocol::nvxs::GuestEvent; use crate::frame::{Decoder, FrameError, Message}; -/// The receive window that the guest acknowledges each epoch with: the smallest that NVXS allows, -/// so that the rest of a longer message arrives only as the guest credits what it took. +/// The receive window that the guest acknowledges each epoch with: the smallest that the session +/// allows, so that the rest of a longer message arrives only as the guest credits what it took. pub const RECEIVE_WINDOW: u32 = OUTER_MIN_RECEIVE_WINDOW; /// How serving ended, when nothing failed. @@ -27,7 +28,7 @@ pub enum Served { #[derive(Debug)] #[non_exhaustive] pub enum ServeError { - /// The NVXS session failed. + /// The control session failed. Session(GuestConnectionError), /// The host sent a malformed message. Frame(FrameError), @@ -72,15 +73,15 @@ impl From for ServeError { } } -/// Serves the guest's role of the NVXS session on `stream`, which connects to the broker's guest -/// leg, until the host stops the guest or the broker closes the stream. +/// Serves the host program on `stream`, the guest's end of the control session, until the host +/// stops the guest or the broker closes the stream. /// /// The guest attaches, then: /// -/// - It acknowledges each RESET with [`RECEIVE_WINDOW`], once it dropped the earlier epoch's -/// partial message: a new epoch starts with a new message. -/// - It credits each DATA payload back once it holds the payload, so that a message longer than -/// the window arrives whole. +/// - It acknowledges each reset of the session with [`RECEIVE_WINDOW`], once it dropped the +/// earlier epoch's partial message: a new epoch starts with a new message. +/// - It credits each payload back once it holds the payload, so that a message longer than the +/// window arrives whole. /// - It answers [`Message::Ping`] with [`Message::Pong`], sends each [`Message::Echo`] back, and /// answers [`Message::Stop`] with [`Message::Stopped`], then returns without reading further. /// diff --git a/tests/runtime-adapter/src/lib.rs b/tests/runtime-adapter/src/lib.rs index ca4a8e25b..602b28848 100644 --- a/tests/runtime-adapter/src/lib.rs +++ b/tests/runtime-adapter/src/lib.rs @@ -1,22 +1,21 @@ -//! An independent adapter of the NVX runtime. NVX's CI builds it against `nvx` the way an -//! adapter outside the repository would: as a path dependency with default features off, with a -//! lockfile of its own, and through the public API alone. +//! An independent adapter of the NVX runtime, with a host program and a guest agent. NVX's CI +//! builds it against `nvx` the way an adapter outside the repository would: as a path dependency +//! with default features off, with a lockfile of its own, and through the public API alone. //! //! Each configuration enables one side of `nvx`: //! //! - protocol-only, without features: the control protocol, the contract versions, and the other //! items that `nvx` has without features, with the adapter's own [`frame`]s; -//! - host-only, with the `host` feature: the host API too, and the `runtime-adapter-host` -//! binary; -//! - guest-only, with the `guest` feature: the guest API too, the adapter's guest half in its -//! `guest` module, and the `runtime-adapter-guest` binary, a static -//! `x86_64-unknown-linux-musl` executable. +//! - host-only, with the `host` feature: the host API too, and the host program, +//! `runtime-adapter-host`; +//! - guest-only, with the `guest` feature: the guest API too, and the guest agent, in the `guest` +//! module and the `runtime-adapter-guest` binary, a static `x86_64-unknown-linux-musl` +//! executable. //! -//! The adapter brings its own framing to the NVXS session, in place of NVX's `NVXC` frames. Its -//! guest half serves the guest's role of the session with it, and `runtime-adapter-guest serve` -//! runs the guest half on its standard input and output. The host binary is a skeleton that -//! reports the contract it was built against. Later steps fill the host half, give the guest half -//! the rest of the guest API, and add the scenarios that run both halves on a hypervisor. +//! The adapter brings its own framing to the control session between the host and the guest, in +//! place of NVX's `NVXC` frames. The guest agent serves requests in that framing, and +//! `runtime-adapter-guest serve` runs it on its standard input and output. The host program +//! reports the contract that it was built against. pub mod frame; #[cfg(all(feature = "guest", target_os = "linux"))] diff --git a/tests/runtime-adapter/tests/guest_session.rs b/tests/runtime-adapter/tests/guest_session.rs index 005482c8e..d0f9260b6 100644 --- a/tests/runtime-adapter/tests/guest_session.rs +++ b/tests/runtime-adapter/tests/guest_session.rs @@ -1,11 +1,12 @@ -//! Consumer-contract scenario C1, the guest's part: the adapter's guest half serves the guest's -//! role of the NVXS session with the adapter's own framing, through public APIs only. +//! The adapter's guest agent serves the control session between the host and the guest with the +//! adapter's own framing, through public APIs only. //! -//! A test broker plays the broker's guest leg over a Unix socket pair, with the public record API -//! of `nvx`: for the guest half in a thread, and for the guest binary, whose standard input and -//! output are the socket. The tests cover messages that span records, partial records, resets in -//! the middle of a message, disconnects, record lengths, bootstrap identity, ERROR payloads, -//! sequence and identity errors, and the bounded buffer of the adapter's decoder. +//! A test broker plays the broker's end of the guest's session over a Unix socket pair, with the +//! public record API of `nvx`: for the guest agent's `serve` in a thread, and for the +//! `runtime-adapter-guest` binary, whose standard input and output are the socket. The tests cover +//! messages that span records, partial records, resets in the middle of a message, disconnects, +//! record lengths, bootstrap identity, ERROR payloads, sequence and identity errors, and the +//! bounded buffer of the adapter's decoder. #![cfg(all(feature = "guest", target_os = "linux"))] use std::io::{self, Read, Write}; @@ -45,10 +46,10 @@ const INSTANCE_OFFSET: usize = 8; /// Offset of the payload length in a record's header. const LENGTH_OFFSET: usize = 40; -/// The broker's guest leg. It resets the guest once the guest attaches, numbers its records to -/// the guest from 0 in each epoch, and sends DATA only while the guest's receive window has room. -/// It checks the identity and sequence number of each of the guest's records, and panics on any -/// record that the protocol does not allow. +/// The broker's end of the guest's session. It resets the guest once the guest attaches, numbers +/// its records to the guest from 0 in each epoch, and sends DATA only while the guest's receive +/// window has room. It checks the identity and sequence number of each of the guest's records, +/// and panics on any record that the protocol does not allow. struct Broker { /// Stream to the guest. stream: UnixStream, @@ -271,15 +272,15 @@ fn longest() -> Message { Message::Echo((0..=u8::MAX).cycle().take(MAX_BODY).collect()) } -/// Starts the guest half in a thread, on one end of a socket pair, and returns the broker on the -/// other end with the thread. +/// Starts the guest agent's `serve` in a thread, on one end of a socket pair, and returns the +/// broker on the other end with the thread. fn start() -> (Broker, JoinHandle>) { let (broker, guest) = UnixStream::pair().unwrap(); let serving = thread::spawn(move || guest::serve(guest)); (Broker::new(broker), serving) } -/// Starts the guest half, attaches it in the first epoch, sends it `bytes`, and returns the +/// Starts the guest agent, attaches it in the first epoch, sends it `bytes`, and returns the /// error that ended serving. fn refused(bytes: &[u8]) -> ServeError { let (mut broker, serving) = start(); @@ -294,7 +295,7 @@ fn refused(bytes: &[u8]) -> ServeError { } #[test] -fn the_guest_half_answers_messages_that_span_records() { +fn the_guest_agent_answers_messages_that_span_records() { let (mut broker, serving) = start(); broker.attach(FIRST); let replies = broker.exchange(&[Message::Ping], OUTER_MAX_PAYLOAD, 1); @@ -375,7 +376,7 @@ fn a_reset_starts_the_new_epoch_with_a_new_message() { } #[test] -fn the_guest_half_ends_with_its_stream() { +fn the_guest_agent_ends_with_its_stream() { // A stream that ends at a record boundary closes the session. let (mut broker, serving) = start(); broker.attach(FIRST); @@ -401,7 +402,7 @@ fn the_guest_half_ends_with_its_stream() { } #[test] -fn the_guest_half_fails_on_records_that_the_protocol_does_not_allow() { +fn the_guest_agent_fails_on_records_that_the_protocol_does_not_allow() { let ping = Message::Ping.encode().unwrap(); let data = |identity, sequence| record(RecordType::Data, identity, sequence, &ping); let next_epoch = Identity { epoch: 2, ..FIRST }; @@ -567,7 +568,7 @@ fn the_guest_half_fails_on_records_that_the_protocol_does_not_allow() { } #[test] -fn the_guest_half_refuses_messages_that_its_framing_does_not_allow() { +fn the_guest_agent_refuses_messages_that_its_framing_does_not_allow() { // A message is refused at its header, before its body arrives: the decoder never buffers a // body beyond the limit. let too_long = u32::try_from(MAX_BODY + 1).unwrap().to_le_bytes(); @@ -597,7 +598,7 @@ fn the_guest_half_refuses_messages_that_its_framing_does_not_allow() { assert!(failed, "{message:?}: {error:?}"); } - // The guest half answers requests, and refuses the messages that only it sends. + // The guest agent answers requests, and refuses the messages that only it sends. for message in [Message::Pong, Message::Stopped] { let bytes = record(RecordType::Data, FIRST, 1, &message.encode().unwrap()); let error = refused(&bytes); diff --git a/tests/runtime-adapter/tests/independent_build.rs b/tests/runtime-adapter/tests/independent_build.rs index f2d433f7c..b0b51a307 100644 --- a/tests/runtime-adapter/tests/independent_build.rs +++ b/tests/runtime-adapter/tests/independent_build.rs @@ -1,4 +1,4 @@ -//! Consumer-contract scenario C0: the adapter builds on `nvx` independently. +//! The adapter builds on `nvx` independently, as an adapter outside the repository does. //! //! - It is a workspace of its own, with a lockfile of its own, and declares the minimum supported //! Rust version of `nvx`. From b129eb217f5870a40f81e1388d25e6cdec0482f8 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 20 Oct 2026 17:26:35 -0700 Subject: [PATCH 3/6] tests: bound the guest session tests' waits for the guest to end The tests waited without a limit for the guest agent's thread and for the guest binary to end. A regression that kept the guest serving, such as a frame check that no longer refuses a header, hung the test run instead of failing it. Wait at most the broker's receive timeout for each, then fail the test; a binary that is still running is killed and reaped first. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 086f45a6-c39c-4704-85ef-1fa7f8b8b672 --- tests/runtime-adapter/tests/guest_session.rs | 48 +++++++++++++++----- 1 file changed, 37 insertions(+), 11 deletions(-) diff --git a/tests/runtime-adapter/tests/guest_session.rs b/tests/runtime-adapter/tests/guest_session.rs index d0f9260b6..b86a55b5f 100644 --- a/tests/runtime-adapter/tests/guest_session.rs +++ b/tests/runtime-adapter/tests/guest_session.rs @@ -13,10 +13,10 @@ use std::io::{self, Read, Write}; use std::net::Shutdown; use std::os::fd::OwnedFd; use std::os::unix::net::UnixStream; -use std::process::{Child, Command, Stdio}; +use std::process::{Child, Command, Output, Stdio}; use std::sync::mpsc::{self, Receiver, Sender}; use std::thread::{self, JoinHandle}; -use std::time::Duration; +use std::time::{Duration, Instant}; use nvx::guest::nvxs::GuestConnectionError; use nvx::protocol::OUTER_MAX_PAYLOAD; @@ -32,7 +32,7 @@ const FIRST: Identity = Identity { epoch: 1, }; -/// Longest wait for a record from the guest. +/// Longest wait for a record from the guest, or for the guest to end. const TIMEOUT: Duration = Duration::from_secs(30); /// Offset of the protocol version in a record's header. @@ -280,13 +280,24 @@ fn start() -> (Broker, JoinHandle>) { (Broker::new(broker), serving) } +/// Returns how the guest agent's `serve` ended, and fails the test if it does not end in time, so +/// that a guest that keeps serving fails the test instead of hanging it. +fn joined(serving: JoinHandle>) -> Result { + let deadline = Instant::now() + TIMEOUT; + while !serving.is_finished() { + assert!(Instant::now() < deadline, "the guest agent ends in time"); + thread::sleep(Duration::from_millis(10)); + } + serving.join().unwrap() +} + /// Starts the guest agent, attaches it in the first epoch, sends it `bytes`, and returns the /// error that ended serving. fn refused(bytes: &[u8]) -> ServeError { let (mut broker, serving) = start(); broker.attach(FIRST); broker.write(bytes); - let error = serving.join().unwrap().unwrap_err(); + let error = joined(serving).unwrap_err(); // The guest ended its stream. while let Some(record) = broker.read() { assert_eq!(record.record_type(), RecordType::Credit, "{error:?}"); @@ -340,7 +351,7 @@ fn the_guest_agent_answers_messages_that_span_records() { let replies = broker.exchange(&[Message::Stop], OUTER_MAX_PAYLOAD, 1); assert_eq!(replies, [Message::Stopped]); - assert_eq!(serving.join().unwrap().unwrap(), Served::Stopped); + assert_eq!(joined(serving).unwrap(), Served::Stopped); assert!(broker.read().is_none()); } @@ -372,7 +383,7 @@ fn a_reset_starts_the_new_epoch_with_a_new_message() { }); let replies = broker.exchange(&[Message::Ping, Message::Stop], OUTER_MAX_PAYLOAD, 2); assert_eq!(replies, [Message::Pong, Message::Stopped]); - assert_eq!(serving.join().unwrap().unwrap(), Served::Stopped); + assert_eq!(joined(serving).unwrap(), Served::Stopped); } #[test] @@ -383,7 +394,7 @@ fn the_guest_agent_ends_with_its_stream() { let replies = broker.exchange(&[Message::Ping], OUTER_MAX_PAYLOAD, 1); assert_eq!(replies, [Message::Pong]); broker.close(); - assert_eq!(serving.join().unwrap().unwrap(), Served::Closed); + assert_eq!(joined(serving).unwrap(), Served::Closed); assert!(broker.read().is_none()); // One that ends inside a record fails it. @@ -392,7 +403,7 @@ fn the_guest_agent_ends_with_its_stream() { let ping = record(RecordType::Data, FIRST, 1, &Message::Ping.encode().unwrap()); broker.write(&ping[..ping.len() - 1]); broker.close(); - let error = serving.join().unwrap().unwrap_err(); + let error = joined(serving).unwrap_err(); let truncated = matches!( &error, ServeError::Session(GuestConnectionError::Protocol(SessionError::Truncated { buffered })) @@ -622,6 +633,21 @@ fn start_binary() -> (Broker, Child) { (Broker::new(broker), child) } +/// Returns the guest binary's output once it exits. If it does not exit in time, kills and reaps +/// it, and fails the test instead of hanging it. +fn finished(mut child: Child) -> Output { + let deadline = Instant::now() + TIMEOUT; + while child.try_wait().unwrap().is_none() { + if Instant::now() >= deadline { + let _ = child.kill(); + let output = child.wait_with_output().unwrap(); + panic!("the guest binary did not exit in time: {output:?}"); + } + thread::sleep(Duration::from_millis(10)); + } + child.wait_with_output().unwrap() +} + #[test] fn the_guest_binary_serves_the_session_on_its_standard_streams() { // The host stops the guest. @@ -631,7 +657,7 @@ fn the_guest_binary_serves_the_session_on_its_standard_streams() { let messages = [Message::Ping, echo.clone(), Message::Stop]; let replies = broker.exchange(&messages, OUTER_MAX_PAYLOAD, 3); assert_eq!(replies, [Message::Pong, echo, Message::Stopped]); - let output = child.wait_with_output().unwrap(); + let output = finished(child); assert!(output.status.success(), "{output:?}"); assert!(output.stderr.is_empty(), "{output:?}"); @@ -639,7 +665,7 @@ fn the_guest_binary_serves_the_session_on_its_standard_streams() { let (mut broker, child) = start_binary(); broker.attach(FIRST); broker.close(); - let output = child.wait_with_output().unwrap(); + let output = finished(child); assert!(output.status.success(), "{output:?}"); assert!(output.stderr.is_empty(), "{output:?}"); @@ -647,7 +673,7 @@ fn the_guest_binary_serves_the_session_on_its_standard_streams() { let (mut broker, child) = start_binary(); broker.attach(FIRST); broker.send_data(&[0, 0, 0, 0, 9], OUTER_MAX_PAYLOAD); - let output = child.wait_with_output().unwrap(); + let output = finished(child); assert_eq!(output.status.code(), Some(1), "{output:?}"); let diagnostic = String::from_utf8_lossy(&output.stderr); assert_eq!( From 7eaa4c580a5c96858d06cf3924fb0dc1adc3e016 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 20 Oct 2026 17:44:29 -0700 Subject: [PATCH 4/6] tests: bound the test broker's writes to the guest The guest session tests still waited without a limit when the guest stopped reading: the broker wrote with write_all on a blocking socket, and the socket's buffer can fill before the guest's receive window does, as with 7-byte records. Each write now has a deadline of TIMEOUT, so a guest that stops reading fails the test instead of hanging it. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 086f45a6-c39c-4704-85ef-1fa7f8b8b672 --- tests/runtime-adapter/tests/guest_session.rs | 26 +++++++++++++++++--- 1 file changed, 23 insertions(+), 3 deletions(-) diff --git a/tests/runtime-adapter/tests/guest_session.rs b/tests/runtime-adapter/tests/guest_session.rs index b86a55b5f..3c9e429c1 100644 --- a/tests/runtime-adapter/tests/guest_session.rs +++ b/tests/runtime-adapter/tests/guest_session.rs @@ -115,9 +115,29 @@ impl Broker { self.write(&record); } - /// Writes bytes to the guest. - fn write(&mut self, bytes: &[u8]) { - self.stream.write_all(bytes).unwrap(); + /// Writes bytes to the guest. A guest that stops reading fails the test once `TIMEOUT` passes, + /// instead of hanging it: the socket's buffer can fill before the guest's receive window does. + fn write(&mut self, mut bytes: &[u8]) { + let deadline = Instant::now() + TIMEOUT; + while !bytes.is_empty() { + let remaining = deadline.saturating_duration_since(Instant::now()); + assert!( + !remaining.is_zero(), + "the guest reads the broker's bytes in time" + ); + self.stream.set_write_timeout(Some(remaining)).unwrap(); + match self.stream.write(bytes) { + Ok(count) => bytes = &bytes[count..], + Err(error) + if matches!( + error.kind(), + io::ErrorKind::WouldBlock + | io::ErrorKind::TimedOut + | io::ErrorKind::Interrupted + ) => {} + Err(error) => panic!("cannot write to the guest: {error}"), + } + } } /// Ends the broker's stream to the guest. From d3472673b514ab825029cb5ca5c2bc16937c5608 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 20 Oct 2026 17:44:30 -0700 Subject: [PATCH 5/6] tests: describe the adapter's microVM step without claiming scenarios The CI docs said that the microVM jobs run the runtime adapter's hypervisor scenarios, which suggested coverage that the adapter does not have. Say what the step does instead: it builds the host-only configuration and runs any ignored tests that it declares, while the adapter's other tests run in the nvx-rs job. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 086f45a6-c39c-4704-85ef-1fa7f8b8b672 --- .github/workflows/run-nvx-microvm-tests.yml | 7 ++++--- doc/ci.md | 10 ++++++---- tests/runtime-adapter/README.md | 8 ++++---- 3 files changed, 14 insertions(+), 11 deletions(-) diff --git a/.github/workflows/run-nvx-microvm-tests.yml b/.github/workflows/run-nvx-microvm-tests.yml index 350f129d0..7bc311d93 100644 --- a/.github/workflows/run-nvx-microvm-tests.yml +++ b/.github/workflows/run-nvx-microvm-tests.yml @@ -1180,9 +1180,10 @@ jobs: --backend "${{ inputs.backend }}" --output-dir "build/test-results/aci-edge-sandboxes-${{ inputs.backend }}" - # The runtime adapter's scenarios on this backend: the ignored tests of its - # host-only configuration, which read the backend through - # `nvx::Hypervisor::from_env_or_default`. + # The runtime adapter's hypervisor scenarios on this backend: any ignored + # tests of its host-only configuration, which read the backend through + # `nvx::Hypervisor::from_env_or_default`. The step builds the adapter + # whether or not it declares any. - name: Run runtime adapter scenarios on Linux if: runner.os != 'Windows' && !inputs.debug-kernel shell: bash diff --git a/doc/ci.md b/doc/ci.md index d80f4176a..3d641e7f1 100644 --- a/doc/ci.md +++ b/doc/ci.md @@ -684,10 +684,12 @@ it keeps the OpenVMM log under `build/test-results/aci-edge-sandboxes-` uploaded with the other microVM logs. Changes under `aci_edge_sandboxes/` or `nvx-rs/` therefore trigger the NVX microVM test matrix. -The same jobs, except the debug-kernel ones, run the runtime adapter's -scenarios on their backend: the ignored tests of its host-only configuration, -with `NVX_HYPERVISOR` naming the backend. Changes under `tests/runtime-adapter/` -therefore trigger the NVX microVM test matrix too. +The same jobs, except the debug-kernel ones, build the runtime adapter's +host-only configuration on their backend and run any ignored tests that it +declares, its hypervisor scenarios, with `NVX_HYPERVISOR` naming the backend. +The adapter's other tests, including those of its guest agent, run in the +`nvx-rs` job. Changes under `tests/runtime-adapter/` therefore trigger the NVX +microVM test matrix too. ## Adversarial campaigns diff --git a/tests/runtime-adapter/README.md b/tests/runtime-adapter/README.md index a9a38acbd..f323d8c85 100644 --- a/tests/runtime-adapter/README.md +++ b/tests/runtime-adapter/README.md @@ -125,7 +125,7 @@ supported Rust version, on Linux and Windows, where it skips the guest-only configuration. A change to the adapter is a test input of CI's [change classification](../../doc/ci.md#change-classification), so it runs the NVX microVM tests. Their -[workflow](../../.github/workflows/run-nvx-microvm-tests.yml) runs the -adapter's hypervisor scenarios on each backend: its ignored tests in the -host-only configuration, with `NVX_HYPERVISOR` naming the backend, which -`nvx::Hypervisor::from_env_or_default` reads. +[workflow](../../.github/workflows/run-nvx-microvm-tests.yml) builds the +host-only configuration on each backend and runs any ignored tests that it +declares, the adapter's hypervisor scenarios, with `NVX_HYPERVISOR` naming the +backend, which `nvx::Hypervisor::from_env_or_default` reads. From 40fe8acd1b235864b996f59406bd65c2b2e4137f Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 20 Oct 2026 18:01:51 -0700 Subject: [PATCH 6/6] tests: size the adapter's longest message body in session records MAX_BODY was written as 256 KiB, and its doc said that four of the session's records carry less than that, though they carry exactly that much. Define it as four of the session's largest record payloads, the relation that the tests rely on: the longest message, with its header, spans five records, and the guest's minimum receive window holds a quarter of its body. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 086f45a6-c39c-4704-85ef-1fa7f8b8b672 --- tests/runtime-adapter/README.md | 2 +- tests/runtime-adapter/src/frame.rs | 7 +++++-- tests/runtime-adapter/tests/guest_session.rs | 2 +- 3 files changed, 7 insertions(+), 4 deletions(-) diff --git a/tests/runtime-adapter/README.md b/tests/runtime-adapter/README.md index f323d8c85..afab47e02 100644 --- a/tests/runtime-adapter/README.md +++ b/tests/runtime-adapter/README.md @@ -36,7 +36,7 @@ length, a little-endian `u32`, then the message's kind: | --- | --- | --- | | 1 | `Ping`: asks the guest to answer with `Pong` | none | | 2 | `Pong`: the guest's answer to `Ping` | none | -| 3 | `Echo`: bytes that the guest sends back in an `Echo` of its own | at most 256 KiB | +| 3 | `Echo`: bytes that the guest sends back in an `Echo` of its own | at most 256 KiB, what four of the session's largest records carry | | 4 | `Stop`: asks the guest to stop serving | none | | 5 | `Stopped`: the guest's confirmation of `Stop`, its last message | none | diff --git a/tests/runtime-adapter/src/frame.rs b/tests/runtime-adapter/src/frame.rs index a91b3a17e..3a4656d4a 100644 --- a/tests/runtime-adapter/src/frame.rs +++ b/tests/runtime-adapter/src/frame.rs @@ -13,11 +13,14 @@ use std::fmt; +use nvx::protocol::OUTER_MAX_PAYLOAD; + /// Bytes of a message's header: the body's length, then the message's kind. pub const HEADER_LEN: usize = 5; -/// Most bytes of a message's body, which is more than four of the session's records carry. -pub const MAX_BODY: usize = 256 * 1024; +/// Most bytes of a message's body: as many as four of the session's largest records carry, so that +/// the longest message, with its header, spans five records. +pub const MAX_BODY: usize = 4 * OUTER_MAX_PAYLOAD; /// A message of the adapter's framing. #[derive(Clone, Debug, PartialEq, Eq)] diff --git a/tests/runtime-adapter/tests/guest_session.rs b/tests/runtime-adapter/tests/guest_session.rs index 3c9e429c1..283bba86b 100644 --- a/tests/runtime-adapter/tests/guest_session.rs +++ b/tests/runtime-adapter/tests/guest_session.rs @@ -287,7 +287,7 @@ fn decode(decoder: &mut Decoder, mut bytes: &[u8]) -> Vec { messages } -/// Returns an echo of the longest body, which spans five records. +/// Returns an echo of the longest body: with its header, the message spans five records. fn longest() -> Message { Message::Echo((0..=u8::MAX).cycle().take(MAX_BODY).collect()) }