diff --git a/nvx-rs/README.md b/nvx-rs/README.md index c64299109..824db9f77 100644 --- a/nvx-rs/README.md +++ b/nvx-rs/README.md @@ -67,7 +67,7 @@ nvx-rs/ | `nvx::testing::fake_openvmm` | A fake OpenVMM that serves the control console and runs scripted workloads without a VM, for direct launches and for the managed and one-shot sandboxes of `nvx-rs sandbox` and `nvx.py sandbox` | | `nvx::host::cli` | Feature `cli`: the `nvx-rs` executable; see [The nvx-rs CLI](#the-nvx-rs-cli) | | `nvx::guest` | The programs that run in Linux guests; see [Guest binaries](#guest-binaries) | -| `nvx::guest::nvxs` | `GuestConnection`: the guest's role of the NVXS session over a byte stream, such as the control console, for guest agents. It never reads past the record in progress, so a caller that polls the stream between records sees every record that it has not returned | +| `nvx::guest::nvxs` | `GuestConnection`: the guest's role of the NVXS session over a byte stream, such as the control console, on which NVX's managed agent and other guest agents run their sessions. It never reads past the record in progress, so a caller that polls the stream between records sees every record that it has not returned | | `nvx::guest::spawn` | `Command`: starts a program in a new process for any guest init or agent, with its arguments, environment, root and working directories, standard streams, inherited descriptors, cgroup, and process group, without NVX's control protocol, managed agent, or workload identity. Its `LaunchReport` tells whether the program started, or which `LaunchStage` failed and why. NVX's managed agent is one caller | Lowering functions take the names by which messages refer to their inputs, such diff --git a/nvx-rs/src/guest/managed_agent/dispatch.rs b/nvx-rs/src/guest/managed_agent/dispatch.rs index 453706808..60f090177 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch.rs @@ -1,14 +1,13 @@ use crate::guest::sys::Errno; use crate::protocol::{ self, APP_CANCEL, APP_EXEC, APP_FEATURES, APP_MAPS, APP_PING, APP_READY, APP_STOP, APP_STOPPED, - GuestFeatures, INVALID_REQUEST, MAPPINGS_INCOMPLETE, OUTER_DATA, OUTER_RESET, - UNSUPPORTED_OPERATION, + GuestFeatures, INVALID_REQUEST, MAPPINGS_INCOMPLETE, UNSUPPORTED_OPERATION, }; use super::diagnostics; use super::mappings::{HostMappings, MappingError}; use super::seam::System; -use super::session::{Record, Session}; +use super::session::{Event, ReceiveFailure, Session}; use super::{AgentConfig, exec, tty}; /// Terminal outcome of the long-lived dispatch loop. @@ -27,43 +26,66 @@ pub(crate) fn run( ) -> Result { session.send_attach()?; loop { - let record = session.read_record().inspect_err(|error| { - diagnostics::error(system, "outer-read", *error as i32); - })?; - match record.header.record_type { - OUTER_RESET => session.acknowledge_reset(&record).inspect_err(|error| { - diagnostics::error(system, "reset-ack", *error as i32); - })?, - // As in the C agent, a DATA record that fails also reports the `data` stage, after any - // more specific stage. Its status is the error that ends the session, which - // `control-session` reports too: `EPROTO` where C's `errno` was incidental, as after - // a validation failure. - OUTER_DATA => match handle_data(system, session, config, mappings, &record) { - Ok(Some(outcome)) => return Ok(outcome), - Ok(None) => {} - Err(error) => { - diagnostics::error(system, "data", error as i32); - return Err(error); - } - }, - record_type => { - diagnostics::error(system, "outer-type", i32::from(record_type)); - return Err(Errno::EPROTO); + let payload = match session.next_event() { + Ok(Event::Reset) => { + session.acknowledge().inspect_err(|error| { + diagnostics::error(system, "reset-ack", *error as i32); + })?; + continue; + } + Ok(Event::Data(payload)) => payload, + Err(failure) => return Err(report_failure(system, failure)), + }; + // As in the C agent, a DATA record that fails also reports the `data` stage, after any + // more specific stage. Its status is the error that ends the session, which + // `control-session` reports too: `EPROTO` where C's `errno` was incidental, as after a + // validation failure. + match handle_data(system, session, config, mappings, &payload) { + Ok(Some(outcome)) => return Ok(outcome), + Ok(None) => {} + Err(error) => { + diagnostics::error(system, "data", error as i32); + return Err(error); } } } } -/// Validates one DATA record and serves its application request, returning the outcome that -/// ends the session, if any. +/// Reports a record from the broker that ended the session at the C agent's stage, and returns +/// the error that ends the session. +fn report_failure(system: &mut impl System, failure: ReceiveFailure) -> Errno { + match failure { + ReceiveFailure::Read(error) => { + diagnostics::error(system, "outer-read", error as i32); + error + } + ReceiveFailure::Type(code) => { + diagnostics::error(system, "outer-type", i32::from(code)); + Errno::EPROTO + } + ReceiveFailure::Reset => { + diagnostics::error(system, "reset-ack", Errno::EPROTO as i32); + Errno::EPROTO + } + ReceiveFailure::Data(failure) => { + diagnostics::error(system, failure.kind.stage(), failure.status); + let error = failure.into_error(); + diagnostics::error(system, "data", error as i32); + error + } + } +} + +/// Decodes the NVXC request of one DATA payload and serves it, returning the outcome that ends +/// the session, if any. fn handle_data( system: &mut impl System, session: &mut Session, config: &AgentConfig, mappings: &mut HostMappings, - record: &Record, + payload: &[u8], ) -> Result, Errno> { - let request = match session.validate_data(record) { + let request = match session.accept_request(payload) { Ok(request) => request, Err(failure) => { diagnostics::error(system, failure.kind.stage(), failure.status); @@ -156,10 +178,11 @@ mod tests { use std::thread; use crate::protocol::{ - APP_ERROR, APP_EXIT, APP_STDERR, APP_STDOUT, OUTER_ACK, OUTER_CREDIT, OUTER_GUEST_ATTACH, - OUTER_HEADER_LEN, OUTER_RECEIVE_WINDOW, OuterHeader, ProtocolError, Workload, - WorkloadEnvironment, decode_ack_payload, decode_app, decode_credit_payload, - decode_outer_header, encode_app, encode_exec_payload, encode_outer, + APP_ERROR, APP_EXIT, APP_STDERR, APP_STDOUT, OUTER_ACK, OUTER_CREDIT, OUTER_DATA, + OUTER_GUEST_ATTACH, OUTER_HEADER_LEN, OUTER_MAX_PAYLOAD, OUTER_RECEIVE_WINDOW, OUTER_RESET, + OUTER_WAIT, OuterHeader, ProtocolError, Workload, WorkloadEnvironment, decode_ack_payload, + decode_app, decode_credit_payload, decode_outer_header, encode_app, encode_exec_payload, + encode_outer, }; use super::super::seam::{Paths, TestSystem}; @@ -182,6 +205,7 @@ mod tests { struct Harness { host: UnixStream, thread: thread::JoinHandle<(Result, TestSystem)>, + epoch: u64, host_sequence: u64, guest_sequence: u64, } @@ -191,7 +215,14 @@ mod tests { Self::with_system(direct, TestSystem::new(Paths::default()), 0) } - fn with_system(direct: bool, mut system: TestSystem, expected_mappings: u32) -> Self { + fn with_system(direct: bool, system: TestSystem, expected_mappings: u32) -> Self { + let mut harness = Self::spawn(direct, system, expected_mappings); + harness.reset(1, 7); + harness + } + + /// Starts the agent, which attaches and waits for the broker's first RESET. + fn spawn(direct: bool, mut system: TestSystem, expected_mappings: u32) -> Self { let (host, guest) = UnixStream::pair().unwrap(); let descriptor = FileDescriptor::from_owned(guest.into()); let thread = thread::spawn(move || { @@ -215,20 +246,30 @@ mod tests { let mut harness = Self { host, thread, - host_sequence: 8, - guest_sequence: 1, + epoch: 0, + host_sequence: 0, + guest_sequence: 0, }; let (header, payload) = harness.read_outer(); assert_eq!(header.record_type, OUTER_GUEST_ATTACH); assert!(payload.is_empty()); harness - .host - .write_all(&encode_outer(OUTER_RESET, &INSTANCE, 1, 7, &[]).unwrap()) + } + + /// Resets the session to `epoch` with a RESET at `sequence`, which the agent acknowledges. + fn reset(&mut self, epoch: u64, sequence: u64) { + self.host + .write_all(&encode_outer(OUTER_RESET, &INSTANCE, epoch, sequence, &[]).unwrap()) .unwrap(); - let (header, payload) = harness.read_outer(); - assert_eq!((header.record_type, header.sequence), (OUTER_ACK, 0)); + let (header, payload) = self.read_outer(); + assert_eq!( + (header.record_type, header.epoch, header.sequence), + (OUTER_ACK, epoch, 0) + ); assert_eq!(decode_ack_payload(&payload).unwrap(), OUTER_RECEIVE_WINDOW); - harness + self.epoch = epoch; + self.host_sequence = sequence + 1; + self.guest_sequence = 1; } fn request(&mut self, kind: u8, payload: &[u8]) -> Response { @@ -241,14 +282,21 @@ mod tests { let frame = encode_app(kind, request_id, 0, payload); self.host .write_all( - &encode_outer(OUTER_DATA, &INSTANCE, 1, self.host_sequence, &frame).unwrap(), + &encode_outer( + OUTER_DATA, + &INSTANCE, + self.epoch, + self.host_sequence, + &frame, + ) + .unwrap(), ) .unwrap(); self.host_sequence += 1; let (credit, payload) = self.read_outer(); assert_eq!( - (credit.record_type, credit.sequence), - (OUTER_CREDIT, self.guest_sequence) + (credit.record_type, credit.epoch, credit.sequence), + (OUTER_CREDIT, self.epoch, self.guest_sequence) ); assert_eq!(decode_credit_payload(&payload).unwrap(), frame.len() as u32); self.guest_sequence += 1; @@ -297,6 +345,15 @@ mod tests { system.diagnostics } + /// Ends the stream after `partial`, a part of a record or nothing. + fn close(mut self, partial: &[u8]) -> Vec { + self.host.write_all(partial).unwrap(); + self.host.shutdown(std::net::Shutdown::Write).unwrap(); + let (outcome, system) = self.thread.join().unwrap(); + assert_eq!(outcome.err(), Some(Errno::EIO)); + system.diagnostics + } + fn read_outer(&mut self) -> (OuterHeader, Vec) { let mut header = [0u8; OUTER_HEADER_LEN]; self.host.read_exact(&mut header).unwrap(); @@ -961,4 +1018,144 @@ mod tests { assert!(system.diagnostics.is_empty()); Ok(()) } + + #[test] + fn refused_records_report_the_stage_at_which_the_c_agent_reported_them() { + let line = |stage: &str, status: i32| { + format!("NVX-MANAGED-ERROR: stage={stage} status={status}\n") + }; + let eproto = Errno::EPROTO as i32; + let outer = |record_type, instance: &[u8; 16], epoch, sequence, payload: &[u8]| { + encode_outer(record_type, instance, epoch, sequence, payload).unwrap() + }; + let ping = encode_app(APP_PING, 42, 0, &[]); + let mut bad_magic = outer(OUTER_DATA, &INSTANCE, 1, 8, &ping); + bad_magic[0] = b'X'; + let mut oversized = outer(OUTER_DATA, &INSTANCE, 1, 8, &[]); + oversized[40..44].copy_from_slice(&(OUTER_MAX_PAYLOAD as u32 + 1).to_le_bytes()); + let window = OUTER_RECEIVE_WINDOW.to_le_bytes(); + let cases = [ + // Headers that the C agent refused. + (bad_magic, line("outer-read", eproto)), + (oversized, line("outer-read", eproto)), + // Types that it does not take, whatever their payloads. + (outer(0, &INSTANCE, 1, 8, &[]), line("outer-type", 0)), + (outer(10, &INSTANCE, 1, 8, b"abc"), line("outer-type", 10)), + ( + outer(OUTER_ACK, &INSTANCE, 1, 8, &window), + line("outer-type", i32::from(OUTER_ACK)), + ), + ( + outer(OUTER_CREDIT, &INSTANCE, 1, 8, &[0; 4]), + line("outer-type", i32::from(OUTER_CREDIT)), + ), + ( + outer(OUTER_WAIT, &INSTANCE, 1, 8, b"x"), + line("outer-type", i32::from(OUTER_WAIT)), + ), + ( + outer(OUTER_GUEST_ATTACH, &INSTANCE, 1, 8, &[]), + line("outer-type", i32::from(OUTER_GUEST_ATTACH)), + ), + // RESET records with a payload or without an epoch. + ( + outer(OUTER_RESET, &INSTANCE, 2, 0, b"x"), + line("reset-ack", eproto), + ), + ( + outer(OUTER_RESET, &INSTANCE, 0, 0, &[]), + line("reset-ack", eproto), + ), + // DATA records without an NVXC frame, or of another instance. + ( + outer(OUTER_DATA, &INSTANCE, 1, 8, &[]), + line("data-application", 0) + &line("data", eproto), + ), + ( + outer(OUTER_DATA, &[8; 16], 1, 8, &ping), + line("data-identity", 1) + &line("data", eproto), + ), + // Faults that only the shared session refuses: a RESET without an instance ID, or of an + // epoch that does not advance the instance's. + ( + outer(OUTER_RESET, &[0; 16], 2, 0, &[]), + line("reset-ack", eproto), + ), + ( + outer(OUTER_RESET, &INSTANCE, 1, 0, &[]), + line("reset-ack", eproto), + ), + ]; + for (record, expected) in cases { + let diagnostics = Harness::new(true).fail(&record); + let header = &record[..OUTER_HEADER_LEN]; + assert_eq!( + String::from_utf8(diagnostics).unwrap(), + expected, + "{header:?}" + ); + } + } + + #[test] + fn data_before_the_first_reset_is_a_type_that_the_agent_does_not_take_yet() { + let harness = Harness::spawn(true, TestSystem::new(Paths::default()), 0); + let frame = encode_app(APP_PING, 42, 0, &[]); + let diagnostics = harness.fail(&encode_outer(OUTER_DATA, &INSTANCE, 1, 0, &frame).unwrap()); + assert_eq!( + diagnostics, + format!("NVX-MANAGED-ERROR: stage=outer-type status={OUTER_DATA}\n").as_bytes() + ); + } + + #[test] + fn data_at_the_last_sequence_number_is_out_of_sequence() { + // After a RESET at the second-to-last sequence number, DATA at the last one leaves none + // for the broker's next record. + let mut harness = Harness::spawn(true, TestSystem::new(Paths::default()), 0); + harness.reset(1, u64::MAX - 1); + let frame = encode_app(APP_PING, 42, 0, &[]); + let record = encode_outer(OUTER_DATA, &INSTANCE, 1, u64::MAX, &frame).unwrap(); + assert_eq!( + String::from_utf8(harness.fail(&record)).unwrap(), + format!( + "NVX-MANAGED-ERROR: stage=data-sequence status=-1\n\ + NVX-MANAGED-ERROR: stage=data status={}\n", + Errno::EPROTO as i32 + ) + ); + } + + #[test] + fn a_stream_that_ends_reports_a_read_failure_inside_a_record_or_not() { + let expected = format!( + "NVX-MANAGED-ERROR: stage=outer-read status={}\n", + Errno::EIO as i32 + ); + let frame = encode_app(APP_PING, 42, 0, &[]); + let record = encode_outer(OUTER_DATA, &INSTANCE, 1, 8, &frame).unwrap(); + for length in [0, 20, OUTER_HEADER_LEN + 1] { + let diagnostics = Harness::new(true).close(&record[..length]); + assert_eq!(diagnostics, expected.as_bytes(), "after {length} bytes"); + } + } + + #[test] + fn a_reset_between_requests_starts_a_new_epoch() { + let mut harness = Harness::new(true); + assert_eq!( + harness.request(APP_PING, &[]), + response(APP_READY, 42, 0, b"") + ); + // A RESET of a later epoch may come at any sequence number. + for (epoch, sequence) in [(2, 0), (5, 11)] { + harness.reset(epoch, sequence); + assert_eq!( + harness.request(APP_PING, &[]), + response(APP_READY, 42, 0, b"") + ); + } + let system = harness.stop(); + assert!(system.diagnostics.is_empty()); + } } diff --git a/nvx-rs/src/guest/managed_agent/exec.rs b/nvx-rs/src/guest/managed_agent/exec.rs index 3d09a4eef..414a56373 100644 --- a/nvx-rs/src/guest/managed_agent/exec.rs +++ b/nvx-rs/src/guest/managed_agent/exec.rs @@ -6,7 +6,7 @@ use crate::guest::sys::{ }; use crate::protocol::{ self, APP_CANCEL, APP_EXIT, APP_STDERR, APP_STDOUT, BUSY, CWD_FAILED, GuestExecPayload, - LAUNCH_FAILED, OUTER_DATA, OUTER_RESET, + LAUNCH_FAILED, OUTER_DATA, }; use super::AgentConfig; @@ -15,7 +15,7 @@ use super::diagnostics::with_description; use super::exec_config; use super::launch::{self, LaunchRequest}; use super::seam::{System, path_c_string}; -use super::session::{DataFailureKind, Session}; +use super::session::{DataFailureKind, Event, ReceiveFailure, Session}; use super::tty; /// Maximum bytes read from one output descriptor at a time. @@ -365,26 +365,19 @@ fn handle_control( session: &mut Session, exec_id: u64, ) -> Result { - let record = session.read_record().inspect_err(|error| { - super::diagnostics::error(system, "exec-outer-read", *error as i32); - })?; - if record.header.record_type == OUTER_RESET { - session.acknowledge_reset(&record)?; - return Ok(ExecControl::SessionLost); - } - if record.header.record_type != OUTER_DATA { - super::diagnostics::error(system, "exec-record", i32::from(record.header.record_type)); - return Err(Errno::EPROTO); - } - let request = match session.validate_data(&record) { + let payload = match session.next_event() { + Ok(Event::Reset) => { + session.acknowledge()?; + return Ok(ExecControl::SessionLost); + } + Ok(Event::Data(payload)) => payload, + Err(failure) => return Err(report_control_failure(system, failure)), + }; + let request = match session.accept_request(&payload) { Ok(request) => request, Err(failure) => { if failure.kind != DataFailureKind::Credit { - super::diagnostics::error( - system, - "exec-record", - i32::from(record.header.record_type), - ); + super::diagnostics::error(system, "exec-record", i32::from(OUTER_DATA)); } return Err(failure.into_error()); } @@ -402,6 +395,27 @@ fn handle_control( Ok(ExecControl::None) } +/// Reports a control record that ended the session while a workload ran, at the C agent's stage, +/// and returns the error that ends the session. +fn report_control_failure(system: &mut impl System, failure: ReceiveFailure) -> Errno { + match failure { + ReceiveFailure::Read(error) => { + super::diagnostics::error(system, "exec-outer-read", error as i32); + error + } + ReceiveFailure::Type(code) => { + super::diagnostics::error(system, "exec-record", i32::from(code)); + Errno::EPROTO + } + // The C agent reported a RESET that it refused only as the error that ends the session. + ReceiveFailure::Reset => Errno::EPROTO, + ReceiveFailure::Data(failure) => { + super::diagnostics::error(system, "exec-record", i32::from(OUTER_DATA)); + failure.into_error() + } + } +} + /// Result of draining one nonblocking output descriptor. #[derive(Clone, Copy, PartialEq, Eq)] enum StreamResult { diff --git a/nvx-rs/src/guest/managed_agent/mod.rs b/nvx-rs/src/guest/managed_agent/mod.rs index 041745155..ae373c609 100644 --- a/nvx-rs/src/guest/managed_agent/mod.rs +++ b/nvx-rs/src/guest/managed_agent/mod.rs @@ -65,7 +65,7 @@ mod loopback; mod mappings; /// Replaceable privileged-system seam. mod seam; -/// Outer control-session sequencing and credit. +/// The control session: the shared NVXS session's guest role, and the NVXC frames that it carries. mod session; /// Control-terminal setup and drain behavior. mod tty; diff --git a/nvx-rs/src/guest/managed_agent/session.rs b/nvx-rs/src/guest/managed_agent/session.rs index aa014e11a..bbf3f834c 100644 --- a/nvx-rs/src/guest/managed_agent/session.rs +++ b/nvx-rs/src/guest/managed_agent/session.rs @@ -1,38 +1,51 @@ +use std::io; + +use crate::guest::nvxs::{GuestConnection, GuestConnectionError}; use crate::guest::sys::{Errno, FileDescriptor}; -use crate::protocol::{ - self, APP_ERROR, OUTER_ACK, OUTER_CREDIT, OUTER_DATA, OUTER_GUEST_ATTACH, OUTER_HEADER_LEN, - OUTER_MAX_PAYLOAD, OUTER_RECEIVE_WINDOW, OUTER_RESET, OuterHeader, -}; - -/// One decoded outer control record. -pub(crate) struct Record { - /// Validated outer header. - pub(crate) header: OuterHeader, - /// Record payload bytes. - pub(crate) payload: Vec, -} +use crate::protocol::nvxs::{GuestEvent, RecordError, RecordType, SessionError}; +use crate::protocol::{self, APP_ERROR, OUTER_MAX_PAYLOAD, OUTER_RECEIVE_WINDOW}; -/// Sequence, identity, and credit state of one control-console attachment. +/// 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. pub(crate) struct Session { - /// Owned control-console descriptor. - descriptor: FileDescriptor, - /// Attached instance capability identity. - pub(crate) instance_id: [u8; 16], - /// Current nonzero reset epoch. - pub(crate) epoch: u64, - /// Next sequence sent by the guest. - pub(crate) guest_sequence: u64, - /// Next sequence expected from the host. - pub(crate) host_sequence: u64, + /// The guest's NVXS connection on the owned control-console descriptor. + connection: GuestConnection, + /// Error of the connection's latest failure, which a connection that a failure ended reports + /// again. + last_error: Option, +} + +/// Record from the broker that the session accepted. +pub(crate) enum Event { + /// The broker reset the session. The agent abandons the earlier epoch's work, then + /// acknowledges the new epoch with [`Session::acknowledge`]. + Reset, + /// The payload of a DATA record: one NVXC frame, which [`Session::accept_request`] decodes. + Data(Vec), +} + +/// A record from the broker that ended the session, classified by the stage at which the C agent +/// reported it. +pub(crate) enum ReceiveFailure { + /// The stream failed or ended, or a record header was malformed, with the error to report. + Read(Errno), + /// A record of a type that the agent does not take in its state, with the type's code. + Type(u8), + /// A RESET that the session refused. + Reset, + /// A DATA record that the session refused. + Data(DataFailure), } /// Protocol stage that rejected an outer DATA record. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) enum DataFailureKind { - /// Record type or host sequence mismatch. + /// Host sequence mismatch. Sequence, /// Instance identity or epoch mismatch. Identity, + /// A payload beyond the receive window, which the C agent did not check. + Window, /// Malformed application frame or request metadata. Application, /// Failure to replenish host receive credit. @@ -45,6 +58,7 @@ impl DataFailureKind { match self { Self::Sequence => "data-sequence", Self::Identity => "data-identity", + Self::Window => "data-window", Self::Application => "data-application", Self::Credit => "data-credit", } @@ -72,72 +86,62 @@ impl Session { /// Starts unattached session state on an owned control descriptor. pub(crate) fn new(descriptor: FileDescriptor) -> Self { Self { - descriptor, - instance_id: [0; 16], - epoch: 0, - guest_sequence: 0, - host_sequence: 0, + connection: GuestConnection::new(descriptor), + last_error: None, } } - /// Borrows the control descriptor for polling and TTY operations. + /// Borrows the control descriptor for polling and TTY operations. The connection never reads + /// past the record in progress and holds no partial record between calls, so polling sees + /// every record that the session has not returned. pub(crate) fn descriptor(&self) -> &FileDescriptor { - &self.descriptor + self.connection.get_ref() } /// Sends the initial guest-attach record. pub(crate) fn send_attach(&mut self) -> Result<(), Errno> { - self.write_outer(OUTER_GUEST_ATTACH, &[0; 16], 0, 0, &[]) + let attached = self.connection.attach(); + attached.map_err(|error| self.error(error)) } - /// Validates and acknowledges a host RESET record. - pub(crate) fn acknowledge_reset(&mut self, record: &Record) -> Result<(), Errno> { - if record.header.record_type != OUTER_RESET - || !record.payload.is_empty() - || record.header.epoch == 0 - { - return Err(Errno::EPROTO); + /// Blocks until the broker's next record completes, and returns what the session accepted. + pub(crate) fn next_event(&mut self) -> Result { + match self.connection.next_event() { + Ok(GuestEvent::Reset(_)) => Ok(Event::Reset), + Ok(GuestEvent::Data(payload)) => Ok(Event::Data(payload)), + Err(error) => Err(self.receive_failure(error)), } - self.instance_id = record.header.instance_id; - self.epoch = record.header.epoch; - self.guest_sequence = 0; - self.host_sequence = record.header.sequence.wrapping_add(1); - let payload = protocol::encode_ack_payload(OUTER_RECEIVE_WINDOW).map_err(protocol_error)?; - self.write_outer( - OUTER_ACK, - &self.instance_id, - self.epoch, - self.guest_sequence, - &payload, - )?; - self.guest_sequence = 1; - Ok(()) } - /// Sends one outer record in the attached epoch. - pub(crate) fn send_record(&mut self, record_type: u8, payload: &[u8]) -> Result<(), Errno> { - self.write_outer( - record_type, - &self.instance_id, - self.epoch, - self.guest_sequence, - payload, - )?; - self.guest_sequence = self.guest_sequence.wrapping_add(1); - Ok(()) + /// Acknowledges the broker's reset with NVX's receive window. + pub(crate) fn acknowledge(&mut self) -> Result<(), Errno> { + let acknowledged = self.connection.acknowledge(OUTER_RECEIVE_WINDOW); + acknowledged.map_err(|error| self.error(error)) } - /// Replenishes host receive credit after consuming bytes. - pub(crate) fn send_credit(&mut self, bytes: usize) -> Result<(), Errno> { - if bytes == 0 { - return Ok(()); + /// 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. + pub(crate) fn accept_request<'a>( + &mut self, + payload: &'a [u8], + ) -> 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 bytes = u32::try_from(bytes).map_err(|_| Errno::EOVERFLOW)?; - let payload = protocol::encode_credit_payload(bytes).map_err(protocol_error)?; - self.send_record(OUTER_CREDIT, &payload) + let credited = self.connection.credit(payload.len() as u32); + credited.map_err(|error| DataFailure { + kind: DataFailureKind::Credit, + status, + error: self.error(error), + })?; + Ok(frame) } - /// Encodes and sends one application frame. + /// Encodes and sends one application frame, in one DATA record. pub(crate) fn send_app( &mut self, kind: u8, @@ -152,7 +156,8 @@ impl Session { return Err(Errno::EMSGSIZE); } let frame = protocol::encode_app(kind, request_id, status, payload); - self.send_record(OUTER_DATA, &frame) + let sent = self.connection.send(&frame); + sent.map_err(|error| self.error(error)) } /// Sends one application ERROR frame with an ASCII category. @@ -165,69 +170,100 @@ impl Session { self.send_app(APP_ERROR, request_id, status, category.as_bytes()) } - /// Reads and decodes one complete outer record. - pub(crate) fn read_record(&mut self) -> Result { - let mut header = [0u8; OUTER_HEADER_LEN]; - self.descriptor.read_exact(&mut header)?; - let header = protocol::decode_outer_header(&header).map_err(protocol_error)?; - let mut payload = vec![0; header.length]; - self.descriptor.read_exact(&mut payload)?; - Ok(Record { header, payload }) + /// Classifies a failure to receive the broker's next record by the stage at which the C agent + /// reported it. + fn receive_failure(&mut self, error: GuestConnectionError) -> ReceiveFailure { + let epoch = self + .connection + .session() + .identity() + .map_or(0, |identity| identity.epoch); + let failure = match &error { + GuestConnectionError::Protocol(error) => classify(error, epoch), + _ => None, + }; + let error = self.error(error); + failure.unwrap_or(ReceiveFailure::Read(error)) } - /// Validates a DATA record and advances host sequence and credit. - pub(crate) fn validate_data<'a>( - &mut self, - record: &'a Record, - ) -> Result, DataFailure> { - if record.header.record_type != OUTER_DATA || record.header.sequence != self.host_sequence { - return Err(data_failure( - DataFailureKind::Sequence, - record.header.sequence as i32, - )); - } - if record.header.epoch != self.epoch || record.header.instance_id != self.instance_id { - return Err(data_failure( - DataFailureKind::Identity, - record.header.epoch as i32, - )); - } - let frame = protocol::decode_app(&record.payload) - .map_err(|_| data_failure(DataFailureKind::Application, record.payload.len() as i32))?; - if frame.request_id == 0 || frame.status != 0 { - return Err(data_failure( - DataFailureKind::Application, - record.payload.len() as i32, - )); - } - self.host_sequence = self.host_sequence.wrapping_add(1); - self.send_credit(record.payload.len()) - .map_err(|error| DataFailure { - kind: DataFailureKind::Credit, - status: record.payload.len() as i32, - error, - })?; - Ok(frame) + /// Converts a failure of the connection into the error that the agent reports. A connection + /// that an earlier failure ended reports that failure's error again, as another write to a + /// failed descriptor would fail again. + fn error(&mut self, error: GuestConnectionError) -> Errno { + let error = match error { + GuestConnectionError::Ended => return self.last_error.unwrap_or(Errno::EPROTO), + GuestConnectionError::Io(error) => io_error(&error), + // `FileDescriptor::read_exact` reports end of file as `EIO`, inside a record or not. + GuestConnectionError::Closed + | GuestConnectionError::Protocol(SessionError::Truncated { .. }) => Errno::EIO, + GuestConnectionError::Protocol(_) => Errno::EPROTO, + }; + self.last_error = Some(error); + error } +} - /// Encodes and writes one outer record. - fn write_outer( - &self, - record_type: u8, - instance_id: &[u8; 16], - epoch: u64, - sequence: u64, - payload: &[u8], - ) -> Result<(), Errno> { - let record = protocol::encode_outer(record_type, instance_id, epoch, sequence, payload) - .map_err(protocol_error)?; - self.descriptor.write_all(&record) +/// 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. +/// +/// The C agent read each record whole, checking only its header's magic, version, flags, and +/// length limit, then took RESET and DATA records and reported every other type. It checked a +/// RESET's payload and epoch, then a DATA record's sequence, identity, and NVXC frame. A record +/// with one fault thus reports the C agent's stage and status, except for the faults that only +/// the shared session checks: a RESET without an instance ID, of an earlier epoch, or at the +/// last sequence number, DATA at the last sequence number, which the C agent took and followed +/// with sequence number 0 and which reports `data-sequence`, DATA before the first RESET, and DATA +/// beyond the receive window. +fn classify(error: &SessionError, epoch: u64) -> Option { + let data = |kind, status| Some(ReceiveFailure::Data(data_failure(kind, status))); + match *error { + SessionError::Record(ref error) => match *error { + RecordError::Magic(_) | RecordError::Version(_) | RecordError::Flags(_) => None, + RecordError::Length { length, .. } if length > OUTER_MAX_PAYLOAD => None, + RecordError::Type(code) => Some(ReceiveFailure::Type(code)), + RecordError::Length { + record_type: RecordType::Reset, + .. + } => Some(ReceiveFailure::Reset), + // The only other length that DATA does not allow is 0, which holds no NVXC frame. + RecordError::Length { + record_type: RecordType::Data, + length, + } => data(DataFailureKind::Application, length as i32), + RecordError::Length { record_type, .. } + | RecordError::BootstrapIdentity(record_type) => { + Some(ReceiveFailure::Type(record_type.code())) + } + RecordError::ErrorCode(_) => Some(ReceiveFailure::Type(RecordType::Error.code())), + RecordError::ReceiveWindow(_) => Some(ReceiveFailure::Type(RecordType::Ack.code())), + RecordError::Credit(_) => Some(ReceiveFailure::Type(RecordType::Credit.code())), + }, + SessionError::Unexpected(record_type) => Some(ReceiveFailure::Type(record_type.code())), + // Only a RESET at the last sequence number leaves none for the broker's next record. + SessionError::MissingIdentity(_) + | SessionError::StaleEpoch { .. } + | SessionError::SequenceOverflow(RecordType::Reset) => Some(ReceiveFailure::Reset), + // DATA at the last sequence number, which leaves none for the broker's next record, is + // reported as DATA out of sequence, with its sequence number. + SessionError::SequenceOverflow(RecordType::Data) => { + data(DataFailureKind::Sequence, u64::MAX as i32) + } + SessionError::Sequence { actual, .. } => data(DataFailureKind::Sequence, actual as i32), + SessionError::Epoch { actual, .. } => data(DataFailureKind::Identity, actual as i32), + // With one fault, a record of another instance carries the session's epoch, which the C + // agent reported. + SessionError::Instance { .. } => data(DataFailureKind::Identity, epoch as i32), + SessionError::Window { length, .. } => data(DataFailureKind::Window, length as i32), + SessionError::Truncated { .. } + | SessionError::Credit { .. } + | SessionError::SequenceOverflow(_) + | SessionError::InvalidState => None, } } -/// Maps an in-memory codec error to the control protocol's `EPROTO`. -fn protocol_error(_: protocol::ProtocolError) -> Errno { - Errno::EPROTO +/// Returns the `errno` of an I/O failure, or `EIO` for one without, such as a write of no bytes. +fn io_error(error: &io::Error) -> Errno { + error.raw_os_error().map_or(Errno::EIO, Errno::from_raw) } /// Constructs one typed DATA validation failure. diff --git a/nvx-rs/src/guest/mod.rs b/nvx-rs/src/guest/mod.rs index 2b2ff6bc4..be9a6d16c 100644 --- a/nvx-rs/src/guest/mod.rs +++ b/nvx-rs/src/guest/mod.rs @@ -40,11 +40,12 @@ //! it, except that a test may hand one to `FileDescriptor::from_owned`, which takes it over //! whole. No descriptor thus has two owners or two buffers; a standard-library adapter such as //! `BufReader` may add the one buffer of a `sys` descriptor through the `std::io::Read` that -//! `FileDescriptor` implements. `sys::Stdout` and `sys::DirectStdout` write standard output -//! directly, past the standard library's line buffer, so the programs never print with the -//! standard library's macros: `print!` would reorder its buffered output with theirs, and -//! `eprint!` would bypass the error streams that the programs' tests capture. Clippy denies -//! both here. +//! `FileDescriptor` implements, and a caller that takes `std::io::Write`, such as [`nvxs`], +//! writes one through the `std::io::Write` that it implements, which buffers nothing. +//! `sys::Stdout` and `sys::DirectStdout` write standard output directly, past the standard +//! library's line buffer, so the programs never print with the standard library's macros: +//! `print!` would reorder its buffered output with theirs, and `eprint!` would bypass the error +//! streams that the programs' tests capture. Clippy denies both here. //! - Every program fails with the shared `CommandError`, whose kind selects the program's exit //! status from the shared `ExitStatus`. Every module, type, field, variant, constant, static, //! and function is documented, which CI enforces with `clippy::missing_docs_in_private_items`. diff --git a/nvx-rs/src/guest/sys/fd.rs b/nvx-rs/src/guest/sys/fd.rs index 3c58c17d2..fca62ab41 100644 --- a/nvx-rs/src/guest/sys/fd.rs +++ b/nvx-rs/src/guest/sys/fd.rs @@ -180,6 +180,22 @@ impl io::Read for FileDescriptor { } } +/// Writes through the standard library's I/O traits, so that a caller that takes `io::Write`, +/// such as the guest's NVXS connection, writes a descriptor that this module opened. +/// +/// Each call is one `write` system call. An interrupted call is `ErrorKind::Interrupted`, which +/// `write_all` retries; a write of no bytes ends `write_all` with `ErrorKind::WriteZero`. Nothing +/// is buffered, so `flush` does nothing. +impl io::Write for FileDescriptor { + fn write(&mut self, bytes: &[u8]) -> io::Result { + FileDescriptor::write(self, bytes).map_err(io::Error::from) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + /// Gives up the descriptor to its new owner, which `crate::guest::spawn` hands to a child. impl From for OwnedFd { fn from(descriptor: FileDescriptor) -> Self { diff --git a/nvx-rs/src/protocol/nvxs/guest.rs b/nvx-rs/src/protocol/nvxs/guest.rs index 055cb2bfe..0bfadc209 100644 --- a/nvx-rs/src/protocol/nvxs/guest.rs +++ b/nvx-rs/src/protocol/nvxs/guest.rs @@ -169,7 +169,7 @@ impl GuestSession { let (GuestState::Active, Some(identity)) = (self.state, self.identity) else { return Err(SessionError::InvalidState); }; - let next = next_sequence(self.send_sequence)?; + let next = next_sequence(self.send_sequence, RecordType::Data)?; let taken = encode_data(&identity, self.send_sequence, data, output)?; self.send_sequence = next; Ok(taken) @@ -187,7 +187,7 @@ impl GuestSession { bytes, }); } - let next = next_sequence(self.send_sequence)?; + let next = next_sequence(self.send_sequence, RecordType::Credit)?; encode_record( RecordType::Credit, &identity.instance_id, @@ -243,7 +243,7 @@ impl GuestSession { actual: identity.epoch, }); } - let next = next_sequence(record.sequence())?; + let next = next_sequence(record.sequence(), RecordType::Reset)?; self.identity = Some(identity); self.receive_sequence = next; self.send_sequence = 0; @@ -261,7 +261,7 @@ impl GuestSession { }; check_sequence(&record, self.receive_sequence)?; check_identity(&record, &identity)?; - let next = next_sequence(self.receive_sequence)?; + let next = next_sequence(self.receive_sequence, RecordType::Data)?; // A DATA payload has at most `OUTER_MAX_PAYLOAD` bytes, so its length fits a `u32`. let length = record.payload().len() as u32; if length > self.window.saturating_sub(self.outstanding) { diff --git a/nvx-rs/src/protocol/nvxs/host.rs b/nvx-rs/src/protocol/nvxs/host.rs index 284905b49..5a1509f99 100644 --- a/nvx-rs/src/protocol/nvxs/host.rs +++ b/nvx-rs/src/protocol/nvxs/host.rs @@ -135,7 +135,7 @@ impl HostSession { let (HostState::Attached, Some(identity)) = (self.state, self.identity) else { return Err(SessionError::InvalidState); }; - let next = next_sequence(self.send_sequence)?; + let next = next_sequence(self.send_sequence, RecordType::Data)?; let taken = encode_data(&identity, self.send_sequence, data, output)?; self.send_sequence = next; Ok(taken) @@ -176,7 +176,7 @@ impl HostSession { return Err(SessionError::Unexpected(record_type)); } check_sequence(&record, self.receive_sequence)?; - let next = next_sequence(self.receive_sequence)?; + let next = next_sequence(self.receive_sequence, record_type)?; let event = match (record_type, self.identity) { (RecordType::Wait, _) => HostEvent::Wait, (RecordType::Ready, _) => { diff --git a/nvx-rs/src/protocol/nvxs/mod.rs b/nvx-rs/src/protocol/nvxs/mod.rs index 5dd88503a..4af9b6136 100644 --- a/nvx-rs/src/protocol/nvxs/mod.rs +++ b/nvx-rs/src/protocol/nvxs/mod.rs @@ -159,8 +159,9 @@ pub enum SessionError { /// Sequence number of the record. actual: u64, }, - /// The sequence numbers of a direction ran out. - SequenceOverflow, + /// The sequence numbers of a direction ran out at a record of this type, which took or would + /// take the last one and leave none for the next record. + SequenceOverflow(RecordType), /// A DATA record from the broker that exceeds the guest's receive window. Window { /// Receive window that the guest advertised. @@ -216,7 +217,10 @@ impl fmt::Display for SessionError { formatter, "NVXS record has sequence {actual} instead of {expected}" ), - Self::SequenceOverflow => formatter.write_str("NVXS sequence numbers ran out"), + Self::SequenceOverflow(record_type) => write!( + formatter, + "NVXS sequence numbers ran out at the {record_type} record" + ), Self::Window { window, outstanding, @@ -264,11 +268,11 @@ fn record_identity(record: &Record) -> Option { }) } -/// Returns the sequence number after `sequence`. -fn next_sequence(sequence: u64) -> Result { +/// Returns the sequence number after `sequence`, which a record of `record_type` takes. +fn next_sequence(sequence: u64, record_type: RecordType) -> Result { sequence .checked_add(1) - .ok_or(SessionError::SequenceOverflow) + .ok_or(SessionError::SequenceOverflow(record_type)) } /// Checks that `record` carries the sequence number `expected`. diff --git a/nvx-rs/src/protocol/nvxs/tests/guest.rs b/nvx-rs/src/protocol/nvxs/tests/guest.rs index a12e14b31..747a8104c 100644 --- a/nvx-rs/src/protocol/nvxs/tests/guest.rs +++ b/nvx-rs/src/protocol/nvxs/tests/guest.rs @@ -207,7 +207,8 @@ fn the_guest_checks_the_identity_of_each_reset() { // A RESET at the last sequence number leaves none for the broker's next record. let mut guest = attaching_guest(); let reset = record(RecordType::Reset, IDENTITY, u64::MAX, &[]); - assert_eq!(guest.receive(&reset), Err(SessionError::SequenceOverflow)); + let expected = SessionError::SequenceOverflow(RecordType::Reset); + assert_eq!(guest.receive(&reset), Err(expected)); assert_eq!(guest.state(), GuestState::Failed); } @@ -275,6 +276,20 @@ fn the_guest_checks_the_sequence_and_identity_of_data() { actual: 1, }; assert_eq!(guest.receive(&data), Err(expected)); + + // After a RESET at the second-to-last sequence number, DATA at the last one leaves none for + // the broker's next record. + let mut guest = attaching_guest(); + let reset = record(RecordType::Reset, IDENTITY, u64::MAX - 1, &[]); + let event = Some(GuestEvent::Reset(IDENTITY)); + assert_eq!(guest.receive(&reset), Ok((reset.len(), event))); + guest + .acknowledge(OUTER_RECEIVE_WINDOW, &mut Vec::new()) + .unwrap(); + let last = record(RecordType::Data, IDENTITY, u64::MAX, b"x"); + let expected = SessionError::SequenceOverflow(RecordType::Data); + assert_eq!(guest.receive(&last), Err(expected)); + assert_eq!(guest.state(), GuestState::Failed); } #[test]