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); +}