tests: run the runtime adapter's executions in VMs through its host program

tests/runtime-adapter/src/exec.rs

@@ -1,7 +1,9 @@
 //! The adapter's messages about executions: programs that the guest runs for the host, whose
 //! standard streams and outcome the execution engine of `nvx::exec` carries. The host's side of an
 //! execution is the engine's [`Peer`], and the guest's side its [`Runner`], which the guest agent's
-//! worker binds to the program through `nvx::guest::exec::binding`.
+//! worker binds to the program through `nvx::guest::exec::binding`. A [`HostExecution`] holds the
+//! peer with a sink of the program's output and a source of its input, as the host program's
+//! driver, `host::Executions`, and the in-process tests of the worker run it.
 //!
 //! Each message of the engine travels in a frame of its own kind. Its body starts with the
 //! execution's [`Key`]: the execution's ID, then its generation, each a `u64`:
@@ -71,6 +73,14 @@ pub const KEY_LEN: usize = 16;
 /// longest body holds after the key and the stream.
 pub const MAX_CHUNK: u32 = (MAX_BODY - KEY_LEN - 1) as u32;
 
+/// The most executions that the guest's worker runs at once, beyond which the host's driver
+/// starts none while those that it started have not finished.
+pub const MAX_EXECUTIONS: usize = 8;
+
+/// The most bytes of each output stream of an execution that a [`HostExecution`] keeps: it
+/// cancels an execution whose stream would exceed them, so that its host holds no more.
+pub const MAX_CAPTURE: usize = 16 * 1024 * 1024;
+
 /// The code of a cause that the framing does not know, which no side decodes.
 const UNKNOWN_CAUSE: u8 = 0xff;
 
@@ -575,6 +585,173 @@ impl<'a> GuestMessage<'a> {
     }
 }
 
+/// The host's side of an execution: its [`Peer`], with a sink that takes the program's output as
+/// it arrives, unless its caller stalls it, and a source that hands over the program's input, if
+/// the start says that the host sends it, then ends it.
+///
+/// The sink keeps at most [`MAX_CAPTURE`] bytes of each stream. Once more of a stream arrives, the
+/// host's side cancels the execution, and its peer discards the rest once the grace ends, so that
+/// the outcome marks the stream truncated, with the cause [`Cause::Cancelled`].
+///
+/// Its caller sends the execution's [`Start`], hands it each of the guest's messages about the
+/// execution with [`deliver`](Self::deliver), and sends the frames that [`serve`](Self::serve)
+/// returns, until the peer [finished](Self::outcome). The host program's driver,
+/// `host::Executions`, runs several at once over the session.
+#[derive(Debug)]
+pub struct HostExecution {
+    /// The execution's start.
+    start: Start,
+    /// The host's side of the execution.
+    peer: Peer,
+    /// Whether the sink takes nothing.
+    stalled: bool,
+    /// The bytes of each stream that the sink took.
+    output: [Vec<u8>; 2],
+    /// The input that the source hands over, if the start says that the host sends it.
+    input: Vec<u8>,
+    /// How many bytes of the input the source handed over.
+    sent: usize,
+    /// Whether the source ended the input.
+    ended: bool,
+    /// The peer's events, in order.
+    events: Vec<peer::Event>,
+    /// Whether a stream outgrew [`MAX_CAPTURE`], so that the host's side cancelled the execution.
+    overflowed: bool,
+}
+
+impl HostExecution {
+    /// Returns the host's side of the execution of `start`, created at `now`, from which the
+    /// timeout of its peer's `timers` runs, with a source of `input`.
+    pub fn new(start: Start, timers: Timers, input: Vec<u8>, now: Instant) -> Self {
+        let peer = Peer::new(start.key, start.limits, timers, now);
+        Self {
+            start,
+            peer,
+            stalled: false,
+            output: [Vec::new(), Vec::new()],
+            input,
+            sent: 0,
+            ended: false,
+            events: Vec::new(),
+            overflowed: false,
+        }
+    }
+
+    /// Returns the execution's start.
+    pub fn start(&self) -> &Start {
+        &self.start
+    }
+
+    /// Returns the host's side of the execution.
+    pub fn peer(&self) -> &Peer {
+        &self.peer
+    }
+
+    /// Returns the host's side of the execution, with which the caller ends it, or tells it that
+    /// the session was reset or lost.
+    pub fn peer_mut(&mut self) -> &mut Peer {
+        &mut self.peer
+    }
+
+    /// Makes the sink take nothing while `stalled` holds, so that the program's output waits in
+    /// the peer, which grants the guest no credit for it.
+    pub fn stall(&mut self, stalled: bool) {
+        self.stalled = stalled;
+    }
+
+    /// Returns the bytes of `stream` that the sink took, at most [`MAX_CAPTURE`].
+    pub fn output(&self, stream: Stream) -> &[u8] {
+        &self.output[position(stream)]
+    }
+
+    /// Returns the peer's events, in order.
+    pub fn events(&self) -> &[peer::Event] {
+        &self.events
+    }
+
+    /// Returns the execution's outcome, once the peer finished.
+    pub fn outcome(&self) -> Option<Outcome> {
+        self.events.iter().find_map(|event| match event {
+            peer::Event::Finished(outcome) => Some(*outcome),
+            _ => None,
+        })
+    }
+
+    /// Returns when the peer's next timer fires, for the caller to [serve](Self::serve) the
+    /// execution then, or `None` while no timer runs.
+    pub fn deadline(&self) -> Option<Instant> {
+        self.peer.deadline()
+    }
+
+    /// Hands the guest's `message` about the execution to the peer at `now`, and returns what the
+    /// peer returns.
+    pub fn deliver(
+        &mut self,
+        message: &GuestMessage<'_>,
+        now: Instant,
+    ) -> Result<Receipt, Violation> {
+        message.deliver(&mut self.peer, now)
+    }
+
+    /// Takes every step that the source, the sink, and the peer can take at `now`, and returns
+    /// the peer's messages for the guest, as frames, in order, the cancellation of an execution
+    /// that the peer gave up on in this step included.
+    pub fn serve(&mut self, now: Instant) -> Vec<Message> {
+        if self.start.input && !self.ended {
+            self.sent += self.peer.input(&self.input[self.sent..]);
+            if self.sent == self.input.len() {
+                self.peer.end_input();
+                self.ended = true;
+            }
+        }
+        if !self.stalled {
+            for stream in Stream::ALL {
+                let output = &mut self.output[position(stream)];
+                loop {
+                    let pending = self.peer.pending_output(stream);
+                    if pending.is_empty() {
+                        break;
+                    }
+                    let room = MAX_CAPTURE - output.len();
+                    if room == 0 {
+                        if !self.overflowed {
+                            self.overflowed = true;
+                            self.peer.cancel(now);
+                        }
+                        break;
+                    }
+                    let length = pending.len().min(room);
+                    output.extend_from_slice(&pending[..length]);
+                    self.peer.consumed(stream, length);
+                }
+            }
+        }
+        self.peer.tick(now);
+        // The messages go first: a peer without grace or settlement time gives up on an
+        // execution in the step that ends it, and its last event, once taken, discards the
+        // cancellation that waits to be sent.
+        let mut buffer = vec![0; self.start.limits.chunk()];
+        let mut messages = Vec::new();
+        while let Some(message) = self.peer.next_message(&mut buffer) {
+            if let Some(message) = HostMessage::from_peer(self.start.key, message, &buffer) {
+                messages.push(message.encode());
+            }
+        }
+        while let Some(event) = self.peer.next_event() {
+            self.events.push(event);
+        }
+        messages
+    }
+}
+
+/// Returns the position of `stream` among [`Stream::ALL`].
+fn position(stream: Stream) -> usize {
+    match stream {
+        Stream::Stdout => 0,
+        Stream::Stderr => 1,
+    }
+}
+
 /// Returns `key` as a name for problems: its ID, then its generation.
 pub fn name(key: Key) -> String {
     format!("{}.{}", key.id(), key.generation())

tests/runtime-adapter/src/exec/tests.rs

@@ -1,5 +1,7 @@
 //! The encodings of the execution messages: round trips, the layout of a start, the bodies that
-//! break the rules, and a peer and a runner that exchange their messages through the frames.
+//! break the rules, and a peer and a runner that exchange their messages through the frames; and
+//! the host's side of an execution, which sends the cancellation of one that it gives up on at
+//! once, and cancels one whose stream outgrows what it keeps.
 
 use std::time::{Duration, Instant};
 
@@ -8,7 +10,10 @@ use nvx::exec::{
     Timers, Window, peer, runner,
 };
 
-use super::{AccountMode, GuestMessage, HostMessage, Identity, KEY_LEN, MAX_CHUNK, Start, name};
+use super::{
+    AccountMode, GuestMessage, HostExecution, HostMessage, Identity, KEY_LEN, MAX_CAPTURE,
+    MAX_CHUNK, Start, name,
+};
 use crate::frame::{MAX_BODY, Message};
 
 /// The key of most executions in these tests.
@@ -655,3 +660,96 @@ fn a_peer_and_a_runner_exchange_their_messages_through_the_frames() {
         .collect::<Vec<_>>();
     assert_eq!(interleaved, input);
 }
+
+#[test]
+fn the_host_s_side_sends_the_cancellation_of_an_execution_that_it_gives_up_on_at_once() {
+    // Without grace or settlement time, the host's side gives up on an execution in the step that
+    // ends it, for its timeout or its caller's cancellation, and still sends the cancellation.
+    let now = Instant::now();
+    let immediate = Timers::new(Duration::ZERO, Duration::ZERO);
+    let timeout = Duration::from_secs(1);
+    for (timers, cause) in [
+        (immediate.timeout(timeout), Cause::TimedOut),
+        (immediate, Cause::Cancelled),
+    ] {
+        let start = Start::new(KEY, shared(), timers, ["/bin/program"]);
+        let mut execution = HostExecution::new(start, timers, Vec::new(), now);
+        // The credit that opens the output's pool.
+        assert_eq!(execution.serve(now).len(), 1, "{cause:?}");
+        assert!(execution.outcome().is_none(), "{cause:?}");
+        let later = now + timeout;
+        if cause == Cause::Cancelled {
+            execution.peer_mut().cancel(later);
+        }
+        assert_eq!(
+            execution.serve(later),
+            [HostMessage::Cancel(KEY, cause).encode()],
+            "{cause:?}"
+        );
+        let outcome = execution.outcome().expect("the host's side gave up");
+        assert_eq!((outcome.cause(), outcome.status()), (Some(cause), None));
+        assert!(!outcome.is_reusable(), "{cause:?}");
+        assert!(execution.serve(later).is_empty(), "{cause:?}");
+    }
+}
+
+#[test]
+fn the_host_s_side_keeps_at_most_max_capture_bytes_of_a_stream_and_cancels_the_rest() {
+    // A program that writes a chunk to its standard output at each step, without end: the host's
+    // side keeps the first MAX_CAPTURE bytes, then cancels the execution, and its peer, without
+    // grace, discards the rest at once. The runner takes and drops what the program writes once
+    // it discards the stream, so each step writes once.
+    let now = Instant::now();
+    let timers = Timers::new(Duration::ZERO, Duration::from_secs(5));
+    let start = Start::new(KEY, shared(), timers, ["/usr/bin/yes"]);
+    let mut host = HostExecution::new(start, timers, Vec::new(), now);
+    let mut runner = Runner::new(KEY, shared(), timers, now);
+    runner.launched(Ok(()));
+    let written = vec![b'y'; shared().chunk()];
+    let mut buffer = vec![0u8; shared().chunk()];
+    let mut running = true;
+    for _ in 0..2 * MAX_CAPTURE / written.len() {
+        if host.outcome().is_some() {
+            break;
+        }
+        for frame in host.serve(now) {
+            match HostMessage::decode(&frame).unwrap().unwrap() {
+                HostMessage::Credit(key, pool, units) => runner.credit(key, pool, units),
+                HostMessage::Cancel(key, cause) => runner.cancel(now, key, cause),
+                message => panic!("{message:?}"),
+            }
+            .unwrap();
+        }
+        if running {
+            runner.output(Stream::Stdout, &written);
+        }
+        while let Some(request) = runner.next_request() {
+            match request {
+                runner::Request::Terminate(_) => {
+                    running = false;
+                    for stream in Stream::ALL {
+                        runner.output_closed(stream);
+                    }
+                    runner.exited(now, Some(Status::Signaled(15)));
+                }
+                runner::Request::Settle => runner.settled(Settlement::Settled),
+                _ => {}
+            }
+        }
+        while let Some(message) = runner.next_message(&mut buffer) {
+            let frame = GuestMessage::from_runner(KEY, message, &buffer)
+                .unwrap()
+                .encode();
+            let decoded = GuestMessage::decode(&frame).unwrap().unwrap();
+            host.deliver(&decoded, now).unwrap();
+        }
+    }
+    let outcome = host.outcome().expect("the execution finished");
+    let kept = host.output(Stream::Stdout);
+    assert_eq!(kept.len(), MAX_CAPTURE);
+    assert!(kept.iter().all(|byte| *byte == b'y'));
+    assert!(host.output(Stream::Stderr).is_empty());
+    assert_eq!(outcome.cause(), Some(Cause::Cancelled));
+    assert!(outcome.is_truncated(Stream::Stdout));
+    assert!(!outcome.is_truncated(Stream::Stderr));
+}

tests/runtime-adapter/src/host.rs

@@ -1,7 +1,8 @@
 //! 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.
+//! transport, and the driver of the [executions](crate::exec) that the host program runs over it,
+//! [`Executions`].
 //!
 //! # The host's order
 //!
@@ -22,6 +23,7 @@
 //! waits for the VM to end, and ends a VM whose guest did not stop, as it ends one after any other
 //! failure of the order, an [`OrderError`].
 
+mod executions;
 mod map_reply;
 mod order_error;
 
@@ -37,6 +39,7 @@ use crate::frame::{BodyError, Decoder, FrameError, Message};
 use crate::probe::{Facts, Probe, ProbeReply};
 use crate::root::{MapFailure, Ready, RootConfig, RootFailure, RootReply, encode_maps};
 
+pub use executions::{Executions, ExecutionsError, MAX_OUTPUT_FRAMES};
 pub use map_reply::MapReply;
 pub use order_error::OrderError;
 
@@ -117,7 +120,7 @@ impl From<FrameError> for ClientError {
 /// returns the guest agent's next message, whether it spans records or shares one with the next
 /// message. [`ping`](Self::ping), [`echo`](Self::echo), [`configure_root`](Self::configure_root),
 /// [`map`](Self::map), [`start`](Self::start), [`probe`](Self::probe), and [`stop`](Self::stop)
-/// each send a request and wait for its answer.
+/// each send a request and wait for its answer, and [`Executions`] runs executions over the client.
 ///
 /// 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
@@ -208,7 +211,11 @@ impl Client {
                         | Message::RootConfig(_)
                         | Message::Probe(_)
                         | Message::Maps(_)
-                        | Message::Start),
+                        | Message::Start
+                        | Message::ExecStart(_)
+                        | Message::ExecInput(_)
+                        | Message::ExecInputEnd(_)
+                        | Message::ExecCancel(_)),
                     ) => {
                         return Err(self.fail(ClientError::Unexpected(message.kind())));
                     }