tests: attach the runtime adapter's host program to the control session

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,

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
 

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:

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<HostConnectionError> for ClientError {
+    fn from(error: HostConnectionError) -> Self {
+        Self::Session(error)
+    }
+}
+
+impl From<FrameError> 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<u8>,
+    /// 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<dyn Transport>,
+        capability: &[u8; CAPABILITY_LEN],
+        deadline: Instant,
+    ) -> Result<Self, ClientError> {
+        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<Instant>,
+    ) -> 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<Instant>) -> Result<Message, ClientError> {
+        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<Message, ClientError> {
+        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()
+    }
+}