From d354674ab3bdd251ff40b39ffd629c135c1e882b Mon Sep 17 00:00:00 2001 From: Pedro Henrique Penna Date: Thu, 12 Nov 2026 01:45:22 -0800 Subject: [PATCH 1/3] guest: poll descriptors for writing as well as reading The managed agent's loop waits on its control console and on an execution's pipes with readable_or_hung_up, which polls each descriptor for input or hangup. Serving NVXC version 2 (SL4.4b, #1128) needs that wait to cover descriptors that the agent writes too: the console while records wait in its send queue, and an execution's standard input while bytes wait to enter its pipe. sys::ready polls each descriptor for input or hangup, or, where its flag is set, for space to write, and returns whether each one woke, in the caller's order. readable_or_hung_up is ready with every flag clear, so its callers keep their behavior. Part of #1128 (SL4.4b). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- nvx-rs/src/guest/sys/poll.rs | 22 +++++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/nvx-rs/src/guest/sys/poll.rs b/nvx-rs/src/guest/sys/poll.rs index a6f9cfc0e..a30346532 100644 --- a/nvx-rs/src/guest/sys/poll.rs +++ b/nvx-rs/src/guest/sys/poll.rs @@ -15,10 +15,30 @@ use super::{ChildSignalWait, Errno, FileDescriptor, SignalDescriptor, Timer}; pub(crate) fn readable_or_hung_up( descriptors: &[BorrowedFd<'_>], timeout_milliseconds: u16, +) -> Result, Errno> { + let descriptors: Vec<_> = descriptors + .iter() + .map(|&descriptor| (descriptor, false)) + .collect(); + ready(&descriptors, timeout_milliseconds) +} + +/// Polls each descriptor for readable data or hangup, as [`readable_or_hung_up`] does, or, where +/// its flag is set, for space to write, and returns whether each one woke, in the caller's order. +pub(crate) fn ready( + descriptors: &[(BorrowedFd<'_>, bool)], + timeout_milliseconds: u16, ) -> Result, Errno> { let mut polls = descriptors .iter() - .map(|descriptor| PollFd::new(*descriptor, PollFlags::POLLIN | PollFlags::POLLHUP)) + .map(|&(descriptor, write)| { + let interest = if write { + PollFlags::POLLOUT + } else { + PollFlags::POLLIN | PollFlags::POLLHUP + }; + PollFd::new(descriptor, interest) + }) .collect::>(); nix::poll::poll(&mut polls, timeout_milliseconds)?; Ok(polls From 6908204077339d4c7d357dfffeb85c89a8039f2e Mon Sep 17 00:00:00 2001 From: Pedro Henrique Penna Date: Thu, 12 Nov 2026 01:46:23 -0800 Subject: [PATCH 2/3] guest: serve NVXC v2 in the loop of nvx-init and the managed agent NVXC v2's codec (SL4.4a) has no serving peer, so no host can stream an execution's input and output. SL4.4b (#1128) has the guest loop that nvx-init's layered roots and the managed agent share serve v2 beside v1, on nvx-init's plain-process core. The session decodes a request of either version, and answers it in its version. v2 keeps v1's header and the codes of the kinds that both have, so PING, EXEC, MAPS through the MappingTable, CAPTURE, ACTIVATE, CANCEL and STOP reach the existing handlers, in their existing order, and only their framing changes: READY carries the protocol version, the busy refusal is BUSY, and EXIT names v2's outcome. STDIN and STDIN_END feed the running execution's standard input. FEATURES, which v2 lacks, fails to decode, as any malformed frame does. ACTIVATE requests that join one activation each get READY in their own version, and STDIN or STDIN_END while the activation's record applies fails it, as any other request does. An execution of v2 runs alone: as in v1, a second EXEC gets BUSY, as does any request but its CANCEL and nvx-init's CAPTURE and ACTIVATE, which keep their handling. Its workload reads a pipe, into which each turn of the loop writes what STDIN sent, as the pipe takes it, and acknowledges with STDIN_ACK what it wrote, and which it closes once STDIN_END came and every byte went in. Bytes beyond the 64 KiB window, or after the end, end the session, as a malformed request does, and bytes that a closed pipe never took stay unacknowledged. Its output has no cap, and each turn reads at most one chunk of it, so that its output cannot keep the loop from serving control requests. A STOP of v2 cancels it, so that its EXIT reports the cancellation before STOPPED, and the session ends; a RESET abandons that STOP with the rest of the earlier epoch's work, and a STOP of v1, or during an execution of v1, still gets BUSY. STDIN and STDIN_END of an execution that ended get no answer, as its CANCEL gets none. While an execution of v2 runs, the session streams: its records go through a send queue, which never waits for the console, so that the agent keeps serving CANCEL, STOP and STDIN while the host does not read, and the loop polls the console writable while the queue holds bytes. While the queue holds 64 KiB, the loop reads none of the workload's output, which then waits on its pipes, and holds back STDIN_ACK, which acknowledges before EXIT what the pipe took meanwhile. Each turn sends what the console takes of the queue before it acknowledges input, and reads output only then, so that output cannot starve the acknowledgements. A RESET sends the queue before its ACK, and the EXIT, the first record sent without streaming, waits until the console took the queue and itself. The control descriptor is nonblocking in every session, as nvx-init's was, so the session's reads and writes always poll. v1 stays as it was, FEATURES and its output cap included. The codec's expect(dead_code) now covers only the builds without the guest loop, and names the fake OpenVMM and SL4.4c's host. dispatch::tests::streams covers v2's PING, MAPS through the table, STOP, capture and activation, beside v1 on one session; the absence of FEATURES; BUSY, CANCEL and STOP during an execution; input that cat reads until its explicit end, with acknowledgements that follow its writes; a workload that never reads its input, which leaves exactly the window unacknowledged, and a byte beyond it that ends the session; output past v1's cap; and a host that reads nothing, while which the workload waits on its pipes and a CANCEL still ends it. session::tests covers ACTIVATE requests of both versions that join one activation, and STDIN while the record applies; it and dispatch::tests::runtime take the session's request type. Part of #1128 (SL4.4b). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- nvx-rs/src/guest/managed_agent/dispatch.rs | 39 +- .../src/guest/managed_agent/dispatch/tests.rs | 1 + .../managed_agent/dispatch/tests/runtime.rs | 3 +- .../managed_agent/dispatch/tests/streams.rs | 461 ++++++++++++++++++ nvx-rs/src/guest/managed_agent/exec.rs | 208 +++++++- nvx-rs/src/guest/managed_agent/mod.rs | 10 +- nvx-rs/src/guest/managed_agent/session.rs | 234 +++++++-- .../src/guest/managed_agent/session/tests.rs | 105 +++- nvx-rs/src/guest/sys/mod.rs | 2 +- nvx-rs/src/protocol/app.rs | 7 +- nvx-rs/src/protocol/mod.rs | 2 + 11 files changed, 980 insertions(+), 92 deletions(-) create mode 100644 nvx-rs/src/guest/managed_agent/dispatch/tests/streams.rs diff --git a/nvx-rs/src/guest/managed_agent/dispatch.rs b/nvx-rs/src/guest/managed_agent/dispatch.rs index af0749ead..bcc1144bb 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch.rs @@ -12,7 +12,7 @@ use crate::protocol::{ use super::diagnostics; use super::mappings::{HostMappings, MappingError}; use super::seam::{Paths, RealSystem, System}; -use super::session::{Event, ReceiveFailure, Session}; +use super::session::{Event, ReceiveFailure, Request, Session}; use super::{AgentConfig, exec, launch, tty}; impl AgentConfig { @@ -108,6 +108,9 @@ pub(crate) enum Outcome { } /// Serves control requests until STOP or a fatal session error. +/// +/// The control descriptor is nonblocking, so that the session's send queue never waits for the +/// console while an execution of NVXC version 2 runs. pub(crate) fn run( system: &mut impl System, session: &mut Session, @@ -117,6 +120,7 @@ pub(crate) fn run( if session.stopping() { return Ok(Outcome::Stopped); } + session.descriptor().set_nonblocking()?; if let Err(error) = session.send_attach() { return if session.stopping() { Ok(Outcome::Stopped) @@ -193,7 +197,9 @@ fn handle_data( payload: &[u8], ) -> Result, Errno> { let request = match session.accept_request(payload) { - Ok(request) => request, + Ok(Request::App(request)) => request, + // The input of an execution that ended takes no more bytes. + Ok(Request::Stdin(..)) => return Ok(None), Err(failure) => { diagnostics::error(system, failure.kind.stage(), failure.status); return Err(failure.into_error()); @@ -259,7 +265,18 @@ fn handle_data( } }; session.start_activation()?; - exec::run(system, session, config, request.request_id, &execution)?; + let mut stop = None; + exec::run( + system, + session, + config, + request.request_id, + &execution, + &mut stop, + )?; + if let Some(request_id) = stop { + return stopped(system, session, request_id); + } if session.capture_pending() { return session.finish_capture(); } @@ -293,9 +310,7 @@ fn handle_data( if !request.payload.is_empty() { return Err(Errno::EPROTO); } - session.send_app(APP_STOPPED, request.request_id, 0, &[])?; - tty::drain(system, session.descriptor()); - return Ok(Some(Outcome::Stopped)); + return stopped(system, session, request.request_id); } _ => session.send_error( request.request_id, @@ -306,5 +321,17 @@ fn handle_data( Ok(None) } +/// Acknowledges the `STOP` of `request_id`, and drains the control terminal before the session +/// ends. +fn stopped( + system: &mut impl System, + session: &mut Session, + request_id: u64, +) -> Result, Errno> { + session.send_app(APP_STOPPED, request_id, 0, &[])?; + tty::drain(system, session.descriptor()); + Ok(Some(Outcome::Stopped)) +} + #[cfg(all(test, feature = "cli"))] mod tests; diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs index 4eb84a829..ef4010e94 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs @@ -225,6 +225,7 @@ fn read_outer(host: &mut UnixStream) -> (OuterHeader, Vec) { } mod runtime; +mod streams; fn response(kind: u8, request_id: u64, status: i32, payload: &[u8]) -> Response { Response { diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs index 262a6ed36..85d85d27c 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs @@ -237,7 +237,7 @@ fn prelaunch_termination(helper_ends: bool) { let Ok(Event::Data(payload)) = session.next_event() else { panic!("the complete EXEC record was not received"); }; - let Ok(request) = session.accept_request(&payload) else { + let Ok(Request::App(request)) = session.accept_request(&payload) else { panic!("the complete EXEC request was not accepted"); }; assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT); @@ -258,6 +258,7 @@ fn prelaunch_termination(helper_ends: bool) { &config, request.request_id, &execution, + &mut None, ) .err(); drop(session); diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests/streams.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests/streams.rs new file mode 100644 index 000000000..00795b901 --- /dev/null +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests/streams.rs @@ -0,0 +1,461 @@ +//! NVXC version 2 through the shared loop: its requests beside version 1's, with `READY`'s +//! version; one execution at a time, with `BUSY` for another, and `CANCEL` and `STOP` while it +//! runs; its standard input, which the agent acknowledges once its pipe took it, under the +//! window, until its explicit end; its output past version 1's cap; and the send queue, while +//! which is full the agent reads no output and still serves control requests. + +use std::os::fd::AsFd; +use std::time::{Duration, Instant}; + +use crate::guest::activation::ActivationAction; +use crate::protocol::v2::{self, Frame, STDIN_WINDOW, VERSION}; + +use super::*; + +/// A frame of version 2 that the agent sent, with its request ID. +#[derive(Debug, PartialEq, Eq)] +enum Reply { + Ready(u64, Vec), + Stdout(u64, Vec), + Stderr(u64, Vec), + Exit(u64, i32, v2::Outcome), + Stopped(u64), + Ack(u64, u32), + Busy(u64), + Error(u64, i32, Vec), +} + +impl Harness { + /// Sends a request of version 2. The agent's credit for it comes among its replies. + fn send_v2(&mut self, request_id: u64, frame: Frame<'_>) { + let frame = frame.encode(request_id).unwrap(); + let record = encode_outer( + OUTER_DATA, + &INSTANCE, + self.epoch, + self.host_sequence, + &frame, + ); + self.host.write_all(&record.unwrap()).unwrap(); + self.host_sequence += 1; + } + + /// Reads the agent's next frame, of version 2, after the credits before it. + fn next_v2(&mut self) -> Reply { + loop { + if let Some(reply) = self.next_record() { + return reply; + } + } + } + + /// Reads the agent's next frame, as [`next_v2`](Self::next_v2) does, unless none arrives + /// within `timeout`. + fn next_v2_within(&mut self, timeout: Duration) -> Option { + let deadline = Instant::now() + timeout; + loop { + let left = deadline + .saturating_duration_since(Instant::now()) + .as_millis(); + let ready = sys::readable_or_hung_up(&[self.host.as_fd()], left as u16).unwrap(); + if !ready[0] { + return None; + } + if let Some(reply) = self.next_record() { + return Some(reply); + } + } + } + + /// Reads the agent's next record: a credit, as `None`, or a frame of version 2. + fn next_record(&mut self) -> Option { + let (header, payload) = self.read_outer(); + assert_eq!( + (header.epoch, header.sequence), + (self.epoch, self.guest_sequence) + ); + self.guest_sequence += 1; + if header.record_type == OUTER_CREDIT { + return None; + } + assert_eq!(header.record_type, OUTER_DATA); + let (id, frame) = Frame::decode(&payload).unwrap(); + Some(match frame { + Frame::Ready { + version, + activation, + } => { + assert_eq!(version, VERSION); + Reply::Ready(id, activation.to_vec()) + } + Frame::Stdout(bytes) => Reply::Stdout(id, bytes.to_vec()), + Frame::Stderr(bytes) => Reply::Stderr(id, bytes.to_vec()), + Frame::Exit { status, outcome } => Reply::Exit(id, status, outcome), + Frame::Stopped => Reply::Stopped(id), + Frame::StdinAck(count) => Reply::Ack(id, count), + Frame::Busy => Reply::Busy(id), + Frame::Error { status, category } => Reply::Error(id, status, category.to_vec()), + frame => panic!("the agent sent a request: {frame:?}"), + }) + } + + /// Reads the output of execution `id` up to its first line, which its shell's `echo $$` + /// wrote, and returns that process, and what followed that line in the same reads. + fn shell_process(&mut self, id: u64) -> (sys::ProcessId, Vec) { + let mut output = Vec::new(); + let end = loop { + if let Some(end) = output.iter().position(|&byte| byte == b'\n') { + break end; + } + let Reply::Stdout(reply, bytes) = self.next_v2() else { + panic!("the shell did not report its process ID"); + }; + assert_eq!(reply, id); + output.extend(bytes); + }; + let rest = output.split_off(end + 1); + let process = std::str::from_utf8(&output).unwrap().trim_end().parse(); + ( + sys::ProcessId::from_positive(process.unwrap()).unwrap(), + rest, + ) + } + + /// Stops the agent over version 2, and returns its system. + fn stop_v2(mut self, request_id: u64) -> TestSystem { + self.send_v2(request_id, Frame::Stop); + assert_eq!(self.next_v2(), Reply::Stopped(request_id)); + let (outcome, system) = self.thread.join().unwrap(); + assert!(matches!(outcome, Ok(Outcome::Stopped))); + system + } +} + +/// Waits until `process` has ended and its status was collected, or fails the test. +fn until_gone(process: sys::ProcessId) { + let deadline = Instant::now() + Duration::from_secs(20); + while sys::process_exists(process) { + assert!(Instant::now() < deadline, "the workload did not end"); + thread::sleep(Duration::from_millis(5)); + } +} + +/// Returns the payload of an `EXEC` request for `script` in `/bin/sh`. +fn shell(script: &str) -> Vec { + workload_payload(&["/bin/sh", "-c", script], None).unwrap() +} + +#[test] +fn version_2_serves_ping_and_the_mapping_table_beside_version_1() { + let (temporary, system) = TestSystem::with_host_export(); + let export = system.paths.hostfs_dir.join("root").canonicalize().unwrap(); + let source = temporary.path().canonicalize().unwrap().join("guest/src"); + let maps = maps_payload( + 0, + &[( + b"0/src", + source.as_os_str().as_bytes(), + protocol::MAP_READ_ONLY, + )], + None, + ); + let mut harness = Harness::with_system(system, 1); + harness.send_v2(1, Frame::Ping); + assert_eq!(harness.next_v2(), Reply::Ready(1, Vec::new())); + harness.send_v2(2, Frame::Exec(&exec_payload(None))); + assert_eq!( + harness.next_v2(), + Reply::Error(2, Errno::EAGAIN as i32, b"mappings-incomplete".to_vec()) + ); + harness.send_v2(3, Frame::Maps(&maps)); + assert_eq!(harness.next_v2(), Reply::Ready(3, Vec::new())); + // The same session still serves version 1, its features request included. + assert_eq!( + harness.request(APP_PING, &[]), + response(APP_READY, 42, 0, b"") + ); + let features = harness.request(APP_FEATURES, &[]); + assert_eq!( + GuestFeatures::decode(&features.payload), + Ok(GuestFeatures::DIRECT_AGENT) + ); + let system = harness.stop_v2(4); + assert_eq!( + system.kernel.calls(), + attach_calls(&export.join("0/src"), &source, Access::ReadOnly) + ); + assert_eq!(system.tty_drained, 1); + assert!(system.diagnostics.is_empty()); +} + +#[test] +fn version_2_has_no_features_request() { + let frame = [&b"NVXC\x02\x05\0\0"[..], &42u64.to_le_bytes(), &[0; 8]].concat(); + let diagnostics = + Harness::new().fail(&encode_outer(OUTER_DATA, &INSTANCE, 1, 8, &frame).unwrap()); + let line = + |stage: &str, status: i32| format!("NVX-MANAGED-ERROR: stage={stage} status={status}\n"); + assert_eq!( + diagnostics, + [ + line("data-application", 24), + line("data", Errno::EPROTO as i32) + ] + .concat() + .as_bytes() + ); +} + +#[test] +fn version_2_carries_capture_and_activation_with_their_payloads() { + let config = AgentConfig::for_test(); + let mappings = HostMappings::announced(0); + let system = TestSystem::new(Paths::default()); + let mut harness = + Harness::spawn_with_capture(system, config, mappings, Some(Capture::fixture(true))); + harness.reset(1, 7); + harness.send_v2(1, Frame::Ping); + assert_eq!( + harness.next_v2(), + Reply::Error(1, Errno::ENOTSUP as i32, b"unsupported-operation".to_vec()) + ); + // Outside PID 1, the capture's preparation fails, which the agent reports as version 1 + // does. + harness.send_v2(2, Frame::Capture(&[0])); + assert_eq!( + harness.next_v2(), + Reply::Error(2, Errno::EPERM as i32, b"capture-failed".to_vec()) + ); + assert!(harness.stop_v2(3).diagnostics.is_empty()); + + // A restored session's activate request gets the READY of version 2, with what Ready + // reports about the activation. + let config = AgentConfig::for_test(); + let mappings = HostMappings::announced(0); + let (mut capture, payload) = Capture::activation_fixture(); + let restored = capture.restored.as_mut().unwrap(); + let now = Instant::now(); + assert!(matches!( + restored.accept(&payload, now).unwrap().0, + ActivationAction::Apply(_) + )); + let ready = restored.activation.complete(now, Ok(())).unwrap().encode(); + let system = TestSystem::new(Paths::default()); + let mut harness = Harness::spawn_with_capture(system, config, mappings, Some(capture)); + harness.reset(1, 7); + harness.send_v2(4, Frame::Activate(&payload)); + assert_eq!(harness.next_v2(), Reply::Ready(4, ready)); + assert!(harness.stop_v2(5).diagnostics.is_empty()); +} + +#[test] +fn a_version_2_execution_takes_its_input_until_its_end_and_acknowledges_what_it_wrote() -> TestResult +{ + if super::super::super::tests::isolate_process_test( + "NVX_V2_INPUT_TEST_CHILD", + "guest::managed_agent::dispatch::tests::streams::a_version_2_execution_takes_its_input_until_its_end_and_acknowledges_what_it_wrote", + ) { + return Ok(()); + } + let _process = super::super::super::tests::PROCESS_TEST_LOCK.lock()?; + let (_temporary, mut harness) = direct_harness(); + harness.send_v2(42, Frame::Exec(&workload_payload(&["/bin/cat"], None)?)); + let input: Vec = (0..200_000u32).map(|index| (index % 251) as u8).collect(); + let (mut sent, mut acknowledged, mut ended) = (0, 0, false); + let mut output = Vec::new(); + let exit = loop { + // The host never has more than the window unacknowledged. + let room = STDIN_WINDOW as usize - (sent - acknowledged); + let chunk = room.min(16 * 1024).min(input.len() - sent); + if chunk != 0 { + harness.send_v2(42, Frame::Stdin(&input[sent..sent + chunk])); + sent += chunk; + continue; + } + if sent == input.len() && !ended { + harness.send_v2(42, Frame::StdinEnd); + ended = true; + } + match harness.next_v2() { + Reply::Ack(42, count) => { + acknowledged += count as usize; + assert!(acknowledged <= sent); + } + Reply::Stdout(42, bytes) => output.extend(bytes), + Reply::Exit(42, status, outcome) => break (status, outcome), + reply => panic!("unexpected {reply:?}"), + } + }; + // `cat` ends only once its input ends, after every byte that the agent wrote into its pipe. + assert_eq!(exit, (0, v2::Outcome::Exit)); + assert_eq!(output, input); + assert_eq!(acknowledged, input.len()); + // Input of an execution that ended takes no more bytes, and gets no answer. + harness.send_v2(42, Frame::Stdin(b"late")); + harness.send_v2(42, Frame::StdinEnd); + harness.send_v2(43, Frame::Ping); + assert_eq!(harness.next_v2(), Reply::Ready(43, Vec::new())); + assert!(harness.stop_v2(44).diagnostics.is_empty()); + Ok(()) +} + +#[test] +fn a_version_2_execution_never_has_more_than_the_window_unacknowledged() -> TestResult { + if super::super::super::tests::isolate_process_test( + "NVX_V2_WINDOW_TEST_CHILD", + "guest::managed_agent::dispatch::tests::streams::a_version_2_execution_never_has_more_than_the_window_unacknowledged", + ) { + return Ok(()); + } + let _process = super::super::super::tests::PROCESS_TEST_LOCK.lock()?; + let (_temporary, mut harness) = direct_harness(); + // The workload never reads its input, so its pipe takes what it holds, and no more. + harness.send_v2(42, Frame::Exec(&shell("echo $$; exec sleep 30"))); + let (process, _) = harness.shell_process(42); + let window = STDIN_WINDOW as usize; + let chunk = [0x5a; 16 * 1024]; + let (mut sent, mut acknowledged) = (0, 0); + loop { + while sent - acknowledged < window { + let length = (window - (sent - acknowledged)).min(chunk.len()); + harness.send_v2(42, Frame::Stdin(&chunk[..length])); + sent += length; + } + match harness.next_v2_within(Duration::from_millis(500)) { + Some(Reply::Ack(42, count)) => acknowledged += count as usize, + None => break, + Some(reply) => panic!("unexpected {reply:?}"), + } + } + // The agent acknowledged what the pipe took, and holds the rest of a full window. + assert!(acknowledged > 0); + assert_eq!(sent - acknowledged, window); + assert!(sys::process_exists(process)); + // A byte beyond the window ends the session, and the workload with it. + harness.send_v2(42, Frame::Stdin(b"x")); + let (outcome, system) = harness.thread.join().unwrap(); + assert_eq!(outcome.err(), Some(Errno::EPROTO)); + assert_eq!( + system.diagnostics, + format!( + "NVX-MANAGED-ERROR: stage=data status={}\n", + Errno::EPROTO as i32 + ) + .as_bytes() + ); + until_gone(process); + Ok(()) +} + +#[test] +fn a_version_2_execution_runs_alone_and_ends_on_cancel_or_stop() -> TestResult { + if super::super::super::tests::isolate_process_test( + "NVX_V2_BUSY_TEST_CHILD", + "guest::managed_agent::dispatch::tests::streams::a_version_2_execution_runs_alone_and_ends_on_cancel_or_stop", + ) { + return Ok(()); + } + let _process = super::super::super::tests::PROCESS_TEST_LOCK.lock()?; + let (_temporary, mut harness) = direct_harness(); + let sleeper = shell("echo $$; exec sleep 30"); + harness.send_v2(42, Frame::Exec(&sleeper)); + let (process, _) = harness.shell_process(42); + harness.send_v2(43, Frame::Exec(&sleeper)); + assert_eq!(harness.next_v2(), Reply::Busy(43)); + harness.send_v2(44, Frame::Ping); + assert_eq!(harness.next_v2(), Reply::Busy(44)); + // A cancellation of another request leaves the execution running. + harness.send_v2(41, Frame::Cancel); + harness.send_v2(42, Frame::Cancel); + assert_eq!( + harness.next_v2(), + Reply::Exit(42, 128 + 9, v2::Outcome::Cancelled) + ); + until_gone(process); + // `STOP` ends the running execution, which reports it, and then the session. + harness.send_v2(45, Frame::Exec(&sleeper)); + let (process, _) = harness.shell_process(45); + harness.send_v2(46, Frame::Stop); + assert_eq!( + harness.next_v2(), + Reply::Exit(45, 128 + 9, v2::Outcome::Cancelled) + ); + assert_eq!(harness.next_v2(), Reply::Stopped(46)); + let (outcome, system) = harness.thread.join().unwrap(); + assert!(matches!(outcome, Ok(Outcome::Stopped))); + assert_eq!(system.tty_drained, 1); + assert!(system.diagnostics.is_empty()); + until_gone(process); + Ok(()) +} + +#[test] +fn a_version_2_execution_forwards_output_past_the_old_cap() -> TestResult { + if super::super::super::tests::isolate_process_test( + "NVX_V2_OUTPUT_TEST_CHILD", + "guest::managed_agent::dispatch::tests::streams::a_version_2_execution_forwards_output_past_the_old_cap", + ) { + return Ok(()); + } + let _process = super::super::super::tests::PROCESS_TEST_LOCK.lock()?; + let (_temporary, mut harness) = direct_harness(); + let total = protocol::MAX_OUTPUT_BYTES * 3 / 2; + harness.send_v2( + 42, + Frame::Exec(&shell(&format!("head -c {total} /dev/zero"))), + ); + let mut forwarded = 0; + loop { + match harness.next_v2() { + Reply::Stdout(42, bytes) => { + assert!(bytes.iter().all(|&byte| byte == 0)); + forwarded += bytes.len(); + } + Reply::Exit(42, 0, v2::Outcome::Exit) => break, + reply => panic!("unexpected {reply:?}"), + } + } + assert_eq!(forwarded, total); + assert!(harness.stop_v2(43).diagnostics.is_empty()); + Ok(()) +} + +#[test] +fn the_agent_reads_no_output_while_its_send_queue_is_full() -> TestResult { + if super::super::super::tests::isolate_process_test( + "NVX_V2_BACKPRESSURE_TEST_CHILD", + "guest::managed_agent::dispatch::tests::streams::the_agent_reads_no_output_while_its_send_queue_is_full", + ) { + return Ok(()); + } + let _process = super::super::super::tests::PROCESS_TEST_LOCK.lock()?; + let (temporary, mut harness) = direct_harness(); + let marker = temporary.path().join("finished"); + let lines = 1_000_000; + let script = format!("echo $$; seq 1 {lines}; touch '{}'", marker.display()); + harness.send_v2(42, Frame::Exec(&shell(&script))); + let (process, mut output) = harness.shell_process(42); + // The host reads nothing: the console, then the send queue, then the workload's pipe fill, + // and the workload waits to write the rest of its output. + thread::sleep(Duration::from_secs(1)); + assert!(sys::process_exists(process)); + assert!(!marker.exists()); + // The agent still serves control requests, so the cancellation ends the workload. + harness.send_v2(42, Frame::Cancel); + until_gone(process); + // What it forwarded then arrives, in order, before the outcome. + let exit = loop { + match harness.next_v2() { + Reply::Stdout(42, bytes) => output.extend(bytes), + Reply::Exit(42, status, outcome) => break (status, outcome), + reply => panic!("unexpected {reply:?}"), + } + }; + assert_eq!(exit, (128 + 9, v2::Outcome::Cancelled)); + let expected: String = (1..=lines).map(|line| format!("{line}\n")).collect(); + assert!(output.len() < expected.len()); + assert!(expected.as_bytes().starts_with(&output)); + assert!(!marker.exists()); + assert!(harness.stop_v2(43).diagnostics.is_empty()); + Ok(()) +} diff --git a/nvx-rs/src/guest/managed_agent/exec.rs b/nvx-rs/src/guest/managed_agent/exec.rs index 4decd40e6..f047514de 100644 --- a/nvx-rs/src/guest/managed_agent/exec.rs +++ b/nvx-rs/src/guest/managed_agent/exec.rs @@ -1,18 +1,19 @@ +use std::os::fd::BorrowedFd; use std::os::unix::process::ExitStatusExt; use std::time::{Duration, Instant}; use crate::guest::init::{self, Stream}; -use crate::guest::sys::{self, Errno, FileDescriptor, c_string}; +use crate::guest::sys::{self, Errno, c_string}; use crate::protocol::{ - self, APP_ACTIVATE, APP_CANCEL, APP_CAPTURE, APP_EXIT, APP_STDERR, APP_STDOUT, BUSY, - CWD_FAILED, GuestExecPayload, LAUNCH_FAILED, OUTER_DATA, + self, APP_ACTIVATE, APP_CANCEL, APP_CAPTURE, APP_EXIT, APP_STDERR, APP_STDOUT, APP_STOP, BUSY, + CWD_FAILED, GuestExecPayload, LAUNCH_FAILED, OUTER_DATA, v2::STDIN_WINDOW, }; use super::AgentConfig; use super::diagnostics::with_description; use super::launch; use super::seam::System; -use super::session::{DataFailureKind, Event as SessionEvent, ReceiveFailure, Session}; +use super::session::{DataFailureKind, Event as SessionEvent, ReceiveFailure, Request, Session}; /// Maximum bytes read from one output descriptor at a time. const OUTPUT_CHUNK_BYTES: usize = 32 * 1024; @@ -28,12 +29,21 @@ const ENVIRONMENT_DIAGNOSTIC: &[u8] = b"nvx-managed-agent: cannot configure work /// collects the statuses of the agent's children, and checks the workload's timeout. The agent /// keeps the C agent's policy: the output cap, checked for each read of up to 32 KiB, and the /// precedence of a cancellation, a timeout, and the cap over the workload's own status. +/// +/// An execution of NVXC version 2 has no output cap. Its workload reads a pipe, into which each +/// turn writes what the host sent in `STDIN` frames, under [`STDIN_WINDOW`], and acknowledges +/// what it wrote, and which it closes after `STDIN_END`. The session streams while it runs: its +/// records go through the send queue, which never waits for the console, and the agent reads no +/// more output while the queue is full, so the workload waits on its pipes while the agent serves +/// control requests. `STOP` cancels it, and once its `EXIT` reports that, `stop` holds the +/// `STOP`'s request ID. pub(crate) fn run( system: &mut impl System, session: &mut Session, config: &AgentConfig, request_id: u64, execution: &GuestExecPayload, + stop: &mut Option, ) -> Result<(), Errno> { if let Err(failure) = validate_launch_environment(system, config, execution) { match failure { @@ -58,12 +68,17 @@ pub(crate) fn run( Err(LaunchError::Other) => return session.send_error(request_id, 125, LAUNCH_FAILED), }; let started = Instant::now(); - let mut state = RunState::default(); + let mut state = RunState { + v2: session.v2(), + ..RunState::default() + }; let mut buffer = [0u8; OUTPUT_CHUNK_BYTES]; + session.stream(state.v2); while state.running(&workload) || (workload.has_open_output() && !state.ending()) { let session_ready = match poll( &workload, - (!state.session_lost).then_some(session.descriptor()), + (!state.session_lost).then_some(&*session), + !state.input.pending.is_empty(), ) { Ok(ready) => ready, Err(Errno::EINTR) => false, @@ -73,7 +88,8 @@ pub(crate) fn run( } }; if session_ready { - match handle_control(system, session, request_id) { + let stop = state.v2.then_some(&mut *stop); + match handle_control(system, session, request_id, &mut state.input, stop) { Ok(ExecControl::None) => {} Ok(ExecControl::Cancel) if state.running(&workload) && !state.cancelled => { state.cancelled = true; @@ -104,6 +120,13 @@ pub(crate) fn run( return Err(error); } } + session.answer_in(state.v2); + } + if !state.session_lost { + // The space that the console freed takes the input's acknowledgement before more + // output, so that output cannot starve it. + session.send_queued()?; + state.input.write(&mut workload, session, request_id)?; } let output = if state.session_lost || state.output_limited { &mut [][..] @@ -130,11 +153,16 @@ pub(crate) fn run( } else { state.finish(system, session, request_id, &mut workload, &mut buffer)? }; + session.stream(false); let launch_report = state.launch.take().unwrap_or(Ok(None)); if state.session_lost { return Ok(()); } + // Input that the pipe took while the send queue was full is acknowledged before the end. + if state.input.written != 0 { + session.send_ack(request_id, state.input.written)?; + } match launch_report { Ok(Some(failure)) => match failure.kind() { LaunchKind::WorkingDirectory => { @@ -251,6 +279,27 @@ impl Workload { } } + /// Lends the writer of the workload's standard input while it is open. + fn input(&self) -> Option> { + match self { + Self::Direct(workload) => workload.input(), + } + } + + /// Writes as much of `bytes` as the workload's standard input takes now. + fn write_input(&mut self, bytes: &[u8]) -> std::io::Result { + match self { + Self::Direct(workload) => workload.write_input(bytes), + } + } + + /// Closes the workload's standard input. + fn close_input(&mut self) { + match self { + Self::Direct(workload) => workload.close_input(), + } + } + fn next_event<'a>(&mut self, buffer: &'a mut [u8]) -> std::io::Result>> { match self { Self::Direct(workload) => workload.next_event(buffer).map(|event| { @@ -313,7 +362,7 @@ impl Workload { if let Some(Event::DirectKillFailed(error)) = self.next_event(&mut [])? { return Err(error); } - match poll(self, None) { + match poll(self, None, false) { Ok(_) | Err(Errno::EINTR) => {} Err(error) => return Err(error.into()), } @@ -376,10 +425,81 @@ fn settle_start( Ok(()) } +/// The standard input of an execution of NVXC version 2: what the host sent and the agent has +/// yet to write into the workload's pipe, and what it wrote and has yet to acknowledge. +#[derive(Default)] +struct Input { + /// Bytes that the host sent and the agent has not written. + pending: Vec, + /// Bytes written into the pipe and not acknowledged yet. + written: u32, + /// Bytes that the host sent and the agent has not acknowledged, at most [`STDIN_WINDOW`]: + /// those that a closed pipe never took stay unacknowledged. + outstanding: u32, + /// Whether the host ended the input with `STDIN_END`. + ended: bool, +} + +impl Input { + /// Takes a `STDIN` frame's bytes, or, for `None`, its `STDIN_END`, refusing bytes beyond the + /// window and anything after the end as `EPROTO`. + fn receive(&mut self, bytes: Option<&[u8]>) -> Result<(), Errno> { + if self.ended { + return Err(Errno::EPROTO); + } + let Some(bytes) = bytes else { + self.ended = true; + return Ok(()); + }; + self.outstanding = u32::try_from(bytes.len()) + .ok() + .and_then(|length| self.outstanding.checked_add(length)) + .filter(|&outstanding| outstanding <= STDIN_WINDOW) + .ok_or(Errno::EPROTO)?; + self.pending.extend_from_slice(bytes); + Ok(()) + } + + /// Writes as much as the workload's pipe takes, closes the pipe once the input ended and the + /// agent wrote every byte, and, unless the send queue is full, acknowledges what it wrote. + fn write( + &mut self, + workload: &mut Workload, + session: &mut Session, + request_id: u64, + ) -> Result<(), Errno> { + if !self.pending.is_empty() { + match workload.write_input(&self.pending) { + Ok(written) => { + self.pending.drain(..written); + self.written += written as u32; + } + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {} + // A closed pipe takes nothing more, and the workload reads no more of it. + Err(_) => self.pending.clear(), + } + } + if self.ended && self.pending.is_empty() { + workload.close_input(); + } + if self.written != 0 && !session.queue_full() { + session.send_ack(request_id, self.written)?; + self.outstanding -= self.written; + self.written = 0; + } + Ok(()) + } +} + /// What the agent learned of one workload while it ran: the C agent's flags, which decide the /// outcome that it reports, and the workload's launch report and outcome. #[derive(Default)] struct RunState { + /// Whether the execution is of NVXC version 2, which streams the workload's input, and caps + /// no output. + v2: bool, + /// The workload's standard input, in version 2. + input: Input, /// Bytes of output forwarded so far, which the 1 MiB cap bounds. output_bytes: usize, /// Whether the host cancelled the workload while it ran. @@ -425,11 +545,19 @@ impl RunState { workload: &mut Workload, buffer: &mut [u8], ) -> Result<(), Errno> { + // An execution of version 2, whose output has no cap, reads one chunk of it a turn, so + // that its output cannot keep the agent from serving control requests or its timeout. + let mut read = false; loop { if let Some((id, status)) = session.take_status() { workload.observe(id, status); } - let event = match workload.next_event(buffer) { + let chunk = if session.queue_full() || (self.v2 && read) { + &mut [][..] + } else { + &mut *buffer + }; + let event = match workload.next_event(chunk) { Ok(Some(event)) => event, Ok(None) => return Ok(()), Err(error) => { @@ -439,7 +567,7 @@ impl RunState { }; match event { Event::Output(stream, bytes) => { - if self.output_bytes > protocol::MAX_OUTPUT_BYTES - bytes.len() { + if !self.v2 && self.output_bytes > protocol::MAX_OUTPUT_BYTES - bytes.len() { self.output_limited = true; // The cap ends forwarding: read and drop the rest of both streams, // as a lost session does, so the client receives a prefix of what the @@ -451,6 +579,7 @@ impl RunState { return Ok(()); } self.output_bytes += bytes.len(); + read = true; let kind = match stream { Stream::Stdout => APP_STDOUT, Stream::Stderr => APP_STDERR, @@ -499,7 +628,7 @@ impl RunState { if self.outcome.is_some() || self.status_lost { return Ok(self.outcome); } - match poll(workload, None) { + match poll(workload, None, false) { Ok(_) | Err(Errno::EINTR) => {} Err(error) => return Err(error), } @@ -507,15 +636,23 @@ impl RunState { } } -/// Waits at most one poll interval until one of the workload's descriptors or the session's, -/// if given, polls ready, and returns whether the session's did. -fn poll(workload: &Workload, session: Option<&FileDescriptor>) -> Result { - let mut descriptors = workload.descriptors(); +/// Waits at most one poll interval until one of the workload's descriptors polls ready, but its +/// output while the session's send queue is full, its input, if `input`, writable, or the +/// session's descriptor, if given, readable, or writable while it has queued bytes, and returns +/// whether the session's descriptor polled readable. +fn poll(workload: &Workload, session: Option<&Session>, input: bool) -> Result { + let mut descriptors = Vec::new(); if let Some(session) = session { - descriptors.push(session.borrowed()); + let descriptor = session.descriptor().borrowed(); + descriptors.push((descriptor, false)); + descriptors.extend(session.queued().then_some((descriptor, true))); } - let ready = sys::readable_or_hung_up(&descriptors, POLL_INTERVAL_MS)?; - Ok(session.is_some() && ready.last().copied().unwrap_or(false)) + if !session.is_some_and(Session::queue_full) { + descriptors.extend(workload.descriptors().into_iter().map(|fd| (fd, false))); + } + descriptors.extend(workload.input().filter(|_| input).map(|fd| (fd, true))); + let ready = sys::ready(&descriptors, POLL_INTERVAL_MS)?; + Ok(session.is_some() && ready[0]) } /// Reports the kill failures that the workload's execution has yet to report. @@ -553,9 +690,9 @@ fn launch( config: &AgentConfig, execution: &GuestExecPayload, ) -> Result { - let workload = system - .start_direct(launch::direct_request(config, execution)?) - .map_err(LaunchError::Start)?; + let mut request = launch::direct_request(config, execution)?; + request.input = session.v2(); + let workload = system.start_direct(request).map_err(LaunchError::Start)?; session.watch(workload.id()); Ok(Workload::Direct(workload)) } @@ -640,15 +777,26 @@ enum ExecControl { Capture, } -/// Processes one control-console record while a workload runs. +/// Processes one control-console record while a workload runs: in version 2, the workload's +/// standard input, of which bytes beyond the window or after its end end the session, and, for an +/// execution of version 2, whose `stop` is given, a `STOP`, which cancels the workload and sets +/// `stop`. fn handle_control( system: &mut impl System, session: &mut Session, exec_id: u64, + input: &mut Input, + stop: Option<&mut Option>, ) -> Result { let payload = match session.next_event() { Ok(SessionEvent::Reset) => { + // The new epoch's acknowledgement follows what the earlier one queued, and the + // earlier epoch's `STOP` is abandoned with its work. + session.stream(false); session.acknowledge()?; + if let Some(stop) = stop { + *stop = None; + } return Ok(ExecControl::SessionLost); } Ok(SessionEvent::Data(payload)) => payload, @@ -656,7 +804,14 @@ fn handle_control( Err(failure) => return Err(report_control_failure(system, failure)), }; let request = match session.accept_request(&payload) { - Ok(request) => request, + Ok(Request::App(request)) => request, + // The input of an execution that ended takes no more bytes. + Ok(Request::Stdin(id, bytes)) => { + if id == exec_id { + input.receive(bytes)?; + } + return Ok(ExecControl::None); + } Err(failure) => { if failure.kind != DataFailureKind::Credit { super::diagnostics::error(system, "exec-record", i32::from(OUTER_DATA)); @@ -664,6 +819,13 @@ fn handle_control( return Err(failure.into_error()); } }; + if request.kind == APP_STOP + && session.v2() + && let Some(stop) = stop + { + *stop = Some(request.request_id); + return Ok(ExecControl::Cancel); + } if request.kind == APP_CANCEL { return Ok( if request.payload.is_empty() && request.request_id == exec_id { diff --git a/nvx-rs/src/guest/managed_agent/mod.rs b/nvx-rs/src/guest/managed_agent/mod.rs index f27f53ec8..e6055dffc 100644 --- a/nvx-rs/src/guest/managed_agent/mod.rs +++ b/nvx-rs/src/guest/managed_agent/mod.rs @@ -9,6 +9,14 @@ //! status and error categories, 1 MiB output and receive windows, and timeout and cancellation //! behavior. //! +//! It also serves NVXC version 2, whose frames carry the same requests, and in which each answer +//! of a request takes its version. An execution of version 2 runs as one of version 1 does, alone, +//! and a second `EXEC` gets `BUSY`, but its workload reads a pipe of the standard input that the +//! host streams, which the agent acknowledges as it writes it, up to its explicit end; its output +//! has no cap; and `STOP` ends it. While it runs, the agent's records wait in a send queue rather +//! than for the console, and the agent reads none of the workload's output while the queue is +//! full. +//! //! The agent exits with 0, or with 125 when it fails. //! //! # Structure @@ -29,7 +37,7 @@ //! ends it, and reports its launch and outcome, and the //! agent forwards the output as NVXC frames, serves the control session while the workload runs, //! and reports the outcome. The agent keeps its own policy: the null device as the workload's -//! standard input, the identity, the output cap, +//! standard input and the output cap of version 1, the identity, //! the timeout, the precedence of a cancellation, a timeout, and the cap over the workload's own //! status, and the outcome that it reports for a failed launch stage. It builds the workload's //! environment in memory, as C's environment functions changed it, and passes it to `execve`, so diff --git a/nvx-rs/src/guest/managed_agent/session.rs b/nvx-rs/src/guest/managed_agent/session.rs index 15aa65004..aac1bbf22 100644 --- a/nvx-rs/src/guest/managed_agent/session.rs +++ b/nvx-rs/src/guest/managed_agent/session.rs @@ -9,10 +9,23 @@ use crate::guest::nvxs::{GuestConnection, GuestConnectionError}; use crate::guest::snapshot::{Captured, Disposition}; use crate::guest::sys::{self, Errno, FileDescriptor, Pollable}; use crate::protocol::nvxs::{GuestEvent, RecordError, RecordType, SessionError}; -use crate::protocol::{self, APP_ERROR, OUTER_MAX_PAYLOAD, OUTER_RECEIVE_WINDOW}; +use crate::protocol::{ + self, APP_ERROR, APP_EXIT, APP_READY, APP_STDERR, APP_STDOUT, APP_STOPPED, OUTER_MAX_PAYLOAD, + OUTER_RECEIVE_WINDOW, v2, +}; + +/// Bytes that the send queue holds before the agent stops reading an execution's output, until +/// the control console takes them. +const SEND_QUEUE_BYTES: usize = 64 * 1024; /// The agent's control session: the guest's role of the shared NVXS session on the control /// console, and the NVXC frames that its DATA records carry. +/// +/// It serves requests of NVXC versions 1 and 2, which keep the codes of the kinds that both have, +/// and answers each in the version of the request that it serves. While an execution of version +/// 2 runs, the session's records go through a send queue, which never waits for the console, so +/// that the agent keeps serving control requests while the host does not read; the record that +/// it sends next without streaming waits until the console has taken the queue and itself. pub(crate) struct Session<'r> { /// The guest's NVXS connection on the owned control-console descriptor. connection: GuestConnection>, @@ -21,6 +34,9 @@ pub(crate) struct Session<'r> { /// Error of the connection's latest failure, which a connection that a failure ended reports /// again. last_error: Option, + /// Whether the request that the session serves is of version 2, whose frames its answers + /// take. + v2: bool, } /// Byte transport only: the existing connection still owns all record framing. @@ -32,6 +48,10 @@ struct Control<'r> { outcome: RefCell>, /// Bounds partial I/O until the restored activation's Ready has been written. deadline: Cell>, + /// Records that the console has not taken yet, in order. + queue: RefCell>, + /// Whether writes add to the queue rather than wait for the console. + queueing: Cell, } impl Control<'_> { @@ -74,6 +94,21 @@ impl Control<'_> { .copied() .map_err(|_| Errno::ECHILD) } + + /// Writes as much of the queue as the console takes without waiting. + fn send_queued(&self) -> io::Result<()> { + let mut queue = self.queue.borrow_mut(); + while !queue.is_empty() { + match self.descriptor.write(&queue) { + Ok(0) => return Err(Errno::EIO.into()), + Ok(written) => drop(queue.drain(..written)), + Err(Errno::EAGAIN) => break, + Err(Errno::EINTR) => {} + Err(error) => return Err(error.into()), + } + } + Ok(()) + } } impl Read for Control<'_> { @@ -81,9 +116,6 @@ impl Read for Control<'_> { if bytes.is_empty() { return Read::read(&mut self.descriptor, bytes); } - if self.runtime.is_none() && self.deadline.get().is_none() { - return Read::read(&mut self.descriptor, bytes); - } loop { if self.runtime.is_some() { self.service_leader()?; @@ -114,12 +146,16 @@ impl Read for Control<'_> { impl Write for Control<'_> { fn write(&mut self, bytes: &[u8]) -> io::Result { - if self.runtime.is_none() && self.deadline.get().is_none() { - return Write::write(&mut self.descriptor, bytes); - } if bytes.is_empty() { return Write::write(&mut self.descriptor, bytes); } + if self.queueing.get() { + self.queue.get_mut().extend_from_slice(bytes); + self.send_queued()?; + return Ok(bytes.len()); + } + let queued = std::mem::take(self.queue.get_mut()); + self.write_all(&queued)?; loop { self.remaining()?; if self.runtime.is_some() { @@ -169,6 +205,16 @@ pub(crate) enum Event { Data(Vec), } +/// An NVXC request that the session accepted, whose answers take its version. +pub(crate) enum Request<'a> { + /// A request of version 1, or one of version 2 of a kind that version 1 has too, with the + /// code that both give it. + App(protocol::AppFrame<'a>), + /// Version 2's standard input of the execution of this ID: a `STDIN` frame's bytes, or `None` + /// for its `STDIN_END`. + Stdin(u64, Option<&'a [u8]>), +} + /// A record from the broker that ended the session, classified by the stage at which the C agent /// reported it. pub(crate) enum ReceiveFailure { @@ -242,9 +288,12 @@ impl<'r> Session<'r> { status: Cell::new(None), outcome: RefCell::new(Ok(false)), deadline: Cell::new(None), + queue: RefCell::new(Vec::new()), + queueing: Cell::new(false), }), capture: None, last_error: None, + v2: false, } } @@ -452,7 +501,8 @@ impl<'r> Session<'r> { )); } }; - let mut replies = vec![request.request_id]; + // Each joined request gets its Ready in its own version. + let mut replies = vec![(request.request_id, self.v2)]; while sys::readable_or_hung_up(&[self.descriptor().borrowed()], 0)?[0] { match self .next_event() @@ -464,17 +514,22 @@ impl<'r> Session<'r> { replies.clear(); } Event::Data(bytes) => { - let retry = self + let retry = match self .accept_request(&bytes) - .map_err(DataFailure::into_error)?; - if retry.kind != protocol::APP_ACTIVATE { - return Err(activation_failure( - restored - .activation - .input(Instant::now(), ActivationInput::Request) - .expect_err("an activation takes no request yet"), - )); - } + .map_err(DataFailure::into_error)? + { + Request::App(retry) if retry.kind == protocol::APP_ACTIVATE => { + retry + } + _ => { + return Err(activation_failure( + restored + .activation + .input(Instant::now(), ActivationInput::Request) + .expect_err("an activation takes no request yet"), + )); + } + }; if !matches!( restored .accept(retry.payload, Instant::now()) @@ -484,7 +539,7 @@ impl<'r> Session<'r> { ) { return Err(Errno::EPROTO); } - replies.push(retry.request_id); + replies.push((retry.request_id, self.v2)); } } } @@ -494,10 +549,13 @@ impl<'r> Session<'r> { .map_err(activation_failure)?; (Some(retained), ready, replies) } - ActivationAction::Ready(ready) => (None, ready, vec![request.request_id]), + ActivationAction::Ready(ready) => { + (None, ready, vec![(request.request_id, self.v2)]) + } _ => return Err(Errno::EPROTO), }; - for id in replies { + for (id, v2) in replies { + self.v2 = v2; self.send_app(protocol::APP_READY, id, 0, &ready.encode())?; } self.connection.get_ref().deadline.set(None); @@ -616,29 +674,63 @@ impl<'r> Session<'r> { acknowledged.map_err(|error| self.error(error)) } - /// Decodes the NVXC request that a DATA payload carries, then returns the payload's bytes to - /// the receive window before the agent serves the request, as the C agent did. + /// Decodes the NVXC request that a DATA payload carries, of version 1 or 2, then returns the + /// payload's bytes to the receive window before the agent serves the request, as the C agent + /// did. The session answers in the request's version until it accepts another. pub(crate) fn accept_request<'a>( &mut self, payload: &'a [u8], - ) -> Result, DataFailure> { + ) -> Result, DataFailure> { // A DATA payload has at most `OUTER_MAX_PAYLOAD` bytes, so its length fits either type. let status = payload.len() as i32; - let frame = protocol::decode_app(payload) - .map_err(|_| data_failure(DataFailureKind::Application, status))?; - if frame.request_id == 0 || frame.status != 0 { - return Err(data_failure(DataFailureKind::Application, status)); + let streams = payload.get(4) == Some(&v2::VERSION); + let request = if streams { + v2::Frame::decode(payload) + .ok() + .map(|(request_id, frame)| match frame { + v2::Frame::Stdin(bytes) => Request::Stdin(request_id, Some(bytes)), + v2::Frame::StdinEnd => Request::Stdin(request_id, None), + // Version 2 keeps version 1's header, and the codes of the kinds that both + // have, which its decoder checked. + _ => Request::App(protocol::AppFrame { + kind: payload[5], + request_id, + status: 0, + payload: &payload[protocol::APP_HEADER_LEN..], + }), + }) + } else { + protocol::decode_app(payload) + .ok() + .filter(|frame| frame.request_id != 0 && frame.status == 0) + .map(Request::App) } + .ok_or_else(|| data_failure(DataFailureKind::Application, status))?; let credited = self.connection.credit(payload.len() as u32); credited.map_err(|error| DataFailure { kind: DataFailureKind::Credit, status, error: self.error(error), })?; - Ok(frame) + self.v2 = streams; + Ok(request) } - /// Encodes and sends one application frame, in one DATA record. + /// Whether the session answers in version 2. + pub(crate) fn v2(&self) -> bool { + self.v2 + } + + /// Answers in version 2, or in version 1, such as an execution's frames after a control + /// request of its own version. + pub(crate) fn answer_in(&mut self, v2: bool) { + self.v2 = v2; + } + + /// Encodes and sends one application frame, in one DATA record, in the session's version. + /// + /// In version 2, `READY` starts with the protocol version, an `ERROR` of the `busy` category + /// is `BUSY`, and an `EXIT` names its outcome as version 1 does. pub(crate) fn send_app( &mut self, kind: u8, @@ -646,17 +738,74 @@ impl<'r> Session<'r> { status: i32, payload: &[u8], ) -> Result<(), Errno> { - if protocol::APP_HEADER_LEN - .checked_add(payload.len()) - .is_none_or(|length| length > OUTER_MAX_PAYLOAD) - { - return Err(Errno::EMSGSIZE); - } - let frame = protocol::encode_app(kind, request_id, status, payload); + let frame = if self.v2 { + let frame = match kind { + APP_READY => v2::Frame::Ready { + version: v2::VERSION, + activation: payload, + }, + APP_STDOUT => v2::Frame::Stdout(payload), + APP_STDERR => v2::Frame::Stderr(payload), + APP_EXIT => v2::Frame::Exit { + status, + outcome: outcome(payload), + }, + APP_STOPPED => v2::Frame::Stopped, + APP_ERROR if payload == protocol::BUSY.as_bytes() => v2::Frame::Busy, + APP_ERROR => v2::Frame::Error { + status, + category: payload, + }, + _ => return Err(Errno::EINVAL), + }; + frame.encode(request_id).map_err(|_| Errno::EMSGSIZE)? + } else { + if protocol::APP_HEADER_LEN + .checked_add(payload.len()) + .is_none_or(|length| length > OUTER_MAX_PAYLOAD) + { + return Err(Errno::EMSGSIZE); + } + protocol::encode_app(kind, request_id, status, payload) + }; + let sent = self.connection.send(&frame); + sent.map_err(|error| self.error(error)) + } + + /// Acknowledges `count` bytes written into the standard input of the execution of + /// `request_id`, in version 2. + pub(crate) fn send_ack(&mut self, request_id: u64, count: u32) -> Result<(), Errno> { + let frame = v2::Frame::StdinAck(count) + .encode(request_id) + .map_err(|_| Errno::EINVAL)?; let sent = self.connection.send(&frame); sent.map_err(|error| self.error(error)) } + /// Streams, while an execution of version 2 runs: every record goes through the send queue, + /// which never waits for the console, and which the next record sent without streaming + /// follows. + pub(crate) fn stream(&self, on: bool) { + self.connection.get_ref().queueing.set(on); + } + + /// Whether the send queue holds [`SEND_QUEUE_BYTES`], so that the agent reads no more of an + /// execution's output. + pub(crate) fn queue_full(&self) -> bool { + self.connection.get_ref().queue.borrow().len() >= SEND_QUEUE_BYTES + } + + /// Whether the send queue holds bytes, so that the agent polls the console writable. + pub(crate) fn queued(&self) -> bool { + !self.connection.get_ref().queue.borrow().is_empty() + } + + /// Writes as much of the send queue as the console takes without waiting. + pub(crate) fn send_queued(&mut self) -> Result<(), Errno> { + let sent = self.connection.get_ref().send_queued(); + sent.map_err(|error| self.error(GuestConnectionError::Io(error))) + } + /// Sends one application ERROR frame with an ASCII category. pub(crate) fn send_error( &mut self, @@ -705,6 +854,17 @@ fn activation_failure(error: impl std::fmt::Display) -> Errno { Errno::EPROTO } +/// Returns version 2's outcome of the execution whose version-1 `EXIT` category is `category`. +fn outcome(category: &[u8]) -> v2::Outcome { + match category { + b"exit" => v2::Outcome::Exit, + b"signal" => v2::Outcome::Signal, + b"timeout" => v2::Outcome::Timeout, + b"cancelled" => v2::Outcome::Cancelled, + _ => v2::Outcome::Failed, + } +} + /// Returns the stage at which the C agent reported a record that the session refused, or `None` /// for one that it reported as a failure to read. /// diff --git a/nvx-rs/src/guest/managed_agent/session/tests.rs b/nvx-rs/src/guest/managed_agent/session/tests.rs index 62232fd52..482e5340b 100644 --- a/nvx-rs/src/guest/managed_agent/session/tests.rs +++ b/nvx-rs/src/guest/managed_agent/session/tests.rs @@ -78,6 +78,19 @@ impl Peer { self.host_sequence += 1; } + fn send_v2(&mut self, id: u64, frame: v2::Frame<'_>) { + let bytes = frame.encode(id).unwrap(); + let record = encode_outer( + OUTER_DATA, + &INSTANCE, + self.epoch, + self.host_sequence, + &bytes, + ); + self.stream.write_all(&record.unwrap()).unwrap(); + self.host_sequence += 1; + } + fn credit(&mut self) { let (header, _) = self.read(); assert_eq!(header.record_type, OUTER_CREDIT); @@ -142,9 +155,9 @@ fn a_replacement_session_joins_one_activation_and_caches_ready_without_reapplyin let (record, _) = protocol::decode_activate_payload(&payload).unwrap(); let ready = crate::activation::ReadyActivation::of(&record).encode(); let bytes = activate_bytes(&mut session, &mut peer, 61, &payload); - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid activate")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; peer.credit(); let retained = session .activate_with(request, |restored, _, accepted, workload| { @@ -167,9 +180,9 @@ fn a_replacement_session_joins_one_activation_and_caches_ready_without_reapplyin assert!(!session.rootless()); assert_eq!(session.connection.get_ref().deadline.get(), None); let bytes = activate_bytes(&mut session, &mut peer, 63, &payload); - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid retry")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid retry"); + }; peer.credit(); assert!( session @@ -188,6 +201,37 @@ fn a_replacement_session_joins_one_activation_and_caches_ready_without_reapplyin ); } +#[test] +fn joined_activations_get_ready_in_their_own_versions() { + let (mut session, mut peer, payload) = restored(); + let (record, _) = protocol::decode_activate_payload(&payload).unwrap(); + let ready = crate::activation::ReadyActivation::of(&record).encode(); + let bytes = activate_bytes(&mut session, &mut peer, 61, &payload); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; + peer.credit(); + let retained = session.activate_with(request, |_, _, _, _| { + peer.send_v2(62, v2::Frame::Activate(&payload)); + Ok("one applied space") + }); + assert_eq!(retained.unwrap(), Some("one applied space")); + peer.credit(); + peer.app(APP_READY, 61, 0, &ready); + let (_, frame) = peer.read(); + let activation = ready.as_slice(); + assert_eq!( + v2::Frame::decode(&frame).unwrap(), + ( + 62, + v2::Frame::Ready { + version: v2::VERSION, + activation + } + ) + ); +} + #[test] fn postactivation_capture_stays_refused_without_pending_cleanup() { let (mut session, mut peer, payload) = restored(); @@ -229,9 +273,9 @@ fn changed_record_or_workload_bytes_during_application_fail_without_ready() { changed.extend(record.encode()); } let bytes = activate_bytes(&mut session, &mut peer, 64, &payload); - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid activate")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; peer.credit(); let result = session.activate_with(request, |_, _, _, _| { peer.send(protocol::APP_ACTIVATE, 65, &changed); @@ -252,9 +296,9 @@ fn changed_record_or_workload_bytes_during_application_fail_without_ready() { fn failed_application_never_reports_ready_or_admits_root_dependent_work() { let (mut session, mut peer, payload) = restored(); let bytes = activate_bytes(&mut session, &mut peer, 66, &payload); - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid activate")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; peer.credit(); assert_eq!( session @@ -283,9 +327,9 @@ fn a_failed_restored_activation_gives_init_the_status_of_its_class() { let (mut session, mut peer, payload) = restored(); assert_eq!(session.activation_status(), None); let bytes = activate_bytes(&mut session, &mut peer, 68, &payload); - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid activate")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; peer.credit(); let failure = crate::Error::io("the network settings cannot be applied").with_source( crate::guest::activation::ApplyError::Network { @@ -306,9 +350,9 @@ fn a_failed_restored_activation_gives_init_the_status_of_its_class() { // A request while the record applies fails the activation as the protocol's failure. let (mut session, mut peer, payload) = restored(); let bytes = activate_bytes(&mut session, &mut peer, 69, &payload); - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid activate")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; peer.credit(); let result = session.activate_with(request, |_, _, _, _| { peer.send(protocol::APP_PING, 70, &[]); @@ -325,6 +369,25 @@ fn a_failed_restored_activation_gives_init_the_status_of_its_class() { ); } +#[test] +fn standard_input_while_the_record_applies_fails_the_activation_as_the_protocols_failure() { + let (mut session, mut peer, payload) = restored(); + let bytes = activate_bytes(&mut session, &mut peer, 71, &payload); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; + peer.credit(); + let result = session.activate_with(request, |_, _, _, _| { + peer.send_v2(71, v2::Frame::Stdin(b"x")); + Ok(()) + }); + assert_eq!(result.err(), Some(Errno::EPROTO)); + assert_eq!( + session.activation_status(), + Some(protocol::status::ACTIVATION_PROTOCOL) + ); +} + fn short_deadline(session: &mut Session<'_>) -> Instant { let now = Instant::now(); let repaired = crate::guest::snapshot::repair::tests::repaired_fixture(); @@ -390,9 +453,9 @@ fn queue( else { panic!("the capture request is DATA"); }; - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid frame")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid frame"); + }; let accepted = session.queue_capture(request, active).unwrap(); peer.credit(); accepted diff --git a/nvx-rs/src/guest/sys/mod.rs b/nvx-rs/src/guest/sys/mod.rs index 196b2fe1e..669326e24 100644 --- a/nvx-rs/src/guest/sys/mod.rs +++ b/nvx-rs/src/guest/sys/mod.rs @@ -111,7 +111,7 @@ pub(crate) use ioctl::{ pub(crate) use mapping::MemoryMapping; #[cfg(target_arch = "x86_64")] pub(crate) use msr::{MsrAccess, MsrFile}; -pub(crate) use poll::{Pollable, hung_up, poll, readable_or_hung_up}; +pub(crate) use poll::{Pollable, hung_up, poll, readable_or_hung_up, ready}; pub(crate) use port::{PORT_DEVICE, Port}; #[cfg(target_arch = "x86_64")] pub(crate) use port::{PortAccess, PortError, check_released, power_off, probe_port}; diff --git a/nvx-rs/src/protocol/app.rs b/nvx-rs/src/protocol/app.rs index 5cf505fd6..a8bf72ccf 100644 --- a/nvx-rs/src/protocol/app.rs +++ b/nvx-rs/src/protocol/app.rs @@ -49,10 +49,13 @@ pub(crate) mod workload { all(feature = "guest", target_os = "linux", target_arch = "x86_64") ))] #[cfg_attr( - not(test), + not(any( + test, + all(feature = "guest", target_os = "linux", target_arch = "x86_64") + )), expect( dead_code, - reason = "SL4.4b's guest loop and fake OpenVMM serve NVXC v2, and SL4.4c's host speaks it" + reason = "SL4.4b's fake OpenVMM serves NVXC v2, and SL4.4c's host speaks it" ) )] pub(crate) mod v2; diff --git a/nvx-rs/src/protocol/mod.rs b/nvx-rs/src/protocol/mod.rs index de401d1f1..2f0486c46 100644 --- a/nvx-rs/src/protocol/mod.rs +++ b/nvx-rs/src/protocol/mod.rs @@ -85,6 +85,8 @@ pub(crate) use app::ExitCategory; pub(crate) use app::encode_activate_payload; #[cfg(feature = "cli")] pub(crate) use app::encode_capture_payload; +#[cfg(all(feature = "guest", target_os = "linux", target_arch = "x86_64"))] +pub(crate) use app::v2; #[cfg(any( feature = "cli", all(feature = "guest", target_os = "linux", target_arch = "x86_64") From 23bf183219f9024fd8611b4b1415c0fbc6b7e208 Mon Sep 17 00:00:00 2001 From: Pedro Henrique Penna Date: Thu, 12 Nov 2026 01:46:46 -0800 Subject: [PATCH 3/3] testing: serve NVXC v2 in the fake OpenVMM The CLI's integration tests run against the fake OpenVMM, which served only NVXC v1, so SL4.4c could not switch them to v2. SL4.4b (#1128) has the fake serve v2 beside v1, with the guest loop's observable behavior. The fake decodes a request of either version and answers it in its version: READY carries the protocol version, the busy refusal is BUSY, and EXIT names v2's outcome, while PING, EXEC, MAPS and STOP keep their handling. A workload of v2 runs alone, and another EXEC gets BUSY, as other requests do while it runs, each refusal in the version of the request that it refuses. Its output has no cap. Its standard input is what STDIN frames sent, which the fake holds until the workload reads it, and acknowledges as it does, until STDIN_END: the new script command cat copies it to standard output until its end, or until the workload's timeout. More than the 64 KiB window unacknowledged, or input after its end, ends the session. A STOP while the workload sleeps or reads its input ends it with the cancelled outcome, and the fake acknowledges the STOP and stops, unless fake_ignore_stop=1 ignores the STOP, as between workloads, or fake_drop_stop=1 closes the connection after the EXIT without the acknowledgement. STDIN and STDIN_END of a workload that ended get no answer. A workload of v1 reads no input and keeps its 1 MiB cap, and FEATURES and GuestFeatures stay v1's alone. With both servers in place, the codec is gated by the features of its users, testing and guest on Linux x86-64, so its expect(dead_code) and its seam in the test-layout allowlist go, and cli builds it again once SL4.4c's host speaks it. The codec's docs say what NVX's peers do with an execution of v2, and the README which programs serve it. The fake's tests, on Unix, drive its control session over a socket pair: cat's input until its explicit end, with an acknowledgement of each read, a cat that waits for input until its timeout, and a byte beyond the window that ends the session; output past v1's cap, which v1 still caps on the same session; and BUSY, CANCEL and STOP while a workload runs. Part of #1128 (SL4.4b). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- nvx-rs/README.md | 4 +- nvx-rs/src/protocol/app.rs | 11 -- nvx-rs/src/protocol/app/v2.rs | 6 + nvx-rs/src/protocol/mod.rs | 6 +- nvx-rs/src/testing/fake_openvmm.rs | 213 +++++++++++++++++++--- nvx-rs/src/testing/fake_openvmm/tests.rs | 218 +++++++++++++++++++++++ scripts/test-layout-allowlist.json | 3 +- 7 files changed, 419 insertions(+), 42 deletions(-) diff --git a/nvx-rs/README.md b/nvx-rs/README.md index 9ab0128be..f87896aef 100644 --- a/nvx-rs/README.md +++ b/nvx-rs/README.md @@ -161,7 +161,9 @@ frames, the `EXEC` payload that the host encodes and the guest agent decodes, and the control features that a guest image advertises. Beside version 1's frames, `protocol::app::v2` holds the codec of version 2's, which carry every control path, with version 1's payloads, and stream an execution's standard -input, output and error. +input, output and error. The guest loop that nvx-init and the managed agent +share, and the fake OpenVMM, serve both versions, so `v2` follows their +features, `guest` and `testing`, until the host speaks it too. The [conformance vectors](tests/vectors/README.md) of version 1, `tests/vectors/nvxc-v1.txt`, and OpenVMM's vectors and cases of the `NVXS` diff --git a/nvx-rs/src/protocol/app.rs b/nvx-rs/src/protocol/app.rs index a8bf72ccf..6ca904452 100644 --- a/nvx-rs/src/protocol/app.rs +++ b/nvx-rs/src/protocol/app.rs @@ -44,20 +44,9 @@ pub(crate) mod workload { } #[cfg(any( - feature = "cli", feature = "testing", all(feature = "guest", target_os = "linux", target_arch = "x86_64") ))] -#[cfg_attr( - not(any( - test, - all(feature = "guest", target_os = "linux", target_arch = "x86_64") - )), - expect( - dead_code, - reason = "SL4.4b's fake OpenVMM serves NVXC v2, and SL4.4c's host speaks it" - ) -)] pub(crate) mod v2; /// Magic of every application frame. diff --git a/nvx-rs/src/protocol/app/v2.rs b/nvx-rs/src/protocol/app/v2.rs index d6ba1757a..a509a1e2f 100644 --- a/nvx-rs/src/protocol/app/v2.rs +++ b/nvx-rs/src/protocol/app/v2.rs @@ -46,6 +46,12 @@ //! has written them into the execution's standard input, and the host never has more than //! [`STDIN_WINDOW`] bytes that the guest agent has not acknowledged. `CANCEL` ends the execution, //! whose `EXIT` reports it. Version 1's `FEATURES` request, `0x05`, has no place in version 2. +//! +//! NVX's guest agents and the fake OpenVMM run one execution at a time. A `STOP` while it runs +//! ends it, whose `EXIT` comes before `STOPPED`. Bytes that its standard input never took, as +//! once the workload closed it, stay unacknowledged; `STDIN`, `STDIN_END` and `CANCEL` of an +//! execution that ended get no answer; and standard input beyond the window, or after its end, +//! ends the session. use crate::protocol::{OUTER_MAX_PAYLOAD, ProtocolError, array, violation}; diff --git a/nvx-rs/src/protocol/mod.rs b/nvx-rs/src/protocol/mod.rs index 2f0486c46..e9ae7e0f7 100644 --- a/nvx-rs/src/protocol/mod.rs +++ b/nvx-rs/src/protocol/mod.rs @@ -74,6 +74,7 @@ pub(crate) use app::APP_ACTIVATE; pub(crate) use app::APP_CAPTURE; #[cfg(any( feature = "cli", + feature = "testing", all(feature = "guest", target_os = "linux", target_arch = "x86_64") ))] pub(crate) use app::APP_HEADER_LEN; @@ -85,7 +86,10 @@ pub(crate) use app::ExitCategory; pub(crate) use app::encode_activate_payload; #[cfg(feature = "cli")] pub(crate) use app::encode_capture_payload; -#[cfg(all(feature = "guest", target_os = "linux", target_arch = "x86_64"))] +#[cfg(any( + feature = "testing", + all(feature = "guest", target_os = "linux", target_arch = "x86_64") +))] pub(crate) use app::v2; #[cfg(any( feature = "cli", diff --git a/nvx-rs/src/testing/fake_openvmm.rs b/nvx-rs/src/testing/fake_openvmm.rs index a6da98ec8..b488b66c8 100644 --- a/nvx-rs/src/testing/fake_openvmm.rs +++ b/nvx-rs/src/testing/fake_openvmm.rs @@ -17,10 +17,12 @@ //! or `printenv` in any directory, which prints its environment. A script is a `;`-separated //! list of commands: `echo TEXT`, `echoerr TEXT`, `printerr TEXT`, which writes `TEXT` to //! standard error without a line end, `sleep MS`, `exit CODE`, `signal NUMBER`, `flood BYTES`, -//! `write KEY VALUE`, `read KEY`, `mkdir DIRECTORY`, `touch PATH`, which creates the host file -//! `PATH`, as a workload that writes a read-write share does, `env`, `printenv NAME`, `pwd`, -//! `timeout`, `fail`, and `launchfail`. Like `sh`, a script exits with the status of its -//! last command; `printenv NAME` prints nothing and fails with status 1 when `NAME` is unset. +//! `cat`, which copies the workload's standard input to its standard output until the input ends +//! or the workload times out, `write KEY VALUE`, `read KEY`, `mkdir DIRECTORY`, `touch PATH`, which +//! creates the host file `PATH`, as a workload that writes a read-write share does, `env`, +//! `printenv NAME`, `pwd`, `timeout`, `fail`, and `launchfail`. Like `sh`, a script exits with the +//! status of its last command; `printenv NAME` prints nothing and fails with status 1 when `NAME` +//! is unset. //! Values written with `write` and directories made with `mkdir` live in memory until the VM //! stops, like files in the guest's RAM root file system. A `CANCEL` request ends a sleeping //! workload with the cancelled outcome, and a client that disconnects during an exec abandons it, @@ -30,6 +32,18 @@ //! made with `mkdir`), or that is a mapped regular file, with a diagnostic and the `cwd-failed` //! category. //! +//! Like the guest agent, the fake serves NVXC versions 1 and 2, and answers each request in its +//! version. A workload of version 1 reads no standard input, so its `cat` copies nothing, and +//! ends with `output-limit` once its output would pass 1 MiB. A workload of version 2 runs alone, +//! and another `EXEC` gets `BUSY`, as other requests do while it runs; its output has no cap; its +//! standard input is what the host's `STDIN` frames sent, which the fake acknowledges as the +//! workload reads it, until `STDIN_END`, and of which more than the window unacknowledged, or +//! more after its end, ends the session; and a `STOP` while it sleeps or reads its input ends it +//! with the cancelled outcome before the fake acknowledges the `STOP`, unless `fake_ignore_stop=1` +//! ignores it, as between workloads; with `fake_drop_stop=1`, the fake closes the connection after +//! that `EXIT`, without acknowledging the `STOP`. Each refusal while a workload runs takes the +//! version of the request that it refuses. +//! //! Host paths are mapped as OpenVMM and the guest agent map them: the command line exports //! numbered `--mount-child` directories, and files with the `file` flag, through //! `--mount-aggregate`, the kernel command line announces the number of mappings with @@ -192,12 +206,13 @@ use std::time::{Duration, Instant}; use crate::protocol::nvxs::ErrorCode; use crate::protocol::{ - APP_CANCEL, APP_ERROR, APP_EXEC, APP_EXIT, APP_FEATURES, APP_MAPS, APP_PING, APP_READY, - APP_STDERR, APP_STDOUT, APP_STOP, APP_STOPPED, BUSY, CAPABILITY_LEN, CWD_FAILED, ExitCategory, - GuestFeatures, INVALID_REQUEST, LAUNCH_FAILED, MAPPING_FAILED, MAPPINGS_INCOMPLETE, MAPS_TOKEN, - MAX_HOST_MAPPINGS, MAX_OUTPUT_BYTES, MapsPayload, OUTER_DATA, OUTER_ERROR, OUTER_HEADER_LEN, - OUTER_HOST_ATTACH, OUTER_READY, OUTER_WAIT, OuterHeader, UNSUPPORTED_OPERATION, decode_app, - decode_exec_payload, decode_maps_payload, decode_outer_header, encode_app, encode_outer, + APP_CANCEL, APP_ERROR, APP_EXEC, APP_EXIT, APP_FEATURES, APP_HEADER_LEN, APP_MAPS, APP_PING, + APP_READY, APP_STDERR, APP_STDOUT, APP_STOP, APP_STOPPED, BUSY, CAPABILITY_LEN, CWD_FAILED, + ExitCategory, GuestFeatures, INVALID_REQUEST, LAUNCH_FAILED, MAPPING_FAILED, + MAPPINGS_INCOMPLETE, MAPS_TOKEN, MAX_HOST_MAPPINGS, MAX_OUTPUT_BYTES, MapsPayload, OUTER_DATA, + OUTER_ERROR, OUTER_HEADER_LEN, OUTER_HOST_ATTACH, OUTER_READY, OUTER_WAIT, OuterHeader, + UNSUPPORTED_OPERATION, decode_app, decode_exec_payload, decode_maps_payload, + decode_outer_header, encode_app, encode_outer, v2, }; /// Largest workload-output frame emitted by the fake guest. @@ -2555,12 +2570,22 @@ enum Flow { enum Interrupt { /// No control request is waiting. None, - /// The workload's matching cancellation request arrived. + /// The workload's matching cancellation request, or a `STOP` of version 2, arrived. Cancel, /// The host disconnected and abandoned the workload. Disconnected, } +/// A host request that a DATA record carries. +enum Request { + /// A request: whether it is of NVXC version 2, its kind, whose code version 2 shares with + /// version 1, its ID, and its payload. + App(bool, u8, u64, Vec), + /// Version 2's standard input of the workload with this ID: a `STDIN` frame's bytes, or + /// `None` for its `STDIN_END`. + Stdin(u64, Option>), +} + /// A control stream that can report whether input is waiting without consuming it. trait Pending { /// Returns whether a read would make progress, including at end of stream. @@ -2643,26 +2668,67 @@ struct Session<'a, S> { host_sequence: u64, /// Whether matching cancellation requests are ignored for fault injection. ignore_cancel: bool, + /// Whether stop requests are ignored for fault injection, also while a workload runs. + ignore_stop: bool, + /// Whether the request that the fake serves, or the workload that runs, is of NVXC version + /// 2, whose frames its answers take. + v2: bool, + /// What the host sent of the running workload's standard input in version 2 and the workload + /// has not read, which the fake acknowledges as the workload reads it. + stdin: Vec, + /// Whether the running workload's standard input ended. + stdin_ended: bool, + /// The request ID of a `STOP` of version 2 that ended the running workload. + stop: Option, } impl Session<'_, S> { - /// Encodes and sends one sequenced guest application frame. + /// Encodes and sends one sequenced guest application frame, in the session's version. fn send(&mut self, kind: u8, request_id: u64, status: i32, payload: &[u8]) -> io::Result<()> { - let frame = encode_app(kind, request_id, status, payload); + if !self.v2 { + return self.send_frame(&encode_app(kind, request_id, status, payload)); + } + let frame = match kind { + APP_READY => v2::Frame::Ready { + version: v2::VERSION, + activation: payload, + }, + // Version 2 carries no empty chunk of output. + APP_STDOUT | APP_STDERR if payload.is_empty() => return Ok(()), + APP_STDOUT => v2::Frame::Stdout(payload), + APP_STDERR => v2::Frame::Stderr(payload), + APP_STOPPED => v2::Frame::Stopped, + APP_ERROR if payload == BUSY.as_bytes() => v2::Frame::Busy, + _ => v2::Frame::Error { + status, + category: payload, + }, + }; + self.send_v2(request_id, frame) + } + + /// Encodes and sends one frame of version 2. + fn send_v2(&mut self, request_id: u64, frame: v2::Frame<'_>) -> io::Result<()> { + let frame = frame.encode(request_id).map_err(io::Error::other)?; + self.send_frame(&frame) + } + + /// Sends one sequenced DATA record that carries `frame`. + fn send_frame(&mut self, frame: &[u8]) -> io::Result<()> { let record = encode_outer( OUTER_DATA, &self.instance, self.epoch, self.guest_sequence, - &frame, + frame, ) .map_err(io::Error::other)?; self.guest_sequence += 1; self.stream.write_all(&record) } - /// Reads the next host request, returning `None` at end of stream. - fn request(&mut self) -> io::Result)>> { + /// Reads the next host request, of either version, returning `None` at end of stream. + fn request(&mut self) -> io::Result> { let Some((header, frame)) = read_record(self.stream)? else { return Ok(None); }; @@ -2670,20 +2736,39 @@ impl Session<'_, S> { && header.instance_id == self.instance && header.epoch == self.epoch && header.sequence == self.host_sequence; - let frame = decode_app(&frame) - .ok() + let request = if frame.get(4) == Some(&v2::VERSION) { + v2::Frame::decode(&frame) + .ok() + .map(|(request_id, decoded)| match decoded { + v2::Frame::Stdin(bytes) => Request::Stdin(request_id, Some(bytes.to_vec())), + v2::Frame::StdinEnd => Request::Stdin(request_id, None), + // Version 2 keeps version 1's header and the codes of the kinds that both + // have. + _ => Request::App(true, frame[5], request_id, frame[APP_HEADER_LEN..].to_vec()), + }) + } else { + decode_app(&frame).ok().map(|frame| { + Request::App(false, frame.kind, frame.request_id, frame.payload.to_vec()) + }) + }; + let request = request .filter(|_| valid) .ok_or_else(|| io::Error::other("invalid data record"))?; self.host_sequence += 1; - Ok(Some((frame.kind, frame.request_id, frame.payload.to_vec()))) + Ok(Some(request)) } /// Serves requests until the host disconnects or asks the VM to stop. fn run(&mut self, options: &Options, guest: &mut Guest) -> io::Result { loop { - let Some((kind, request_id, payload)) = self.request()? else { + let Some(request) = self.request()? else { return Ok(Flow::Continue); }; + // The standard input of a workload that ended takes no more bytes. + let Request::App(v2, kind, request_id, payload) = request else { + continue; + }; + self.v2 = v2; match kind { APP_PING => self.send(APP_READY, request_id, 0, &[])?, APP_FEATURES if !options.legacy_guest => { @@ -2696,6 +2781,13 @@ impl Session<'_, S> { if !self.exec(request_id, &payload, guest)? { return Ok(Flow::Continue); } + if let Some(stop) = self.stop.take() { + if options.drop_stop { + return Ok(Flow::Drop); + } + let _ = self.send(APP_STOPPED, stop, 0, &[]); + return Ok(Flow::Exit); + } } APP_MAPS => match guest.map(options, &payload) { Ok(()) => self.send(APP_READY, request_id, 0, &[])?, @@ -2716,16 +2808,41 @@ impl Session<'_, S> { } } - /// Handles control traffic that arrives while workload `request_id` runs. + /// Handles control traffic that arrives while workload `request_id` runs. The workload's + /// standard input of version 2 queues for it, under the window; bytes beyond it, or after its + /// end, fail the session. fn interrupt(&mut self, request_id: u64) -> io::Result { while self.stream.pending()? { match self.request()? { None => return Ok(Interrupt::Disconnected), - Some((APP_CANCEL, id, _)) if id == request_id && !self.ignore_cancel => { + Some(Request::Stdin(id, _)) if id != request_id => {} + Some(Request::Stdin(_, _)) if self.stdin_ended => { + return Err(io::Error::other("standard input after its end")); + } + Some(Request::Stdin(_, None)) => self.stdin_ended = true, + Some(Request::Stdin(_, Some(bytes))) => { + if self.stdin.len() + bytes.len() > v2::STDIN_WINDOW as usize { + return Err(io::Error::other("standard input beyond the window")); + } + self.stdin.extend(bytes); + } + Some(Request::App(true, APP_STOP, _, _)) if self.v2 && self.ignore_stop => {} + Some(Request::App(true, APP_STOP, id, _)) if self.v2 => { + self.stop = Some(id); + return Ok(Interrupt::Cancel); + } + Some(Request::App(_, APP_CANCEL, id, _)) + if id == request_id && !self.ignore_cancel => + { return Ok(Interrupt::Cancel); } - Some((APP_CANCEL, _, _)) => {} - Some((_, id, _)) => self.send(APP_ERROR, id, 16, BUSY.as_bytes())?, + Some(Request::App(_, APP_CANCEL, _, _)) => {} + // The refusal takes the request's version, and the workload's frames their own. + Some(Request::App(v2, _, id, _)) => { + let running = std::mem::replace(&mut self.v2, v2); + self.send(APP_ERROR, id, 16, BUSY.as_bytes())?; + self.v2 = running; + } } } Ok(Interrupt::None) @@ -2747,6 +2864,9 @@ impl Session<'_, S> { let Ok(exec) = decode_exec_payload(payload) else { return self.send_some(APP_ERROR, request_id, 22, INVALID_REQUEST.as_bytes()); }; + // A workload of version 1 has no standard input, as one that reads the null device. + self.stdin.clear(); + self.stdin_ended = !self.v2; let mut environment = if exec.inherit_default_env { guest.default_environment.clone() } else { @@ -2868,6 +2988,27 @@ impl Session<'_, S> { remaining -= chunk; } } + "cat" => loop { + let chunk = self.stdin.len().min(OUTPUT_CHUNK_BYTES); + if limit.is_some_and(|limit| Instant::now() >= limit) { + return self.exit(request_id, 124, ExitCategory::Timeout); + } else if chunk > 0 { + // The workload reads these bytes now, so the fake acknowledges them. + let bytes: Vec = self.stdin.drain(..chunk).collect(); + self.send_v2(request_id, v2::Frame::StdinAck(chunk as u32))?; + self.send(APP_STDOUT, request_id, 0, &bytes)?; + } else if self.stdin_ended { + break; + } else { + match self.interrupt(request_id)? { + Interrupt::None => thread::sleep(Duration::from_millis(5)), + Interrupt::Cancel => { + return self.exit(request_id, 137, ExitCategory::Cancelled); + } + Interrupt::Disconnected => return Ok(None), + } + } + }, "write" => { let (key, value) = argument.split_once(' ').unwrap_or((argument, "")); guest.values.insert(key.to_owned(), value.to_owned()); @@ -2947,11 +3088,24 @@ impl Session<'_, S> { status: i32, category: ExitCategory, ) -> io::Result> { + if self.v2 { + let outcome = match category { + ExitCategory::Exit => v2::Outcome::Exit, + ExitCategory::Signal => v2::Outcome::Signal, + ExitCategory::Timeout => v2::Outcome::Timeout, + ExitCategory::Cancelled => v2::Outcome::Cancelled, + ExitCategory::Failed | ExitCategory::OutputLimit => v2::Outcome::Failed, + }; + return self + .send_v2(request_id, v2::Frame::Exit { status, outcome }) + .map(Some); + } self.send_some(APP_EXIT, request_id, status, category.as_str().as_bytes()) } - /// Forwards workload output unless it would exceed the guest agent's output limit, and - /// returns whether it did. A workload whose output was refused ends with `output-limit`. + /// Forwards workload output unless it would exceed the guest agent's output limit of version + /// 1, and returns whether it did. A workload whose output was refused ends with + /// `output-limit`. fn forward( &mut self, kind: u8, @@ -2959,7 +3113,7 @@ impl Session<'_, S> { forwarded: &mut usize, output: &[u8], ) -> io::Result { - if *forwarded + output.len() > MAX_OUTPUT_BYTES { + if !self.v2 && *forwarded + output.len() > MAX_OUTPUT_BYTES { return Ok(false); } *forwarded += output.len(); @@ -3020,6 +3174,11 @@ fn serve_client( guest_sequence: 1, host_sequence: 0, ignore_cancel: options.ignore_cancel, + ignore_stop: options.ignore_stop, + v2: false, + stdin: Vec::new(), + stdin_ended: true, + stop: None, }; match session.run(options, guest) { Ok(Flow::Exit) => { diff --git a/nvx-rs/src/testing/fake_openvmm/tests.rs b/nvx-rs/src/testing/fake_openvmm/tests.rs index ea78459ad..d03852068 100644 --- a/nvx-rs/src/testing/fake_openvmm/tests.rs +++ b/nvx-rs/src/testing/fake_openvmm/tests.rs @@ -1108,3 +1108,221 @@ fn capture_scenarios_take_defaults_and_refuse_unknown_fields() { ); } } + +/// A frame of version 2 that the fake guest sent, with its request ID. +#[cfg(all(unix, feature = "cli"))] +#[derive(Debug, PartialEq, Eq)] +enum Reply { + Ready(u64, Vec), + Stdout(u64, Vec), + Exit(u64, i32, v2::Outcome), + Stopped(u64), + Ack(u64, u32), + Busy(u64), + Error(u64, i32, Vec), +} + +/// A host attached to the control session of a fake guest of a managed run, which a thread +/// serves over a socket pair, as the fake serves a client of its control console, and which +/// returns whether the session stopped the VM. +#[cfg(all(unix, feature = "cli"))] +struct Host { + stream: std::os::unix::net::UnixStream, + instance: [u8; 16], + epoch: u64, + sequence: u64, + fake: thread::JoinHandle, + _directory: tempfile::TempDir, +} + +#[cfg(all(unix, feature = "cli"))] +impl Host { + /// Attaches to a new fake guest. + fn attach() -> Self { + let managed = boot_with( + "nvx_mode=managed nvx_identity=1000:1001", + &[ + "--virtio-console", + "none", + "--microvm-control-console", + "listen={dir}/control.sock", + "--microvm-control-auth-stdin", + "--microvm-report", + "{dir}/outcome.json", + ], + ); + let (directory, options) = parse_run_in(&managed, private_directory); + let options = options.unwrap(); + let (mut stream, mut fake) = std::os::unix::net::UnixStream::pair().unwrap(); + let capability = [7; CAPABILITY_LEN]; + let fake = thread::spawn(move || { + let mut guest = Guest::new(&options); + let flow = serve_client(&mut fake, &options, &capability, 1, &mut guest); + matches!(flow, Flow::Exit) + }); + let attach = encode_outer(OUTER_HOST_ATTACH, &[0; 16], 0, 0, &capability).unwrap(); + stream.write_all(&attach).unwrap(); + let (ready, _) = read_record(&mut stream).unwrap().unwrap(); + assert_eq!(ready.record_type, OUTER_READY); + Self { + stream, + instance: ready.instance_id, + epoch: ready.epoch, + sequence: 0, + fake, + _directory: directory, + } + } + + /// Sends one frame in a DATA record. + fn send_frame(&mut self, frame: &[u8]) { + let record = encode_outer(OUTER_DATA, &self.instance, self.epoch, self.sequence, frame); + self.stream.write_all(&record.unwrap()).unwrap(); + self.sequence += 1; + } + + /// Sends a request of version 2. + fn send(&mut self, request_id: u64, frame: v2::Frame<'_>) { + self.send_frame(&frame.encode(request_id).unwrap()); + } + + /// Reads the payload of the fake's next DATA record. + fn read(&mut self) -> Vec { + let (header, payload) = read_record(&mut self.stream).unwrap().unwrap(); + assert_eq!(header.record_type, OUTER_DATA); + payload + } + + /// Reads the fake's next frame, of version 2. + fn next(&mut self) -> Reply { + let payload = self.read(); + let (id, frame) = v2::Frame::decode(&payload).unwrap(); + match frame { + v2::Frame::Ready { + version, + activation, + } => { + assert_eq!(version, v2::VERSION); + Reply::Ready(id, activation.to_vec()) + } + v2::Frame::Stdout(bytes) => Reply::Stdout(id, bytes.to_vec()), + v2::Frame::Exit { status, outcome } => Reply::Exit(id, status, outcome), + v2::Frame::Stopped => Reply::Stopped(id), + v2::Frame::StdinAck(count) => Reply::Ack(id, count), + v2::Frame::Busy => Reply::Busy(id), + v2::Frame::Error { status, category } => Reply::Error(id, status, category.to_vec()), + frame => panic!("unexpected {frame:?}"), + } + } + + /// Stops the VM over version 2, which ends the fake's session. + fn stop(mut self, request_id: u64) { + self.send(request_id, v2::Frame::Stop); + assert_eq!(self.next(), Reply::Stopped(request_id)); + assert!(self.fake.join().unwrap()); + } +} + +/// Returns the payload of an `EXEC` request of the fake's workload `script`. +#[cfg(all(unix, feature = "cli"))] +fn script(script: &str) -> Vec { + timed(script, 0) +} + +/// Returns the payload of an `EXEC` request of the fake's workload `script`, which times out +/// after `timeout_ms`, unless it is 0. +#[cfg(all(unix, feature = "cli"))] +fn timed(script: &str, timeout_ms: u32) -> Vec { + crate::protocol::encode_exec_payload(&crate::protocol::Workload { + argv: ["/bin/sh", "-c", script].map(str::to_owned).to_vec(), + timeout_ms, + cwd: None, + environment: crate::protocol::WorkloadEnvironment::Default, + }) + .unwrap() +} + +#[cfg(all(unix, feature = "cli"))] +#[test] +fn version_2_workloads_read_their_input_until_its_end_under_the_window() { + let mut host = Host::attach(); + host.send(1, v2::Frame::Ping); + assert_eq!(host.next(), Reply::Ready(1, Vec::new())); + // The fake acknowledges what the workload reads, and `cat` reads until the input ends. + host.send(2, v2::Frame::Exec(&script("cat; echo done"))); + for input in [b"hello ", b"world\n"] { + host.send(2, v2::Frame::Stdin(input)); + assert_eq!(host.next(), Reply::Ack(2, 6)); + assert_eq!(host.next(), Reply::Stdout(2, input.to_vec())); + } + host.send(2, v2::Frame::StdinEnd); + assert_eq!(host.next(), Reply::Stdout(2, b"done\n".to_vec())); + assert_eq!(host.next(), Reply::Exit(2, 0, v2::Outcome::Exit)); + // A workload that waits for its input still ends at its timeout. + host.send(3, v2::Frame::Exec(&timed("cat", 50))); + assert_eq!(host.next(), Reply::Exit(3, 124, v2::Outcome::Timeout)); + // Input that the workload has not read stays unacknowledged, up to the window, and a byte + // beyond it ends the session without an answer. + host.send(4, v2::Frame::Exec(&script("sleep 60000; cat"))); + let half = vec![b'x'; v2::STDIN_WINDOW as usize / 2]; + host.send(4, v2::Frame::Stdin(&half)); + host.send(4, v2::Frame::Stdin(&half)); + host.send(4, v2::Frame::Stdin(b"x")); + assert!(read_record(&mut host.stream).unwrap().is_none()); + assert!(!host.fake.join().unwrap()); +} + +#[cfg(all(unix, feature = "cli"))] +#[test] +fn version_2_workloads_send_output_past_the_cap_of_version_1() { + let mut host = Host::attach(); + let total = MAX_OUTPUT_BYTES * 3 / 2; + host.send(1, v2::Frame::Exec(&script(&format!("flood {total}")))); + let mut forwarded = 0; + loop { + match host.next() { + Reply::Stdout(1, bytes) => forwarded += bytes.len(), + Reply::Exit(1, 0, v2::Outcome::Exit) => break, + reply => panic!("unexpected {reply:?}"), + } + } + assert_eq!(forwarded, total); + // Version 1, which the same session still serves, caps it. + host.send_frame(&encode_app( + APP_EXEC, + 2, + 0, + &script(&format!("flood {total}")), + )); + let mut capped = 0; + let exit = loop { + let payload = host.read(); + let frame = decode_app(&payload).unwrap(); + match frame.kind { + APP_STDOUT => capped += frame.payload.len(), + _ => break (frame.kind, frame.status, frame.payload.to_vec()), + } + }; + assert!(capped <= MAX_OUTPUT_BYTES); + assert_eq!(exit, (APP_EXIT, 125, b"output-limit".to_vec())); + host.stop(3); +} + +#[cfg(all(unix, feature = "cli"))] +#[test] +fn version_2_workloads_run_alone_and_end_on_cancel_or_stop() { + let mut host = Host::attach(); + host.send(1, v2::Frame::Exec(&script("sleep 60000"))); + host.send(2, v2::Frame::Exec(&script("echo second"))); + assert_eq!(host.next(), Reply::Busy(2)); + // The cancellation of another request leaves the workload running. + host.send(3, v2::Frame::Cancel); + host.send(1, v2::Frame::Cancel); + assert_eq!(host.next(), Reply::Exit(1, 137, v2::Outcome::Cancelled)); + // `STOP` ends the running workload, which reports it, and then the VM. + host.send(4, v2::Frame::Exec(&script("sleep 60000"))); + host.send(5, v2::Frame::Stop); + assert_eq!(host.next(), Reply::Exit(4, 137, v2::Outcome::Cancelled)); + assert_eq!(host.next(), Reply::Stopped(5)); + assert!(host.fake.join().unwrap()); +} diff --git a/scripts/test-layout-allowlist.json b/scripts/test-layout-allowlist.json index 1178a893f..ddb768d07 100644 --- a/scripts/test-layout-allowlist.json +++ b/scripts/test-layout-allowlist.json @@ -17,7 +17,6 @@ "nvx-rs/src/host/platform/windows/process.rs": 2, "nvx-rs/src/host/platform/windows/spawn.rs": 1, "nvx-rs/src/host/session.rs": 5, - "nvx-rs/src/host/vm.rs": 2, - "nvx-rs/src/protocol/app.rs": 1 + "nvx-rs/src/host/vm.rs": 2 } }