From 02decf165f3df5441028fadc650ae7de7ad004fa Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 20 Oct 2026 19:59:04 -0700 Subject: [PATCH] tests: attach the runtime adapter's host program to the control session The runtime adapter's guest agent serves the control session between the host and the guest with the adapter's own framing, but its host program had no end of that session. Add `runtime_adapter::host::Client`, built on `nvx::host::nvxs::HostConnection` over any platform transport: - `Client::attach` attaches with the launch capability and reports the identity of the session's epoch; - `send` and `receive` exchange the adapter's messages as application data, reassembling messages that span records or share one; - `ping`, `echo` and `stop` send a request and check its answer. The client holds at most the rest of one record and one partial message. A reset of the session ends the attachment with its partial message, because the next epoch starts with a new message, in a new attachment. A failure of the session, a send that fails or times out, which the session cannot resume, a malformed message, or a message that only the host sends ends the client. A receive that times out keeps what arrived. A request ends the client on any failure, since a late answer would answer the wrong request. `tests/host_session.rs` checks it in the host-only configuration against a test broker that plays the broker's host-facing end, and the guest agent behind it, with the public record API over an in-memory transport, so it runs on Linux and Windows: the attachment and its bootstrap record, waits and refusals, messages that span records both ways, partial records, record lengths, ERROR payloads, sequence and identity errors, resets and disconnects in the middle of a message, timeouts, sends that fail, the answers that requests check, and the bounded buffers of the framing and the records. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76c7c1bb-20fe-4724-bbc0-dfb01bbec828 --- doc/ci.md | 12 +- doc/project-structure.md | 7 +- tests/runtime-adapter/README.md | 63 +- tests/runtime-adapter/src/host.rs | 266 +++++ tests/runtime-adapter/src/lib.rs | 11 +- tests/runtime-adapter/tests/host_session.rs | 1176 +++++++++++++++++++ 6 files changed, 1519 insertions(+), 16 deletions(-) create mode 100644 tests/runtime-adapter/src/host.rs create mode 100644 tests/runtime-adapter/tests/host_session.rs diff --git a/doc/ci.md b/doc/ci.md index 9da70b499..76853280d 100644 --- a/doc/ci.md +++ b/doc/ci.md @@ -699,11 +699,13 @@ build, the tests, and a check with the minimum Rust version of `nvx`, after 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. +guest binary is a static executable. They also check the adapter's own framing +over the control session between the host and the guest: in the host-only +configuration, on Linux and Windows, that the adapter's host program attaches +to the session and exchanges the framing with the guest agent, and in the +guest-only configuration, that its guest agent serves the session. Each runs +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, diff --git a/doc/project-structure.md b/doc/project-structure.md index f22c3914f..c079a04ce 100644 --- a/doc/project-structure.md +++ b/doc/project-structure.md @@ -335,9 +335,10 @@ 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 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). +independent adapter's does. Its host program and its guest agent exchange the +adapter's own framing over the control session between the host and the +guest, and its tests check each of them 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 afab47e02..838b34ec1 100644 --- a/tests/runtime-adapter/README.md +++ b/tests/runtime-adapter/README.md @@ -11,9 +11,9 @@ an adapter outside the repository does: - with the minimum supported Rust version of `nvx`. 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. +the host and the guest, in place of NVX's `NVXC` frames. Its host program sends +requests in that framing, and its guest agent serves them. The host program's +binary reports the contract that it was built against. ## Configurations @@ -22,7 +22,7 @@ Each configuration enables one side of `nvx`: | Configuration | Cargo options | `nvx` features | Modules | Binary | | --- | --- | --- | --- | --- | | Protocol-only | none | none | `frame` | none | -| Host-only | `--features host` | `host` | `frame` | `runtime-adapter-host` | +| Host-only | `--features host` | `host` | `frame`, `host` | `runtime-adapter-host` | | Guest-only | `--features guest --target x86_64-unknown-linux-musl` | `guest` | `frame`, `guest` | `runtime-adapter-guest`, a static executable | ## Framing @@ -69,6 +69,33 @@ 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. +## Host program + +`host::Client` is the host program's client of the guest agent. It runs over +the host's end of the control session, through `nvx::host::nvxs`, on any +transport to the session's host endpoint, such as the one that +`nvx::host::session::connect` opens to a VM's control endpoint: + +- `Client::attach` attaches with the VM's launch capability, waits while the + broker waits for the guest, and reports the identity of the session's epoch + that the broker attached it to; +- `send` sends a message in as many of the session's records as it takes, and + `receive` returns the guest agent's next message, whether it spans records + or shares one with the next message; +- `ping`, `echo` and `stop` each send a request and check its answer: `Pong`, + an `Echo` of the same bytes, and `Stopped`. + +Besides the session's own buffers, the client holds at most the rest of one of +the guest's records and one partial message. A reset of the session ends the +attachment, and its partial message with it, because the next epoch starts +with a new message, in a new attachment. + +A failure of the session, a send that fails or times out, a malformed message, +and a message that only the host program sends end the client, and its later +calls fail. A receive that times out doesn't: it keeps what arrived. A request +ends the client on any failure after the framing accepted it, a timeout +included, because an answer that arrived later would answer the wrong request. + ## Tests `tests/independent_build.rs` checks that every configuration builds as an @@ -102,6 +129,34 @@ 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. +`tests/host_session.rs` checks, in the host-only configuration, that the host +program's client attaches to the control session and exchanges the adapter's +own framing with the guest agent. A test broker plays the broker's host-facing +end of the session, and the guest agent behind it, with the public record API +of `nvx`, over an in-memory transport, so that the tests run on Linux and +Windows. They cover: + +- the attachment: its bootstrap record with the launch capability, waits for + the guest, a deadline that passes meanwhile, and refusals, with an ERROR or + by the end of the stream; +- messages that span records both ways, records that end a message and start + the next, and records whose bytes arrive in pieces down to one byte; +- a receive that times out inside a message, then returns the whole message; +- a send that times out after it wrote part of a record, and a send whose + transport fails, each of which ends the client with the message that it + held; +- resets and disconnects in the middle of a message, with the next epoch of + the instance and with a new instance, and streams that end inside a record; +- the records that the client refuses: sequence, epoch, and instance errors, + stale and identity-less resets, record lengths, bootstrap identity, ERROR + payloads, a payload beyond the session's limit, which fails at its header, + and header faults; +- the messages that it refuses: malformed headers, which fail before their + body arrives, and messages that only the host program sends; and a message + beyond the framing's limit, which it refuses to send; +- the answers that its requests check, and a request whose answer arrives too + late. + 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/host.rs b/tests/runtime-adapter/src/host.rs new file mode 100644 index 000000000..e901d8750 --- /dev/null +++ b/tests/runtime-adapter/src/host.rs @@ -0,0 +1,266 @@ +//! The adapter's host program: a client of the guest agent that attaches to the host's end of the +//! control session between the host and the guest, and exchanges the adapter's own +//! [framing](crate::frame) with the guest agent, through `nvx::host::nvxs` alone, over any platform +//! transport. + +use std::fmt; +use std::time::Instant; + +use nvx::host::nvxs::{HostConnection, HostConnectionError}; +use nvx::host::platform::Transport; +use nvx::protocol::CAPABILITY_LEN; +use nvx::protocol::nvxs::Identity; + +use crate::frame::{Decoder, FrameError, Message}; + +/// A failure of a [`Client`]. +#[derive(Debug)] +#[non_exhaustive] +pub enum ClientError { + /// The control session failed, or its deadline passed. + Session(HostConnectionError), + /// A malformed message: one that the guest agent sent, or one that the host program asked to + /// send, which the client did not send. + Frame(FrameError), + /// The guest agent sent a message of a kind that only the host program sends. + Unexpected(u8), + /// The guest agent answered a request with a message of another kind than the request's + /// answer. + Reply { + /// Kind of the request. + request: u8, + /// Kind of the message that answered it. + reply: u8, + }, + /// The guest agent answered an `Echo` with other bytes than the request's. + Echo, + /// An earlier failure ended the client. + Ended, +} + +impl fmt::Display for ClientError { + 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 guest sent a message of kind {kind}, which only the host sends" + ), + Self::Reply { request, reply } => write!( + formatter, + "the guest answered a message of kind {request} with one of kind {reply}" + ), + Self::Echo => formatter.write_str("the guest echoed other bytes than the host sent"), + Self::Ended => formatter.write_str("an earlier failure ended the client"), + } + } +} + +impl std::error::Error for ClientError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Session(error) => Some(error), + Self::Frame(error) => Some(error), + Self::Unexpected(_) | Self::Reply { .. } | Self::Echo | Self::Ended => None, + } + } +} + +impl From for ClientError { + fn from(error: HostConnectionError) -> Self { + Self::Session(error) + } +} + +impl From for ClientError { + fn from(error: FrameError) -> Self { + Self::Frame(error) + } +} + +/// The host program's client of the guest agent, over one attachment to the control session. +/// +/// [`attach`](Self::attach) attaches with the launch capability. [`send`](Self::send) sends a +/// message in as many of the session's records as it takes, and [`receive`](Self::receive) +/// returns the guest agent's next message, whether it spans records or shares one with the next +/// message. [`ping`](Self::ping), [`echo`](Self::echo), and [`stop`](Self::stop) each send a +/// request and wait for its answer. +/// +/// Besides the session's own buffers, the client holds at most the rest of one of the guest's +/// records and one partial message, whose header [`Decoder`] checks before it takes any byte of +/// the body. +/// +/// A reset of the session ends the attachment, and the partial message with it: the session's next +/// epoch starts with a new message, in a new attachment. `nvx::host::nvxs` reports a reset that +/// the broker announces as [`HostConnectionError::Reset`], and the end of the stream, with which +/// OpenVMM's broker ends an attachment instead, as [`HostConnectionError::Closed`]. +/// +/// A failure ends the client, and every later call fails with [`ClientError::Ended`]: a failure of +/// the session, a send that fails or times out, because the session cannot resume a message that +/// it took in part, a malformed message, and a message that only the host program sends. A +/// [`receive`](Self::receive) that times out doesn't: it keeps what arrived. A request ends the +/// client on any failure after the framing accepted it, a timeout included, because an answer that +/// arrived later would answer the wrong request. +pub struct Client { + /// The host's end of the control session. + connection: HostConnection, + /// Decoder of the guest agent's messages. + decoder: Decoder, + /// Payload of the guest's last DATA record. + payload: Vec, + /// Bytes of `payload` that the decoder took. + taken: usize, + /// Whether a failure ended the client. + ended: bool, +} + +impl Client { + /// Attaches to the control session over `transport`, a stream to its host endpoint, with the + /// launch `capability`, and waits until `deadline` while the broker waits for the guest. + pub fn attach( + transport: Box, + capability: &[u8; CAPABILITY_LEN], + deadline: Instant, + ) -> Result { + let connection = HostConnection::attach(transport, capability, deadline)?; + Ok(Self { + connection, + decoder: Decoder::new(), + payload: Vec::new(), + taken: 0, + ended: false, + }) + } + + /// Returns the identity of the session's epoch that the broker attached the client to. + pub fn identity(&self) -> Identity { + self.connection.identity() + } + + /// Sends `message` to the guest agent by `deadline`, or without a time limit, in as many of + /// the session's records as it takes. + /// + /// A message that the framing refuses fails before anything is sent, and the client stays + /// usable. Any other failure, a timeout included, ends the client, because the session cannot + /// resume a message that it took in part. + pub fn send( + &mut self, + message: &Message, + deadline: Option, + ) -> Result<(), ClientError> { + self.check()?; + let encoded = message.encode()?; + self.connection + .send(&encoded, deadline) + .map_err(|error| self.fail(error.into())) + } + + /// Returns the guest agent's next message, waiting until `deadline`, or without a time limit. + /// + /// A receive that times out keeps what arrived, so that a later one returns the whole message. + pub fn receive(&mut self, deadline: Option) -> Result { + self.check()?; + loop { + if self.taken < self.payload.len() { + let (consumed, message) = self + .decoder + .push(&self.payload[self.taken..]) + .map_err(|error| self.fail(error.into()))?; + self.taken += consumed; + match message { + None => {} + Some(message @ (Message::Ping | Message::Stop)) => { + return Err(self.fail(ClientError::Unexpected(message.kind()))); + } + Some(message) => return Ok(message), + } + } + // The decoder took the whole payload, so the next one replaces it. + self.payload = match self.connection.receive(deadline) { + Ok(payload) => payload, + Err(HostConnectionError::TimedOut) => { + return Err(HostConnectionError::TimedOut.into()); + } + Err(error) => return Err(self.fail(error.into())), + }; + self.taken = 0; + } + } + + /// Sends `Ping`, and waits until `deadline` for the guest agent's `Pong`. + pub fn ping(&mut self, deadline: Instant) -> Result<(), ClientError> { + let request = Message::Ping; + match self.request(&request, deadline)? { + Message::Pong => Ok(()), + reply => Err(self.mismatched(&request, &reply)), + } + } + + /// Sends `body` in an `Echo`, and waits until `deadline` for the guest agent's `Echo` of the + /// same bytes. + /// + /// A body of more than [`MAX_BODY`](crate::frame::MAX_BODY) bytes fails before anything is + /// sent, and the client stays usable. + pub fn echo(&mut self, body: &[u8], deadline: Instant) -> Result<(), ClientError> { + let request = Message::Echo(body.to_vec()); + match self.request(&request, deadline)? { + Message::Echo(echoed) if echoed == body => Ok(()), + Message::Echo(_) => Err(self.fail(ClientError::Echo)), + reply => Err(self.mismatched(&request, &reply)), + } + } + + /// Sends `Stop`, and waits until `deadline` for the guest agent's `Stopped`, its last message. + pub fn stop(&mut self, deadline: Instant) -> Result<(), ClientError> { + let request = Message::Stop; + match self.request(&request, deadline)? { + Message::Stopped => Ok(()), + reply => Err(self.mismatched(&request, &reply)), + } + } + + /// Sends `request` by `deadline`, and returns the guest agent's next message. Once the framing + /// accepted the request, any failure ends the client. + fn request(&mut self, request: &Message, deadline: Instant) -> Result { + self.send(request, Some(deadline))?; + self.receive(Some(deadline)) + .map_err(|error| self.fail(error)) + } + + /// Ends the client because the guest agent answered `request` with `reply`, of another kind. + fn mismatched(&mut self, request: &Message, reply: &Message) -> ClientError { + self.fail(ClientError::Reply { + request: request.kind(), + reply: reply.kind(), + }) + } + + /// Fails if an earlier failure ended the client. + fn check(&self) -> Result<(), ClientError> { + if self.ended { + Err(ClientError::Ended) + } else { + Ok(()) + } + } + + /// Ends the client with `error`, dropping the partial message, and returns it. + fn fail(&mut self, error: ClientError) -> ClientError { + self.ended = true; + self.decoder.clear(); + self.payload = Vec::new(); + self.taken = 0; + error + } +} + +impl fmt::Debug for Client { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("Client") + .field("identity", &self.identity()) + .field("ended", &self.ended) + .finish_non_exhaustive() + } +} diff --git a/tests/runtime-adapter/src/lib.rs b/tests/runtime-adapter/src/lib.rs index 46c9884e3..50b0b7edb 100644 --- a/tests/runtime-adapter/src/lib.rs +++ b/tests/runtime-adapter/src/lib.rs @@ -6,20 +6,23 @@ //! //! - 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 host program, -//! `runtime-adapter-host`; +//! - host-only, with the `host` feature: the host API too, and the host program, in the `host` +//! module and the `runtime-adapter-host` binary; //! - 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 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. +//! `runtime-adapter-guest serve` runs it on its standard input and output. The host program's +//! client attaches to the session with the launch capability and sends those requests. The +//! `runtime-adapter-host` binary reports the contract that it was built against. pub mod frame; #[cfg(all(feature = "guest", target_os = "linux"))] pub mod guest; +#[cfg(feature = "host")] +pub mod host; /// Command that starts the guest agent on the runtime adapter's standard streams. pub const GUEST_SERVE_COMMAND: &str = "serve"; diff --git a/tests/runtime-adapter/tests/host_session.rs b/tests/runtime-adapter/tests/host_session.rs new file mode 100644 index 000000000..a1cfe1716 --- /dev/null +++ b/tests/runtime-adapter/tests/host_session.rs @@ -0,0 +1,1176 @@ +//! The adapter's host program attaches to the control session between the host and the guest, and +//! exchanges the adapter's own framing with the guest agent, through public APIs only. +//! +//! A test broker plays the broker's host-facing end of the session, and the guest agent behind it, +//! with the public record API of `nvx`, over an in-memory transport, so that the tests run on Linux +//! and on Windows. The tests cover the attachment and its bootstrap identity, waits and refusals, +//! messages that span records and records that share messages, partial records, record lengths, +//! ERROR payloads, sequence and identity errors, resets and disconnects in the middle of a message, +//! timeouts, sends that fail, the answers that requests check, and the bounded buffer of the +//! adapter's decoder. +#![cfg(feature = "host")] + +use std::collections::VecDeque; +use std::io; +use std::sync::{Arc, Condvar, Mutex, MutexGuard}; +use std::thread::{self, JoinHandle}; +use std::time::{Duration, Instant}; + +use nvx::host::nvxs::HostConnectionError; +use nvx::host::platform::Transport; +use nvx::protocol::nvxs::{ + ErrorCode, Identity, Record, RecordError, RecordParser, RecordType, SessionError, encode_record, +}; +use nvx::protocol::{ + CAPABILITY_LEN, OUTER_HEADER_LEN, OUTER_MAX_PAYLOAD, OUTER_MIN_RECEIVE_WINDOW, +}; +use runtime_adapter::frame::{Decoder, FrameError, HEADER_LEN, MAX_BODY, Message}; +use runtime_adapter::host::{Client, ClientError}; + +/// Launch capability that the test broker accepts. +const CAPABILITY: [u8; CAPABILITY_LEN] = [0xa5; CAPABILITY_LEN]; + +/// Identity of the broker's first epoch. +const FIRST: Identity = Identity { + instance_id: [0x5c; 16], + epoch: 1, +}; + +/// Longest wait for a record from the host, for the client's requests, and for a thread to end. +const TIMEOUT: Duration = Duration::from_secs(30); + +/// Wait of a receive, a send, or an attachment that the test expects to time out. +const SHORT: Duration = Duration::from_millis(200); + +/// 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; + +/// One direction of an in-memory stream. +#[derive(Default)] +struct Pipe { + /// Bytes that the writer wrote and the reader has not read. + bytes: VecDeque, + /// Whether either end closed the direction. + closed: bool, +} + +/// One direction of an in-memory stream, which its writer and its reader share. Unless the writer +/// limits the bytes in flight, writes never wait, so one thread can play both ends of an exchange. +#[derive(Clone, Default)] +struct Direction(Arc<(Mutex, Condvar)>); + +impl Direction { + /// Appends `bytes` for the reader while at most `capacity` bytes are in flight, waiting at + /// most `timeout` for the reader to make room, unless either end closed the direction. + fn write( + &self, + mut bytes: &[u8], + capacity: usize, + timeout: Option, + ) -> io::Result<()> { + let (pipe, changed) = &*self.0; + let deadline = timeout.map(|timeout| Instant::now() + timeout); + let mut pipe = pipe.lock().unwrap(); + while !bytes.is_empty() { + if pipe.closed { + return Err(io::ErrorKind::BrokenPipe.into()); + } + let room = capacity.saturating_sub(pipe.bytes.len()).min(bytes.len()); + if room == 0 { + pipe = wait(changed, pipe, deadline)?; + continue; + } + pipe.bytes.extend(&bytes[..room]); + bytes = &bytes[room..]; + changed.notify_all(); + } + Ok(()) + } + + /// Reads into `buffer` once any bytes arrive, waiting at most `timeout`, or returns 0 once the + /// direction is closed and every byte was read. + fn read(&self, buffer: &mut [u8], timeout: Option) -> io::Result { + let (pipe, changed) = &*self.0; + let deadline = timeout.map(|timeout| Instant::now() + timeout); + let mut pipe = pipe.lock().unwrap(); + while pipe.bytes.is_empty() && !pipe.closed { + pipe = wait(changed, pipe, deadline)?; + } + let count = buffer.len().min(pipe.bytes.len()); + for (target, byte) in buffer.iter_mut().zip(pipe.bytes.drain(..count)) { + *target = byte; + } + // The writer may wait for room. + changed.notify_all(); + Ok(count) + } + + /// Closes the direction: the reader reads what is left, then the end of the stream, and + /// writes fail. + fn close(&self) { + let (pipe, changed) = &*self.0; + pipe.lock().unwrap().closed = true; + changed.notify_all(); + } +} + +/// Waits until `pipe` changes or `deadline` passes, and fails once it passed. +fn wait<'a>( + changed: &Condvar, + pipe: MutexGuard<'a, Pipe>, + deadline: Option, +) -> io::Result> { + let Some(deadline) = deadline else { + return Ok(changed.wait(pipe).unwrap()); + }; + let left = deadline.saturating_duration_since(Instant::now()); + if left.is_zero() { + return Err(io::ErrorKind::TimedOut.into()); + } + Ok(changed.wait_timeout(pipe, left).unwrap().0) +} + +/// The host's end of an in-memory stream to the broker. Each read returns at most `piece` bytes, +/// so that the broker's records arrive in pieces of that size, and at most `capacity` bytes of the +/// host's are in flight. +struct HostEnd { + /// Bytes from the broker. + input: Direction, + /// Bytes to the broker. + output: Direction, + /// Most bytes that one read returns. + piece: usize, + /// Most bytes to the broker in flight. + capacity: usize, +} + +impl Transport for HostEnd { + fn read(&mut self, buffer: &mut [u8], timeout: Option) -> io::Result { + let limit = buffer.len().min(self.piece); + self.input.read(&mut buffer[..limit], timeout) + } + + fn write_all(&mut self, data: &[u8], timeout: Option) -> io::Result<()> { + self.output.write(data, self.capacity, timeout) + } +} + +impl Drop for HostEnd { + // The host's stream ends with its transport, as a socket's or a pipe's does. + fn drop(&mut self) { + self.output.close(); + } +} + +/// The broker's host-facing end of the session, and the guest agent behind it. +/// +/// It takes the host's attachment, tells the host to wait for the guest, then attaches it, and +/// numbers its records to the host from 0. It checks the identity and sequence number of each of +/// the host's DATA records, and panics on any record that the protocol does not allow. Its guest +/// agent answers the host's messages as the adapter's guest agent does. +struct Broker { + /// Bytes to the host. + to_host: Direction, + /// Bytes from the host. + from_host: Direction, + /// Parser of the host's records. + parser: RecordParser, + /// Identity of the session's epoch. + identity: Identity, + /// Sequence number of the broker's next record to the host. + send_sequence: u64, + /// Sequence number of the host's next DATA record. + receive_sequence: u64, + /// Decoder of the host's messages. + decoder: Decoder, +} + +impl Broker { + /// Takes the host's attachment: a bootstrap record, which carries no identity, with the launch + /// capability. + fn accept(&mut self) { + let record = self.read().expect("the host attaches"); + assert_eq!(record.record_type(), RecordType::HostAttach); + let header = (record.instance_id(), record.epoch(), record.sequence()); + assert_eq!(header, ([0; 16], 0, 0)); + assert_eq!(record.payload(), CAPABILITY); + } + + /// Takes the host's attachment, tells the host to wait for the guest, then attaches it to the + /// session's identity. + fn attach(&mut self) { + self.accept(); + self.send(RecordType::Wait, &[]); + self.send(RecordType::Ready, &[]); + } + + /// Sends a record of the session's identity 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); + } + + /// Sends `bytes` in DATA records of at most `size` bytes. + fn send_data(&mut self, bytes: &[u8], size: usize) { + for chunk in bytes.chunks(size) { + self.send(RecordType::Data, chunk); + } + } + + /// Writes bytes to the host. + fn write(&self, bytes: &[u8]) { + self.to_host + .write(bytes, usize::MAX, None) + .expect("the broker's stream is open"); + } + + /// Ends the broker's stream to the host. + fn close(&self) { + self.to_host.close(); + } + + /// Returns the host's next record, or `None` once the host closed its stream at a record + /// boundary. + fn read(&mut self) -> Option { + let deadline = Instant::now() + TIMEOUT; + let mut buffer = [0; 4096]; + loop { + let wanted = self.parser.needed().min(buffer.len()); + let left = deadline.saturating_duration_since(Instant::now()); + let count = self + .from_host + .read(&mut buffer[..wanted], Some(left)) + .expect("the host sends a record or ends its stream in time"); + if count == 0 { + assert_eq!( + self.parser.buffered(), + 0, + "the host ended its stream inside a record" + ); + return None; + } + let (consumed, record) = self.parser.push(&buffer[..count]).unwrap(); + assert_eq!(consumed, count); + if record.is_some() { + return record; + } + } + } + + /// Takes the host's next record, a DATA record with the session's identity and the host's + /// next sequence number, and returns the messages that it completes. + fn take(&mut self) -> Vec { + let record = self.read().expect("the host sends DATA"); + assert_eq!(record.record_type(), RecordType::Data); + 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); + self.receive_sequence += 1; + decode(&mut self.decoder, record.payload()) + } + + /// Takes the host's next `count` messages. + fn messages(&mut self, count: usize) -> Vec { + let mut messages = Vec::new(); + while messages.len() < count { + messages.extend(self.take()); + } + assert_eq!(messages.len(), count, "{messages:?}"); + messages + } + + /// Answers the host's next message as the guest agent does, in DATA records of at most `size` + /// bytes, and returns the message. + fn answer(&mut self, size: usize) -> Message { + let [message] = <[Message; 1]>::try_from(self.messages(1)).unwrap(); + self.send_data(&answer(&message).encode().unwrap(), size); + message + } + + /// Serves the host as the guest agent does until it answered `Stop`: it answers the messages + /// that each of the host's records completes, together, in DATA records of at most `size` + /// bytes. Returns the host's messages. + fn serve(&mut self, size: usize) -> Vec { + let mut served = Vec::new(); + while !served.contains(&Message::Stop) { + let messages = self.take(); + let answers = messages.iter().map(answer).collect::>(); + self.send_data(&encode(&answers), size); + served.extend(messages); + } + served + } +} + +/// Connects a new broker of the session `identity` to a transport for the host, whose reads +/// return at most `piece` bytes. +fn connect(identity: Identity, piece: usize) -> (Broker, Box) { + connect_bounded(identity, piece, usize::MAX) +} + +/// Connects a new broker of the session `identity` to a transport for the host, whose reads +/// return at most `piece` bytes, and which holds at most `capacity` bytes of the host's in flight. +fn connect_bounded( + identity: Identity, + piece: usize, + capacity: usize, +) -> (Broker, Box) { + let (to_host, from_host) = (Direction::default(), Direction::default()); + let transport = HostEnd { + input: to_host.clone(), + output: from_host.clone(), + piece, + capacity, + }; + let broker = Broker { + to_host, + from_host, + parser: RecordParser::new(), + identity, + send_sequence: 0, + receive_sequence: 0, + decoder: Decoder::new(), + }; + (broker, Box::new(transport)) +} + +/// Attaches a client to a new broker of the session `identity`, over a transport whose reads +/// return at most `piece` bytes, and returns both. +fn attached(identity: Identity, piece: usize) -> (Broker, Client) { + attach_over(connect(identity, piece)) +} + +/// Attaches a client over the transport of a connection to its broker, and returns both. +fn attach_over((mut broker, transport): (Broker, Box)) -> (Broker, Client) { + let attaching = attach(transport, deadline()); + broker.attach(); + let client = joined(attaching).unwrap(); + assert_eq!(client.identity(), broker.identity); + (broker, client) +} + +/// Starts the client's attachment over `transport` in a thread, with `deadline`. +fn attach( + transport: Box, + deadline: Instant, +) -> JoinHandle> { + thread::spawn(move || Client::attach(transport, &CAPABILITY, deadline)) +} + +/// Runs `request` with `client` in a thread, and returns the thread, which returns the client with +/// the request's result. +fn spawn( + mut client: Client, + request: impl FnOnce(&mut Client) -> T + Send + 'static, +) -> JoinHandle<(Client, T)> { + thread::spawn(move || { + let result = request(&mut client); + (client, result) + }) +} + +/// Returns what `thread` returned, and fails the test if it does not end in time, instead of +/// hanging it. +fn joined(thread: JoinHandle) -> T { + let deadline = Instant::now() + TIMEOUT; + while !thread.is_finished() { + assert!(Instant::now() < deadline, "the client ends in time"); + thread::sleep(Duration::from_millis(5)); + } + thread.join().unwrap() +} + +/// Returns the deadline of an operation that the test expects to complete. +fn deadline() -> Instant { + Instant::now() + TIMEOUT +} + +/// 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 `bytes` with `patch` written at `offset`. +fn patched(mut bytes: Vec, offset: usize, patch: &[u8]) -> Vec { + bytes[offset..offset + patch.len()].copy_from_slice(patch); + bytes +} + +/// Returns a record of `record_type` whose header declares a payload of `length` bytes, followed +/// by that many bytes, which no valid record of the type carries. +fn with_length(record_type: RecordType, identity: Identity, sequence: u64, length: u32) -> Vec { + // Each type allows a payload of one of these lengths. + let valid = match record_type { + RecordType::Data | RecordType::Error => 4, + _ => 0, + }; + let payload = if record_type == RecordType::Error { + ErrorCode::Authentication.code().to_le_bytes().to_vec() + } else { + vec![0; valid] + }; + let mut bytes = patched( + record(record_type, identity, sequence, &payload), + LENGTH_OFFSET, + &length.to_le_bytes(), + ); + bytes.truncate(OUTER_HEADER_LEN); + bytes.resize(OUTER_HEADER_LEN + length as usize, 0); + bytes +} + +/// Returns the guest agent's answer to `message`. +fn answer(message: &Message) -> Message { + match message { + Message::Ping => Message::Pong, + Message::Echo(body) => Message::Echo(body.clone()), + Message::Stop => Message::Stopped, + message => panic!("the host sent {message:?}, which only the guest sends"), + } +} + +/// 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 the longest body: with its header, a message that carries it spans five records. +fn longest() -> Vec { + (0..=u8::MAX).cycle().take(MAX_BODY).collect() +} + +/// Checks that a failure ended `client`: every later call fails. +fn assert_ended(client: &mut Client) { + let received = client.receive(Some(deadline())); + assert!(matches!(received, Err(ClientError::Ended)), "{received:?}"); + let sent = client.send(&Message::Ping, Some(deadline())); + assert!(matches!(sent, Err(ClientError::Ended)), "{sent:?}"); + let pinged = client.ping(deadline()); + assert!(matches!(pinged, Err(ClientError::Ended)), "{pinged:?}"); +} + +#[test] +fn the_client_attaches_and_exchanges_messages_that_span_records() { + // Echoes in one record, of nothing, of the largest record, and of the longest body, which + // spans five records each way. + let bodies = [ + b"one record".to_vec(), + Vec::new(), + vec![0x42; OUTER_MAX_PAYLOAD], + longest(), + ]; + for size in [7, 4096, OUTER_MAX_PAYLOAD] { + let (mut broker, transport) = connect(FIRST, usize::MAX); + let serving = thread::spawn(move || { + broker.attach(); + let served = broker.serve(size); + // The host's stream ends with the client, at a record boundary. + assert!(broker.read().is_none()); + served + }); + let mut client = Client::attach(transport, &CAPABILITY, deadline()).unwrap(); + assert_eq!(client.identity(), FIRST); + client.ping(deadline()).unwrap(); + for body in &bodies { + client.echo(body, deadline()).unwrap(); + } + client.stop(deadline()).unwrap(); + drop(client); + + let mut expected = vec![Message::Ping]; + expected.extend(bodies.iter().cloned().map(Message::Echo)); + expected.push(Message::Stop); + assert_eq!(joined(serving), expected, "records of {size} bytes"); + } +} + +#[test] +fn the_client_reassembles_messages_from_partial_and_shared_records() { + // In records of 7 bytes, most records end a message and start the next. + let messages = [ + Message::Pong, + Message::Echo(b"two".to_vec()), + Message::Echo(Vec::new()), + Message::Echo(vec![0x42; 100_000]), + Message::Pong, + Message::Stopped, + ]; + let stream = encode(&messages); + let cases = [ + (1, 4096), + (5, 7), + (OUTER_HEADER_LEN, OUTER_MAX_PAYLOAD), + (4096, 7), + (usize::MAX, OUTER_MAX_PAYLOAD), + ]; + for (piece, size) in cases { + let (mut broker, mut client) = attached(FIRST, piece); + broker.send_data(&stream, size); + for expected in &messages { + let message = client.receive(Some(deadline())).unwrap(); + assert_eq!( + &message, expected, + "reads of {piece}, records of {size} bytes" + ); + } + // Nothing more arrives. + let timed_out = client.receive(Some(Instant::now() + SHORT)); + let nothing = matches!( + timed_out, + Err(ClientError::Session(HostConnectionError::TimedOut)) + ); + assert!(nothing, "{timed_out:?}"); + } +} + +#[test] +fn a_receive_that_times_out_keeps_the_partial_message() { + let (mut broker, mut client) = attached(FIRST, usize::MAX); + let echo = Message::Echo(vec![7; 1000]); + let encoded = echo.encode().unwrap(); + + // Half of the message, then the first bytes of the record with the rest. + broker.send(RecordType::Data, &encoded[..500]); + let rest = record( + RecordType::Data, + FIRST, + broker.send_sequence, + &encoded[500..], + ); + broker.send_sequence += 1; + broker.write(&rest[..20]); + let timed_out = client.receive(Some(Instant::now() + SHORT)); + let kept = matches!( + timed_out, + Err(ClientError::Session(HostConnectionError::TimedOut)) + ); + assert!(kept, "{timed_out:?}"); + + broker.write(&rest[20..]); + assert_eq!(client.receive(Some(deadline())).unwrap(), echo); + + // The client stays usable. + client.send(&Message::Ping, Some(deadline())).unwrap(); + assert_eq!(broker.answer(OUTER_MAX_PAYLOAD), Message::Ping); + assert_eq!(client.receive(Some(deadline())).unwrap(), Message::Pong); +} + +#[test] +fn a_send_that_fails_or_times_out_ends_the_client_with_what_it_holds() { + // One record carries two messages: the client returns the first, and holds the second. + let shared = encode(&[Message::Pong, Message::Echo(b"held".to_vec())]); + + // The broker stops reading, and the host's stream holds 128 bytes, so a send times out after + // it wrote part of a record, which the session cannot resume. + let (mut broker, mut client) = attach_over(connect_bounded(FIRST, usize::MAX, 128)); + broker.send(RecordType::Data, &shared); + assert_eq!(client.receive(Some(deadline())).unwrap(), Message::Pong); + let echo = Message::Echo(vec![0; 1000]); + let timed_out = client.send(&echo, Some(Instant::now() + SHORT)); + let late = matches!( + timed_out, + Err(ClientError::Session(HostConnectionError::TimedOut)) + ); + assert!(late, "{timed_out:?}"); + assert_ended(&mut client); + + // The transport fails, as when the broker went away. + let (mut broker, mut client) = attached(FIRST, usize::MAX); + broker.send(RecordType::Data, &shared); + assert_eq!(client.receive(Some(deadline())).unwrap(), Message::Pong); + broker.from_host.close(); + let failed = client.send(&Message::Ping, Some(deadline())); + let broken = matches!( + &failed, + Err(ClientError::Session(HostConnectionError::Io(error))) + if error.kind() == io::ErrorKind::BrokenPipe + ); + assert!(broken, "{failed:?}"); + assert_ended(&mut client); +} + +#[test] +fn a_reset_or_a_disconnect_in_the_middle_of_a_message_drops_it() { + let echo = Message::Echo(vec![1; 1000]).encode().unwrap(); + + // Half of a message, then the broker resets the session to the instance's next epoch, which + // ends the attachment. + let (mut broker, mut client) = attached(FIRST, usize::MAX); + broker.send(RecordType::Data, &echo[..500]); + let next = Identity { epoch: 2, ..FIRST }; + broker.identity = next; + broker.send(RecordType::Reset, &[]); + let reset = client.receive(Some(deadline())); + let ended = matches!( + reset, + Err(ClientError::Session(HostConnectionError::Reset(identity))) if identity == next + ); + assert!(ended, "{reset:?}"); + assert_ended(&mut client); + + // The new epoch starts with a new message, in a new attachment: its first bytes complete + // nothing of the earlier epoch's. + let (mut broker, mut client) = attached(next, usize::MAX); + client.send(&Message::Ping, Some(deadline())).unwrap(); + assert_eq!(broker.answer(OUTER_MAX_PAYLOAD), Message::Ping); + assert_eq!(client.receive(Some(deadline())).unwrap(), Message::Pong); + + // Part of a header, then the broker ends the stream at a record boundary, as OpenVMM's broker + // ends an attachment. + broker.send(RecordType::Data, &echo[..HEADER_LEN - 1]); + broker.close(); + let closed = client.receive(Some(deadline())); + let ended = matches!( + closed, + Err(ClientError::Session(HostConnectionError::Closed)) + ); + assert!(ended, "{closed:?}"); + assert_ended(&mut client); + + // A broker restored from a snapshot has a new instance, which starts at epoch 1. + let restored = Identity { + instance_id: [0x7e; 16], + epoch: 1, + }; + let (mut broker, client) = attached(restored, usize::MAX); + let pinging = spawn(client, |client| client.ping(deadline())); + assert_eq!(broker.answer(7), Message::Ping); + let (mut client, pinged) = joined(pinging); + pinged.unwrap(); + + // A stream that ends inside a record fails it. + let data = record(RecordType::Data, restored, broker.send_sequence, &echo); + broker.write(&data[..30]); + broker.close(); + let truncated = client.receive(Some(deadline())); + let ended = matches!( + truncated, + Err(ClientError::Session(HostConnectionError::Protocol( + SessionError::Truncated { buffered: 30 } + ))) + ); + assert!(ended, "{truncated:?}"); + assert_ended(&mut client); +} + +#[test] +fn the_client_waits_for_the_guest_and_reports_refusals_while_attaching() { + // The broker tells the host to wait until the guest acknowledges the session's reset, as often + // as it takes. + let (mut broker, transport) = connect(FIRST, usize::MAX); + let attaching = attach(transport, deadline()); + broker.accept(); + for _ in 0..3 { + broker.send(RecordType::Wait, &[]); + thread::sleep(Duration::from_millis(20)); + } + assert!(!attaching.is_finished(), "the client waits for READY"); + broker.send(RecordType::Ready, &[]); + assert_eq!(joined(attaching).unwrap().identity(), FIRST); + + // The deadline passes while the broker waits for the guest. The broker writes its WAIT ahead, + // so that the case doesn't depend on when the client's thread starts. + let (mut broker, transport) = connect(FIRST, usize::MAX); + broker.send(RecordType::Wait, &[]); + let timed_out = joined(attach(transport, Instant::now() + SHORT)); + let late = matches!( + timed_out, + Err(ClientError::Session(HostConnectionError::TimedOut)) + ); + assert!(late, "{timed_out:?}"); + + // The broker refuses the capability with an ERROR, at once or after a WAIT. + for waits in [0, 1] { + let (mut broker, transport) = connect(FIRST, usize::MAX); + let attaching = attach(transport, deadline()); + broker.accept(); + for _ in 0..waits { + broker.send(RecordType::Wait, &[]); + } + let code = ErrorCode::Authentication.code().to_le_bytes(); + broker.send(RecordType::Error, &code); + let refused = joined(attaching); + let rejected = matches!( + refused, + Err(ClientError::Session(HostConnectionError::Rejected( + ErrorCode::Authentication + ))) + ); + assert!(rejected, "after {waits} WAIT records: {refused:?}"); + } + + // OpenVMM's broker refuses it by ending the stream. + let (mut broker, transport) = connect(FIRST, usize::MAX); + let attaching = attach(transport, deadline()); + broker.accept(); + broker.close(); + let refused = joined(attaching); + let closed = matches!( + refused, + Err(ClientError::Session(HostConnectionError::Closed)) + ); + assert!(closed, "{refused:?}"); +} + +#[test] +fn the_client_fails_on_records_that_the_protocol_does_not_allow() { + let other_instance = Identity { + instance_id: [0x7e; 16], + ..FIRST + }; + let next_epoch = Identity { epoch: 2, ..FIRST }; + let no_identity = Identity { + instance_id: [0; 16], + epoch: 0, + }; + let code = |code: u32| code.to_le_bytes(); + let authentication = code(ErrorCode::Authentication.code()); + let wait = record(RecordType::Wait, FIRST, 0, &[]); + + // Records that the broker sends while it attaches the host, at its sequence 0. + let attaching = [ + ( + "a WAIT that skips a sequence number", + record(RecordType::Wait, FIRST, 1, &[]), + SessionError::Sequence { + expected: 0, + actual: 1, + }, + ), + ( + "a READY that repeats a sequence number", + [wait.clone(), record(RecordType::Ready, FIRST, 0, &[])].concat(), + SessionError::Sequence { + expected: 1, + actual: 0, + }, + ), + ( + "a READY without an instance", + record( + RecordType::Ready, + Identity { + epoch: 1, + ..no_identity + }, + 0, + &[], + ), + SessionError::MissingIdentity(RecordType::Ready), + ), + ( + "a READY without an epoch", + record(RecordType::Ready, Identity { epoch: 0, ..FIRST }, 0, &[]), + SessionError::MissingIdentity(RecordType::Ready), + ), + ( + "a WAIT with a payload", + with_length(RecordType::Wait, FIRST, 0, 4), + SessionError::Record(RecordError::Length { + record_type: RecordType::Wait, + length: 4, + }), + ), + ( + "a READY with a payload", + with_length(RecordType::Ready, FIRST, 0, 1), + SessionError::Record(RecordError::Length { + record_type: RecordType::Ready, + length: 1, + }), + ), + ( + "an ERROR without a code", + with_length(RecordType::Error, FIRST, 0, 0), + SessionError::Record(RecordError::Length { + record_type: RecordType::Error, + length: 0, + }), + ), + ( + "an ERROR with an unknown code", + patched( + record(RecordType::Error, FIRST, 0, &authentication), + OUTER_HEADER_LEN, + &code(2), + ), + SessionError::Record(RecordError::ErrorCode(2)), + ), + ( + "DATA before READY", + record(RecordType::Data, FIRST, 0, b"x"), + SessionError::Unexpected(RecordType::Data), + ), + ( + "a RESET before READY", + record(RecordType::Reset, FIRST, 0, &[]), + SessionError::Unexpected(RecordType::Reset), + ), + ( + "a guest's attachment", + record(RecordType::GuestAttach, no_identity, 0, &[]), + SessionError::Unexpected(RecordType::GuestAttach), + ), + ( + "a bootstrap record with an identity", + patched( + record(RecordType::HostAttach, no_identity, 0, &CAPABILITY), + INSTANCE_OFFSET, + &[1], + ), + SessionError::Record(RecordError::BootstrapIdentity(RecordType::HostAttach)), + ), + ( + "an ACK", + record( + RecordType::Ack, + FIRST, + 0, + &OUTER_MIN_RECEIVE_WINDOW.to_le_bytes(), + ), + SessionError::Unexpected(RecordType::Ack), + ), + ( + "a CREDIT", + record(RecordType::Credit, FIRST, 0, &code(1)), + SessionError::Unexpected(RecordType::Credit), + ), + ( + "an unknown record type", + patched(wait.clone(), TYPE_OFFSET, &[0x7f]), + SessionError::Record(RecordError::Type(0x7f)), + ), + ( + "nonzero flags", + patched(wait.clone(), FLAGS_OFFSET, &[1]), + SessionError::Record(RecordError::Flags(1)), + ), + ( + "another protocol version", + patched(wait.clone(), VERSION_OFFSET, &2u16.to_le_bytes()), + SessionError::Record(RecordError::Version(2)), + ), + ( + "a wrong magic", + patched(wait, 0, b"NVXT"), + SessionError::Record(RecordError::Magic(*b"NVXT")), + ), + ]; + for (name, bytes, expected) in attaching { + let (mut broker, transport) = connect(FIRST, usize::MAX); + let attaching = attach(transport, deadline()); + broker.accept(); + broker.write(&bytes); + let refused = joined(attaching); + let failed = matches!( + &refused, + Err(ClientError::Session(HostConnectionError::Protocol(error))) if *error == expected + ); + assert!(failed, "{name}: {refused:?}"); + } + + // Records that the broker sends once it attached the host, at its sequence 2, after WAIT and + // READY. + let ping = Message::Ping.encode().unwrap(); + let data = |identity, sequence| record(RecordType::Data, identity, sequence, &ping); + let attached_cases = [ + ( + "DATA that skips a sequence number", + data(FIRST, 3), + SessionError::Sequence { + expected: 2, + actual: 3, + }, + ), + ( + "DATA that repeats a sequence number", + data(FIRST, 1), + SessionError::Sequence { + expected: 2, + actual: 1, + }, + ), + ( + "DATA of another epoch", + data(next_epoch, 2), + SessionError::Epoch { + expected: 1, + actual: 2, + }, + ), + ( + "DATA of another instance", + data(other_instance, 2), + SessionError::Instance { + expected: FIRST.instance_id, + actual: other_instance.instance_id, + }, + ), + ( + "an empty DATA record", + with_length(RecordType::Data, FIRST, 2, 0), + SessionError::Record(RecordError::Length { + record_type: RecordType::Data, + length: 0, + }), + ), + ( + // Refused at its header, before its payload arrives: the record parser never buffers a + // payload beyond the limit. + "a DATA header beyond the largest payload", + with_length(RecordType::Data, FIRST, 2, OUTER_MAX_PAYLOAD as u32 + 1) + [..OUTER_HEADER_LEN] + .to_vec(), + SessionError::Record(RecordError::Length { + record_type: RecordType::Data, + length: OUTER_MAX_PAYLOAD + 1, + }), + ), + ( + "a RESET that repeats the epoch", + record(RecordType::Reset, FIRST, 2, &[]), + SessionError::StaleEpoch { + current: 1, + actual: 1, + }, + ), + ( + "a RESET of another instance", + record( + RecordType::Reset, + Identity { + epoch: 2, + ..other_instance + }, + 2, + &[], + ), + SessionError::Instance { + expected: FIRST.instance_id, + actual: other_instance.instance_id, + }, + ), + ( + "a RESET without an identity", + record(RecordType::Reset, no_identity, 2, &[]), + SessionError::MissingIdentity(RecordType::Reset), + ), + ( + "a RESET with a payload", + with_length(RecordType::Reset, next_epoch, 2, 4), + SessionError::Record(RecordError::Length { + record_type: RecordType::Reset, + length: 4, + }), + ), + ( + "a WAIT after READY", + record(RecordType::Wait, FIRST, 2, &[]), + SessionError::Unexpected(RecordType::Wait), + ), + ( + "a second READY", + record(RecordType::Ready, FIRST, 2, &[]), + SessionError::Unexpected(RecordType::Ready), + ), + ( + "an ERROR after READY", + record(RecordType::Error, FIRST, 2, &authentication), + SessionError::Unexpected(RecordType::Error), + ), + ]; + for (name, bytes, expected) in attached_cases { + let (broker, mut client) = attached(FIRST, usize::MAX); + broker.write(&bytes); + let refused = client.receive(Some(deadline())); + let failed = matches!( + &refused, + Err(ClientError::Session(HostConnectionError::Protocol(error))) if *error == expected + ); + assert!(failed, "{name}: {refused:?}"); + assert_ended(&mut client); + } +} + +#[test] +fn the_client_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| [&length[..], &[kind]].concat(); + 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([0xff; 4], 3), + FrameError::Length { + kind: 3, + length: u32::MAX as usize, + }, + ), + ( + header(1u32.to_le_bytes(), 2), + FrameError::Length { kind: 2, length: 1 }, + ), + ]; + for (bytes, expected) in cases { + let (mut broker, mut client) = attached(FIRST, usize::MAX); + broker.send(RecordType::Data, &bytes); + let refused = client.receive(Some(deadline())); + let failed = matches!(&refused, Err(ClientError::Frame(error)) if *error == expected); + assert!(failed, "{bytes:?}: {refused:?}"); + assert_ended(&mut client); + } + + // The client takes the guest agent's messages, and refuses those that only the host program + // sends. + for message in [Message::Ping, Message::Stop] { + let (mut broker, mut client) = attached(FIRST, usize::MAX); + broker.send(RecordType::Data, &message.encode().unwrap()); + let refused = client.receive(Some(deadline())); + let failed = matches!( + &refused, + Err(ClientError::Unexpected(kind)) if *kind == message.kind() + ); + assert!(failed, "{message:?}: {refused:?}"); + assert_ended(&mut client); + } + + // The client refuses to send a message beyond the limit, sends nothing, and stays usable. + let (mut broker, mut client) = attached(FIRST, usize::MAX); + let too_long = vec![0; MAX_BODY + 1]; + let refused = client.send(&Message::Echo(too_long.clone()), Some(deadline())); + let expected = FrameError::Length { + kind: 3, + length: MAX_BODY + 1, + }; + assert!( + matches!(&refused, Err(ClientError::Frame(error)) if *error == expected), + "{refused:?}" + ); + let refused = client.echo(&too_long, deadline()); + assert!( + matches!(&refused, Err(ClientError::Frame(error)) if *error == expected), + "{refused:?}" + ); + let pinging = spawn(client, |client| client.ping(deadline())); + // The host's first DATA record carries the Ping. + assert_eq!(broker.answer(OUTER_MAX_PAYLOAD), Message::Ping); + joined(pinging).1.unwrap(); +} + +#[test] +fn requests_check_their_answers_and_end_the_client_when_they_fail() { + /// A request of the client. + type Request = fn(&mut Client) -> Result<(), ClientError>; + /// The kinds of a request and of its answer that a failure reports, or `None` for an echo of + /// other bytes. + type Kinds = Option<(u8, u8)>; + // Each request, and an answer that doesn't answer it. + let cases: [(&str, Request, Message, Kinds); 5] = [ + ( + "Ping answered with Stopped", + |client| client.ping(deadline()), + Message::Stopped, + Some((1, 5)), + ), + ( + "Echo answered with Pong", + |client| client.echo(b"abc", deadline()), + Message::Pong, + Some((3, 2)), + ), + ( + "Echo answered with other bytes", + |client| client.echo(b"abc", deadline()), + Message::Echo(b"abd".to_vec()), + None, + ), + ( + "Stop answered with Pong", + |client| client.stop(deadline()), + Message::Pong, + Some((4, 2)), + ), + ( + "Stop answered with an Echo", + |client| client.stop(deadline()), + Message::Echo(Vec::new()), + Some((4, 3)), + ), + ]; + for (name, request, reply, kinds) in cases { + let (mut broker, client) = attached(FIRST, usize::MAX); + let requesting = spawn(client, request); + broker.messages(1); + broker.send_data(&reply.encode().unwrap(), OUTER_MAX_PAYLOAD); + let (mut client, result) = joined(requesting); + let error = result.unwrap_err(); + let failed = match (&error, kinds) { + (ClientError::Reply { request, reply }, Some(kinds)) => (*request, *reply) == kinds, + (ClientError::Echo, None) => true, + _ => false, + }; + assert!(failed, "{name}: {error:?}"); + assert_ended(&mut client); + } + + // A request whose answer does not arrive in time ends the client, so that a late answer never + // answers a later request. + let (mut broker, client) = attached(FIRST, usize::MAX); + let requesting = spawn(client, |client| client.ping(Instant::now() + SHORT)); + assert_eq!(broker.messages(1), [Message::Ping]); + let (mut client, result) = joined(requesting); + let late = matches!( + result, + Err(ClientError::Session(HostConnectionError::TimedOut)) + ); + assert!(late, "{result:?}"); + broker.send_data(&Message::Pong.encode().unwrap(), OUTER_MAX_PAYLOAD); + assert_ended(&mut client); +}