diff --git a/packages/runtime-core/README.md b/packages/runtime-core/README.md index 3f650f695..1af11cad9 100644 --- a/packages/runtime-core/README.md +++ b/packages/runtime-core/README.md @@ -490,7 +490,12 @@ that does not name an owned network or `none`. Cold builds need free guest stora cache to 8 GiB, and `runtime reclaim-build-images` removes only owned, superseded, unreferenced images (`provider/image_build/reclaim.rs`; model `tests/models/tla/native-build-retention`). Native `up` calls the build operation -for Compose `build:` services only when `HACK_NATIVE_BUILDS=1` is set. +for Compose `build:` services only when `HACK_NATIVE_BUILDS=1` is set. The client reaches +the engine through a private relay (`provider/image_build/client_relay.rs`): the pinned +provider's macOS socket forwarding can loop forever when a host client closes a forwarded +connection while the guest is still sending, which saturated the guest and stopped every later +agent and engine request. The relay always drains guest bytes, forwards a client's end of +stream only as a half-close, and closes a provider connection only after the guest ended it. Source binaries embed their build checkout. If a copied Cargo target directory retains a different checkout identity, use a fresh `--target-dir` for checks and @@ -500,6 +505,12 @@ tests; Cargo reporting a cached binary as fresh does not qualify that identity. native process/disk identity, guest protocol, and lifecycle decisions. Guest requests use a bounded connection to the existing socket; they cannot silently start or recover a VM. Failed operations retain their phase and receipts. Recovery preserves disks and labels an unclean exit explicitly. +When the guest agent stops responding (three pings over about 13 seconds start no reply), +`runtime down` and `runtime recover` stop this home's recorded provider after verifying its PID, +start time, user and executable, without guest quiescence, and record `recovered-unclean`; a +guest that answers keeps the normal clean stop, and `recover` still refuses it. Both refuse with +`provider_busy` while a native image build holds its fence; finishing or canceling that build +(Ctrl-C its `hack up`) releases the fence, also when a build process was killed. SmolVM seeds a machine's disks during its first start (`machine create` makes none) from expanded `*-template.ext4` files that it creates from the verified compressed templates and then reuses. diff --git a/packages/runtime-core/src/provider/agent.rs b/packages/runtime-core/src/provider/agent.rs index 1aed6bdba..2a7e151db 100644 --- a/packages/runtime-core/src/provider/agent.rs +++ b/packages/runtime-core/src/provider/agent.rs @@ -112,6 +112,59 @@ fn exchange( serde_json::from_slice(&body).map_err(|_| failure("Malformed guest response.")) } +/// Whether the agent at `path` starts any reply to a ping within `timeout`. No +/// connection, a connection that ends before a reply, and no reply by the deadline are +/// silence; any reply byte, even of an unexpected response, is an answer. Only a stop of +/// a guest that stopped responding relies on this. +pub(super) fn answers_ping(path: &Path, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + let Ok(bytes) = encode(&json!({"method":"ping"})) else { + return false; + }; + let Ok(mut stream) = UnixStream::connect(path) else { + return false; + }; + let mut frame = (bytes.len() as u32).to_be_bytes().to_vec(); + frame.extend_from_slice(&bytes); + if stream.set_nonblocking(true).is_err() { + return false; + } + let mut written = 0; + let mut reply = [0_u8; 1]; + loop { + let Some(remaining) = deadline.checked_duration_since(Instant::now()) else { + return false; + }; + if written < frame.len() { + match stream.write(&frame[written..]) { + Ok(count) => written += count, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {} + Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {} + Err(_) => return false, + } + } + match stream.read(&mut reply) { + Ok(0) => return false, + Ok(_) => return true, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {} + Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {} + Err(_) => return false, + } + let mut descriptor = libc::pollfd { + fd: stream.as_raw_fd(), + events: if written < frame.len() { + libc::POLLIN | libc::POLLOUT + } else { + libc::POLLIN + }, + revents: 0, + }; + let milliseconds = (remaining.as_millis() + 1).min(i32::MAX as u128) as i32; + // SAFETY: the descriptor is borrowed from the live stream for this synchronous wait. + unsafe { libc::poll(&mut descriptor, 1, milliseconds) }; + } +} + pub fn ping(path: &Path) -> Result<(), CandidateError> { let response = request(path, json!({"method":"ping"}), Duration::from_secs(3))?; if response["status"] != "pong" || response["version"] != 1 { diff --git a/packages/runtime-core/src/provider/image_build.rs b/packages/runtime-core/src/provider/image_build.rs index 371484822..38a781259 100644 --- a/packages/runtime-core/src/provider/image_build.rs +++ b/packages/runtime-core/src/provider/image_build.rs @@ -2,7 +2,9 @@ //! //! One operation builds one image for one build key inside the guest engine, through //! the pinned host BuildKit client (`docker-buildx`, docker driver) talking to the -//! engine's verified forwarded socket. The built image lands in the guest image store, +//! engine's verified forwarded socket through a private relay ([`client_relay`]) that +//! keeps the provider's socket forwarding out of a known busy loop. The built image +//! lands in the guest image store, //! where graph admission already looks, and the caller pins the service to the content //! ID it returns, so review, receipts and restore treat it like a pulled image. //! @@ -68,6 +70,7 @@ use std::{ time::{Duration, Instant}, }; +mod client_relay; mod output; pub mod reclaim; #[cfg(test)] @@ -875,8 +878,17 @@ pub(super) fn run( // outlives the client run. let _definition = RemoveOnDrop(scratch.join("definition").join(BAKE_DEFINITION)); let _policy = RemoveOnDrop(scratch.join("definition").join(SOURCE_POLICY)); + // The client reaches the engine only through the relay, which keeps the provider's + // forwarded socket out of the state described in `client_relay`. + let relay = client_relay::ClientRelay::start(&identity.socket)?; let mut client = command( - &builder, &scratch, &metadata, inputs, &identity, &pins, &expected, + &builder, + &scratch, + &metadata, + inputs, + relay.socket(), + &pins, + &expected, )?; inherit_locks(&mut client, &[&fence, &key_lock]); let ended = supervise( @@ -886,6 +898,8 @@ pub(super) fn run( output::Filter::new(inputs.args.values().map(String::as_str)), pool.output(), ); + // The client has exited or was stopped; wait for the guest to end its connections. + relay.finish(client_relay::FINISH_GRACE); let ended = match ended { Ok(ended) => ended, Err(error) => { @@ -973,6 +987,20 @@ pub(super) fn run( }) } +/// Hold the shared build fence as an admitted build does, for the macOS-only lifecycle +/// test of the silent-guest stop. +#[cfg(all(test, target_os = "macos"))] +pub(super) fn hold_build_fence_for_test(candidate: &Candidate) -> impl Drop { + let root = builds_root(candidate); + state::private_directory(&root).unwrap(); + Flock::try_lock( + Flock::open(&root.join("active.lock")).unwrap(), + libc::LOCK_SH, + ) + .unwrap() + .unwrap() +} + /// The image config digest the client reports for the built image. fn built_image_id(metadata: &Path) -> Result { let bytes = read_bounded_file(metadata, MAX_METADATA_BYTES)?.ok_or_else(|| { @@ -1101,7 +1129,7 @@ fn command( scratch: &Path, metadata: &Path, inputs: &Inputs, - identity: &PoolIdentity, + engine: &Path, pins: &BTreeMap, labels: &BTreeMap<&'static str, String>, ) -> Result { @@ -1128,10 +1156,7 @@ fn command( .current_dir(scratch) .env("HOME", scratch) .env("TMPDIR", scratch) - .env( - "DOCKER_HOST", - format!("unix://{}", identity.socket.display()), - ) + .env("DOCKER_HOST", format!("unix://{}", engine.display())) // An empty private client configuration: no contexts, credential helpers or // proxies from the user's Docker setup. .env("DOCKER_CONFIG", &config) diff --git a/packages/runtime-core/src/provider/image_build/client_relay.rs b/packages/runtime-core/src/provider/image_build/client_relay.rs new file mode 100644 index 000000000..446d42897 --- /dev/null +++ b/packages/runtime-core/src/provider/image_build/client_relay.rs @@ -0,0 +1,600 @@ +//! A private host relay between the build client and the provider's forwarded engine +//! socket. +//! +//! The pinned provider forwards the guest engine socket to the host through its vsock +//! muxer. On macOS that muxer can enter a permanent busy loop when the host peer of a +//! forwarded connection goes away while the guest is still sending on it: the send fails, +//! the stream is reset, a later credit update from the guest re-arms it with only write +//! interest, and deleting that interest fails (one `kevent` change list stops at the +//! unregistered read filter). The dead socket then reports a hang-up on every wait, each +//! one queues another reset to the guest and restarts the stream's reaper delay, so the +//! stream is never removed. The guest's vCPUs and the shared vsock queue stay saturated +//! and every later agent or engine request times out. +//! +//! The BuildKit client closes many connections before reading the whole response, and +//! closes the rest at exit, so builds reached that state in most runs. The relay keeps the +//! provider-side connection away from it with three rules: +//! +//! - Provider bytes are always read promptly, including after the client has gone (they +//! are then counted and discarded), so the provider never sees a failed send. +//! - The client's end of stream reaches the provider only as a half-close, after every +//! byte the client sent was delivered. +//! - The relay closes a provider-side connection only after the provider reported end of +//! stream, which means the guest side finished. [`ClientRelay::finish`] bounds that wait +//! and then closes what is left, counting it in [`RelayReport::forced`]. +//! +//! The listening socket lives in a fresh mode-0700 directory under the system temporary +//! root, so its path fits `sockaddr_un` and only this user can connect. It accepts only +//! while the client runs; nothing else is relayed and no bytes are inspected or logged. +use crate::CandidateError; +use std::{ + fs::{self, File}, + io::{ErrorKind, Read, Write}, + os::fd::AsRawFd, + os::unix::{ + fs::{DirBuilderExt, MetadataExt}, + net::{UnixListener, UnixStream}, + }, + path::{Path, PathBuf}, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, + thread::JoinHandle, + time::{Duration, Instant}, +}; + +/// How long [`ClientRelay::finish`] waits for the guest to end its connections. +pub(super) const FINISH_GRACE: Duration = Duration::from_secs(10); +const TICK: Duration = Duration::from_millis(50); +const CHUNK: usize = 64 * 1024; +/// Client bytes not yet delivered; reading the client pauses above this. +const TO_PROVIDER_LIMIT: usize = 1024 * 1024; +/// Provider bytes not yet delivered to a live client; reading the provider pauses above +/// this only while the client is still connected but not reading. +const TO_CLIENT_LIMIT: usize = 16 * 1024 * 1024; +/// A larger receive buffer lets the provider's sends complete at once instead of +/// queueing in the provider. +const PROVIDER_RECEIVE_BUFFER: libc::c_int = 4 * 1024 * 1024; +const SOCKET_NAME: &str = "engine.sock"; + +/// What the relay did, for tests and diagnostics. It carries no payload bytes. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub(super) struct RelayReport { + /// Client connections accepted. + pub connections: u32, + /// Provider bytes read after the client had gone. + pub discarded_bytes: u64, + /// Connections closed by the bound in `finish` before the provider ended them. + pub forced: u32, + /// Client connections dropped because the provider socket refused the connection. + pub refused: u32, +} + +struct Shared { + stop: AtomicBool, + force: AtomicBool, + report: Mutex, + connections: Mutex>>, +} + +impl Shared { + fn update(&self, change: impl FnOnce(&mut RelayReport)) { + // A poisoned lock means a connection thread panicked; the counts stay usable. + let mut report = self + .report + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + change(&mut report); + } +} + +/// A running relay. Call [`ClientRelay::finish`] once the client has exited. +pub(super) struct ClientRelay { + directory: PathBuf, + socket: PathBuf, + shared: Arc, + acceptor: Option>, +} + +impl ClientRelay { + /// Listen on a fresh private socket and relay each connection to `provider`. + /// Connecting to `provider` happens per client connection, so starting never touches + /// the engine. + pub(super) fn start(provider: &Path) -> Result { + Self::start_in(temporary_root(), provider) + } + + /// [`Self::start`] with its private directory (and the sweep of abandoned ones) under + /// `root` instead of the temporary directory; tests give each relay its own root. + fn start_in(root: &Path, provider: &Path) -> Result { + let directory = private_directory(root)?; + let socket = directory.join(SOCKET_NAME); + let shared = Arc::new(Shared { + stop: AtomicBool::new(false), + force: AtomicBool::new(false), + report: Mutex::new(RelayReport::default()), + connections: Mutex::new(Vec::new()), + }); + // From here on, dropping `relay` removes the socket and its directory. + let mut relay = Self { + directory, + socket, + shared, + acceptor: None, + }; + let listener = UnixListener::bind(&relay.socket) + .and_then(|listener| listener.set_nonblocking(true).map(|()| listener)) + .map_err(|_| failure("Cannot listen for the build client."))?; + let provider = provider.to_path_buf(); + let accepting = Arc::clone(&relay.shared); + relay.acceptor = Some( + std::thread::Builder::new() + .name("build-client-relay".into()) + .spawn(move || accept(&listener, &provider, &accepting)) + .map_err(|_| failure("Cannot start the build client relay."))?, + ); + Ok(relay) + } + + /// The socket the client connects to. + pub(super) fn socket(&self) -> &Path { + &self.socket + } + + /// Stop accepting, then wait up to `grace` for the provider to end every relayed + /// connection, closing any that remain after it. + pub(super) fn finish(mut self, grace: Duration) -> RelayReport { + self.stop(grace) + } + + fn stop(&mut self, grace: Duration) -> RelayReport { + self.shared.stop.store(true, Ordering::Release); + if let Some(acceptor) = self.acceptor.take() { + let _ = acceptor.join(); + } + let _ = fs::remove_file(&self.socket); + let _ = fs::remove_dir(&self.directory); + let connections = std::mem::take( + &mut *self + .shared + .connections + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()), + ); + let deadline = Instant::now() + grace; + while connections.iter().any(|handle| !handle.is_finished()) && Instant::now() < deadline { + std::thread::sleep(TICK); + } + self.shared.force.store(true, Ordering::Release); + for handle in connections { + let _ = handle.join(); + } + *self + .shared + .report + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } +} + +impl Drop for ClientRelay { + fn drop(&mut self) { + if self.acceptor.is_some() { + // Only an early return reaches this; the client has already been stopped. + self.stop(FINISH_GRACE); + } else { + let _ = fs::remove_file(&self.socket); + let _ = fs::remove_dir(&self.directory); + } + } +} + +fn failure(message: &str) -> CandidateError { + CandidateError::new("native_build_client", message) +} + +const DIRECTORY_PREFIX: &str = "hkb-"; + +fn temporary_root() -> &'static Path { + if cfg!(target_os = "macos") { + Path::new("/private/tmp") + } else { + Path::new("/tmp") + } +} + +/// A new mode-0700 directory with a short path under `parent`, owned by this user. Its +/// name carries this process's PID, so a later relay can remove what a killed process +/// left behind. +fn private_directory(parent: &Path) -> Result { + remove_abandoned(parent); + for _ in 0..8 { + let mut bytes = [0_u8; 6]; + File::open("/dev/urandom") + .and_then(|mut random| random.read_exact(&mut bytes)) + .map_err(|_| failure("Cannot choose a private build client socket."))?; + let name: String = bytes.iter().map(|byte| format!("{byte:02x}")).collect(); + let directory = parent.join(format!("{DIRECTORY_PREFIX}{}-{name}", std::process::id())); + match fs::DirBuilder::new().mode(0o700).create(&directory) { + Ok(()) => {} + Err(error) if error.kind() == ErrorKind::AlreadyExists => continue, + Err(_) => return Err(failure("Cannot create a private build client socket.")), + } + let metadata = fs::symlink_metadata(&directory) + .map_err(|_| failure("Cannot inspect the private build client socket."))?; + // SAFETY: geteuid has no preconditions. + if !metadata.is_dir() + || metadata.uid() != unsafe { libc::geteuid() } + || metadata.mode() & 0o777 != 0o700 + { + return Err(failure( + "The private build client socket directory is not private.", + )); + } + return Ok(directory); + } + Err(failure("Cannot create a private build client socket.")) +} + +/// Remove relay directories whose creating process is gone (it was killed before it +/// could clean up). Only a mode-0700 directory owned by this user, named for a PID that +/// no longer exists and holding nothing but the relay socket, is removed; anything else +/// is left alone. A reused PID only keeps a stale directory longer. +fn remove_abandoned(parent: &Path) { + let Ok(entries) = fs::read_dir(parent) else { + return; + }; + // SAFETY: geteuid has no preconditions. + let user = unsafe { libc::geteuid() }; + for entry in entries.flatten() { + let name = entry.file_name(); + let Some(pid) = name + .to_str() + .and_then(|name| name.strip_prefix(DIRECTORY_PREFIX)) + .and_then(|rest| rest.split_once('-')) + .and_then(|(pid, _)| pid.parse::().ok()) + .filter(|pid| *pid > 0) + else { + continue; + }; + // SAFETY: signal 0 only checks for existence; no signal is delivered. + let gone = unsafe { libc::kill(pid, 0) } != 0 + && std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH); + let directory = entry.path(); + let private = fs::symlink_metadata(&directory).is_ok_and(|metadata| { + metadata.is_dir() && metadata.uid() == user && metadata.mode() & 0o777 == 0o700 + }); + if !gone || !private { + continue; + } + let socket = directory.join(SOCKET_NAME); + if fs::symlink_metadata(&socket).is_ok_and(|metadata| { + use std::os::unix::fs::FileTypeExt; + metadata.file_type().is_socket() && metadata.uid() == user + }) { + let _ = fs::remove_file(&socket); + } + // Fails, and keeps the directory, unless it is now empty. + let _ = fs::remove_dir(&directory); + } +} + +fn accept(listener: &UnixListener, provider: &Path, shared: &Arc) { + while !shared.stop.load(Ordering::Acquire) { + match listener.accept() { + Ok((client, _)) => { + shared.update(|report| report.connections += 1); + let Ok(upstream) = connect(provider) else { + shared.update(|report| report.refused += 1); + continue; + }; + let relaying = Arc::clone(shared); + let spawned = std::thread::Builder::new() + .name("build-client-connection".into()) + .spawn(move || { + if let Ok(mut connection) = Connection::new(client, upstream) { + connection.run(&relaying); + } + }); + if let Ok(handle) = spawned { + shared + .connections + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .push(handle); + } + } + Err(error) if error.kind() == ErrorKind::WouldBlock => std::thread::sleep(TICK), + Err(error) if error.kind() == ErrorKind::Interrupted => {} + Err(_) => std::thread::sleep(TICK), + } + } +} + +fn connect(provider: &Path) -> std::io::Result { + let stream = UnixStream::connect(provider)?; + // SAFETY: the stream owns this live descriptor for the call; `size` points to a + // readable native int of the option's size. A refused size only loses headroom. + unsafe { + libc::setsockopt( + stream.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_RCVBUF, + (&PROVIDER_RECEIVE_BUFFER as *const libc::c_int).cast(), + std::mem::size_of::() as libc::socklen_t, + ); + } + Ok(stream) +} + +/// Pending bytes for one direction. Storage follows the bytes still pending, not the +/// traffic: the delivered prefix is dropped before it outgrows what remains, so the +/// buffer stays within about twice the pending bytes plus one read. +#[derive(Default)] +struct Pending { + bytes: Vec, + start: usize, +} + +impl Pending { + fn len(&self) -> usize { + self.bytes.len() - self.start + } + fn is_empty(&self) -> bool { + self.len() == 0 + } + fn push(&mut self, bytes: &[u8]) { + // Dropping the delivered prefix once it is at least half the buffer moves each + // byte at most once more, so the cost stays linear in the traffic. + if self.start > 0 && self.start >= self.bytes.len() / 2 { + self.bytes.drain(..self.start); + self.start = 0; + } + self.bytes.extend_from_slice(bytes); + } + fn clear(&mut self) { + self.bytes.clear(); + self.start = 0; + } + /// Write what `stream` accepts now. `Err` means the peer can take no more. + fn flush(&mut self, stream: &mut impl Write) -> Result<(), ()> { + while !self.is_empty() { + match stream.write(&self.bytes[self.start..]) { + Ok(0) => return Err(()), + Ok(count) => self.start += count, + Err(error) if error.kind() == ErrorKind::WouldBlock => return Ok(()), + Err(error) if error.kind() == ErrorKind::Interrupted => {} + Err(_) => return Err(()), + } + } + if self.start == self.bytes.len() { + self.clear(); + } + Ok(()) + } +} + +enum Received { + Data(usize), + End, + Again, +} + +fn read_some(stream: &mut UnixStream, buffer: &mut [u8]) -> Received { + loop { + match stream.read(buffer) { + Ok(0) => return Received::End, + Ok(count) => return Received::Data(count), + Err(error) if error.kind() == ErrorKind::WouldBlock => return Received::Again, + Err(error) if error.kind() == ErrorKind::Interrupted => {} + // A reset ends the stream like an orderly end of stream. + Err(_) => return Received::End, + } + } +} + +/// One relayed connection. `client_*` is the build client's side, `provider_*` the +/// provider's forwarded engine socket. +struct Connection { + client: UnixStream, + provider: UnixStream, + to_provider: Pending, + to_client: Pending, + /// The client sent end of stream or can no longer be read. + client_ended: bool, + /// The client can no longer receive; provider bytes are discarded. + client_gone: bool, + /// The provider's end of stream was forwarded to the client. + client_closed_write: bool, + /// The provider reported end of stream: the guest side finished sending. + provider_ended: bool, + /// The provider can no longer receive. + provider_gone: bool, + /// The client's end of stream was forwarded to the provider. + provider_closed_write: bool, + /// Provider bytes that could not reach the client because it had gone. + discarded: u64, +} + +impl Connection { + fn new(client: UnixStream, provider: UnixStream) -> std::io::Result { + client.set_nonblocking(true)?; + provider.set_nonblocking(true)?; + Ok(Self { + client, + provider, + to_provider: Pending::default(), + to_client: Pending::default(), + client_ended: false, + client_gone: false, + client_closed_write: false, + provider_ended: false, + provider_gone: false, + provider_closed_write: false, + discarded: 0, + }) + } + + fn done(&self) -> bool { + self.provider_ended + && (self.to_provider.is_empty() || self.provider_gone) + && (self.to_client.is_empty() || self.client_gone) + } + + fn run(&mut self, shared: &Shared) { + let forced = self.relay(shared); + let discarded = self.discarded; + shared.update(|report| { + report.discarded_bytes += discarded; + report.forced += u32::from(forced); + }); + } + + /// Relay until the connection is done; `true` when `finish` closed it first. + fn relay(&mut self, shared: &Shared) -> bool { + let mut buffer = vec![0_u8; CHUNK]; + while !self.done() { + if shared.force.load(Ordering::Acquire) { + return true; + } + let client_events = self.client_events(); + let provider_events = self.provider_events(); + let mut descriptors = [ + libc::pollfd { + // poll ignores a negative descriptor, so an idle side cannot wake it. + fd: if client_events == 0 { + -1 + } else { + self.client.as_raw_fd() + }, + events: client_events, + revents: 0, + }, + libc::pollfd { + fd: if provider_events == 0 { + -1 + } else { + self.provider.as_raw_fd() + }, + events: provider_events, + revents: 0, + }, + ]; + // SAFETY: `descriptors` is live, writable storage for two entries, and both + // non-negative descriptors are owned by this connection for the call. + let ready = unsafe { + libc::poll( + descriptors.as_mut_ptr(), + descriptors.len() as libc::nfds_t, + TICK.as_millis() as libc::c_int, + ) + }; + if ready <= 0 { + continue; + } + let woke = |entry: &libc::pollfd, wanted: libc::c_short| { + entry.revents & (wanted | libc::POLLHUP | libc::POLLERR) != 0 + }; + if woke(&descriptors[1], libc::POLLIN) && !self.provider_ended { + self.read_provider(&mut buffer); + } + if woke(&descriptors[0], libc::POLLIN) && !self.client_ended { + self.read_client(&mut buffer); + } + if woke(&descriptors[1], libc::POLLOUT) + && !self.provider_gone + && self.to_provider.flush(&mut self.provider).is_err() + { + self.provider_gone = true; + self.to_provider.clear(); + } + if woke(&descriptors[0], libc::POLLOUT) && !self.client_gone { + self.flush_client(); + } + self.forward_ends(); + } + false + } + + fn client_events(&self) -> libc::c_short { + let mut events = 0; + if !self.client_ended && self.to_provider.len() < TO_PROVIDER_LIMIT { + events |= libc::POLLIN; + } + if !self.client_gone && !self.to_client.is_empty() { + events |= libc::POLLOUT; + } + events + } + + fn provider_events(&self) -> libc::c_short { + let mut events = 0; + // Provider bytes are read whenever they can be kept or discarded: this is what + // keeps the provider's sends from failing or queueing. + if !self.provider_ended && (self.client_gone || self.to_client.len() < TO_CLIENT_LIMIT) { + events |= libc::POLLIN; + } + if !self.provider_gone && !self.to_provider.is_empty() { + events |= libc::POLLOUT; + } + events + } + + fn read_provider(&mut self, buffer: &mut [u8]) { + match read_some(&mut self.provider, buffer) { + Received::Data(count) if self.client_gone => self.discarded += count as u64, + Received::Data(count) => { + self.to_client.push(&buffer[..count]); + self.flush_client(); + } + Received::End => self.provider_ended = true, + Received::Again => {} + } + } + + fn read_client(&mut self, buffer: &mut [u8]) { + match read_some(&mut self.client, buffer) { + Received::Data(count) => { + self.to_provider.push(&buffer[..count]); + if !self.provider_gone && self.to_provider.flush(&mut self.provider).is_err() { + self.provider_gone = true; + self.to_provider.clear(); + } + } + Received::End => self.client_ended = true, + Received::Again => {} + } + } + + fn flush_client(&mut self) { + if self.to_client.flush(&mut self.client).is_err() { + self.client_gone = true; + self.client_ended = true; + self.discarded += self.to_client.len() as u64; + self.to_client.clear(); + } + } + + /// Forward each side's end of stream as a half-close once its bytes were delivered. + fn forward_ends(&mut self) { + if self.client_ended + && !self.provider_closed_write + && (self.to_provider.is_empty() || self.provider_gone) + { + self.provider_closed_write = true; + let _ = self.provider.shutdown(std::net::Shutdown::Write); + } + if self.provider_ended + && !self.client_closed_write + && !self.client_gone + && self.to_client.is_empty() + { + self.client_closed_write = true; + let _ = self.client.shutdown(std::net::Shutdown::Write); + } + } +} + +#[cfg(test)] +mod tests; diff --git a/packages/runtime-core/src/provider/image_build/client_relay/tests.rs b/packages/runtime-core/src/provider/image_build/client_relay/tests.rs new file mode 100644 index 000000000..9c6acafd0 --- /dev/null +++ b/packages/runtime-core/src/provider/image_build/client_relay/tests.rs @@ -0,0 +1,279 @@ +//! The relay against a fake provider socket. The fake provider plays the provider's side +//! of the hazard: it keeps sending after the client left and checks that the relay +//! neither fails those sends nor closes first. +use super::*; +use std::{net::Shutdown, os::unix::fs::FileTypeExt}; + +/// A fake provider socket in its own short private directory. +struct Provider { + directory: PathBuf, + listener: UnixListener, +} +impl Provider { + fn new() -> Self { + let directory = private_directory(temporary_root()).unwrap(); + let listener = UnixListener::bind(directory.join("provider.sock")).unwrap(); + Self { + directory, + listener, + } + } + fn path(&self) -> PathBuf { + self.directory.join("provider.sock") + } + fn accept(&self) -> UnixStream { + let (stream, _) = self.listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + stream + } +} +impl Drop for Provider { + fn drop(&mut self) { + let _ = fs::remove_file(self.path()); + let _ = fs::remove_dir(&self.directory); + } +} + +fn read_to_end(stream: &mut UnixStream) -> Vec { + let mut bytes = Vec::new(); + stream.read_to_end(&mut bytes).unwrap(); + bytes +} + +#[test] +fn bytes_pass_both_ways_and_each_end_arrives_as_a_half_close() { + let provider = Provider::new(); + let relay = ClientRelay::start(&provider.path()).unwrap(); + let mut client = UnixStream::connect(relay.socket()).unwrap(); + client + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + let mut upstream = provider.accept(); + client.write_all(b"GET /_ping HTTP/1.1\r\n\r\n").unwrap(); + client.shutdown(Shutdown::Write).unwrap(); + // The client's end of stream reaches the provider after its bytes, as a half-close. + assert_eq!(read_to_end(&mut upstream), b"GET /_ping HTTP/1.1\r\n\r\n"); + // The provider can still answer on the half-closed connection. + upstream.write_all(b"HTTP/1.1 200 OK\r\n\r\nOK").unwrap(); + upstream.shutdown(Shutdown::Write).unwrap(); + assert_eq!(read_to_end(&mut client), b"HTTP/1.1 200 OK\r\n\r\nOK"); + let report = relay.finish(FINISH_GRACE); + assert_eq!( + report, + RelayReport { + connections: 1, + ..RelayReport::default() + } + ); +} + +#[test] +fn provider_bytes_after_the_client_left_are_drained_and_the_provider_ends_the_connection() { + let provider = Provider::new(); + let relay = ClientRelay::start(&provider.path()).unwrap(); + let mut client = UnixStream::connect(relay.socket()).unwrap(); + let mut upstream = provider.accept(); + client.write_all(b"request").unwrap(); + // The client leaves before reading any response, as the build client does. + drop(client); + assert_eq!(read_to_end(&mut upstream), b"request"); + // Every later provider send succeeds: the relay reads and discards it. + upstream + .set_write_timeout(Some(Duration::from_secs(10))) + .unwrap(); + let chunk = vec![7_u8; 256 * 1024]; + for _ in 0..16 { + upstream.write_all(&chunk).unwrap(); + } + // Later still, the relay has not closed the provider side: sends keep succeeding. + std::thread::sleep(Duration::from_millis(200)); + upstream.write_all(&chunk).unwrap(); + // Only after the provider ends its side does the relay close. + upstream.shutdown(Shutdown::Write).unwrap(); + assert!(read_to_end(&mut upstream).is_empty()); + let report = relay.finish(FINISH_GRACE); + assert_eq!(report.discarded_bytes, 17 * 256 * 1024); + assert_eq!(report.forced, 0); +} + +#[test] +fn finish_closes_connections_the_provider_never_ends_after_its_bound() { + let provider = Provider::new(); + let relay = ClientRelay::start(&provider.path()).unwrap(); + let client = UnixStream::connect(relay.socket()).unwrap(); + let mut upstream = provider.accept(); + drop(client); + assert!(read_to_end(&mut upstream).is_empty()); + let started = Instant::now(); + let report = relay.finish(Duration::from_millis(300)); + assert!(started.elapsed() < Duration::from_secs(5)); + assert_eq!(report.forced, 1); + // The forced close reaches the provider. + assert!(read_to_end(&mut upstream).is_empty()); +} + +#[test] +fn a_refused_provider_connection_ends_only_that_client_connection() { + let provider = Provider::new(); + let path = provider.path(); + drop(provider); + let relay = ClientRelay::start(&path).unwrap(); + let mut client = UnixStream::connect(relay.socket()).unwrap(); + // The relay may already have closed this connection (its provider connect failed at + // once); macOS then refuses socket options with EINVAL. That is the outcome under + // test, and the read below returns at once. + match client.set_read_timeout(Some(Duration::from_secs(10))) { + Ok(()) => {} + Err(error) if error.raw_os_error() == Some(libc::EINVAL) => {} + Err(error) => panic!("unexpected socket option failure: {error}"), + } + let mut bytes = Vec::new(); + // The client sees the connection end, as with an absent engine socket. + let _ = client.read_to_end(&mut bytes); + assert!(bytes.is_empty()); + let report = relay.finish(FINISH_GRACE); + assert_eq!((report.connections, report.refused), (1, 1)); +} + +#[test] +fn the_listening_socket_is_short_private_and_removed_when_the_relay_finishes() { + let relay = ClientRelay::start(Path::new("/nonexistent/docker.sock")).unwrap(); + let socket = relay.socket().to_path_buf(); + // sockaddr_un holds 104 bytes on macOS, including the terminator. + assert!(socket.as_os_str().len() < 104); + let directory = socket.parent().unwrap().to_path_buf(); + let metadata = fs::symlink_metadata(&directory).unwrap(); + assert_eq!(metadata.mode() & 0o777, 0o700); + // SAFETY: geteuid has no preconditions. + assert_eq!(metadata.uid(), unsafe { libc::geteuid() }); + assert!( + fs::symlink_metadata(&socket) + .unwrap() + .file_type() + .is_socket() + ); + assert_eq!(relay.finish(FINISH_GRACE), RelayReport::default()); + assert!(!directory.exists()); +} + +#[test] +fn a_relay_left_by_a_killed_process_is_removed_and_a_live_one_is_kept() { + // A process that has exited: its PID names the abandoned directory. + let mut exited = std::process::Command::new("/usr/bin/true").spawn().unwrap(); + let dead = exited.id(); + exited.wait().unwrap(); + let mut running = std::process::Command::new("/bin/sleep") + .arg("30") + .spawn() + .unwrap(); + // Fixtures live under this test's own root, which no other relay sweeps (concurrent + // tests sweep the temporary directory itself), so only the relay started below can + // remove or keep them. + let root = FixtureRoot::new(); + let make = |pid: u32, other: bool| { + let directory = root.0.join(format!("{DIRECTORY_PREFIX}{pid}-0123456789ab")); + fs::DirBuilder::new() + .mode(0o700) + .create(&directory) + .unwrap(); + if other { + fs::write(directory.join("other"), "x").unwrap(); + } + drop(UnixListener::bind(directory.join(SOCKET_NAME)).unwrap()); + directory + }; + let abandoned = make(dead, false); + let live = make(running.id(), false); + // A directory holding anything besides the socket is never removed. + let kept = make(dead + 1_000_000, true); + assert!(abandoned.join(SOCKET_NAME).exists()); + let relay = ClientRelay::start_in(&root.0, Path::new("/nonexistent/docker.sock")).unwrap(); + assert!(relay.socket().starts_with(&root.0)); + assert!(!abandoned.exists()); + assert!(live.join(SOCKET_NAME).exists()); + assert!(kept.join("other").exists()); + relay.finish(FINISH_GRACE); + running.kill().unwrap(); + running.wait().unwrap(); +} + +/// A fresh private directory for one test's relay fixtures, removed when dropped. Its name +/// does not carry the relay prefix, so no relay sweep ever matches it. +struct FixtureRoot(PathBuf); +impl FixtureRoot { + fn new() -> Self { + let mut bytes = [0_u8; 6]; + File::open("/dev/urandom") + .and_then(|mut random| random.read_exact(&mut bytes)) + .unwrap(); + let name: String = bytes.iter().map(|byte| format!("{byte:02x}")).collect(); + let root = temporary_root().join(format!("hkt-{}-{name}", std::process::id())); + fs::DirBuilder::new().mode(0o700).create(&root).unwrap(); + Self(root) + } +} +impl Drop for FixtureRoot { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +/// A peer that takes a little at a time and then reports a full buffer, checking that the +/// bytes arrive in order. +struct SlowPeer { + received: u64, + calls: u64, +} +impl Write for SlowPeer { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.calls += 1; + if self.calls % 3 == 0 { + return Err(ErrorKind::WouldBlock.into()); + } + let count = bytes.len().min(1500); + for byte in &bytes[..count] { + assert_eq!(*byte, (self.received % 251) as u8, "bytes out of order"); + self.received += 1; + } + Ok(count) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +#[test] +fn pending_storage_follows_the_queued_bytes_not_the_whole_transfer() { + // Sustained backpressure: the queue is topped up to its limit while a slow peer + // takes partial writes, for 128 times the limit in total. + let total = 128 * TO_PROVIDER_LIMIT as u64; + let mut pending = Pending::default(); + let mut peer = SlowPeer { + received: 0, + calls: 0, + }; + let mut sent = 0_u64; + let mut largest = 0; + let mut chunk = vec![0_u8; CHUNK]; + while peer.received < total { + if pending.len() < TO_PROVIDER_LIMIT && sent < total { + let count = CHUNK.min((total - sent) as usize); + for (offset, byte) in chunk[..count].iter_mut().enumerate() { + *byte = ((sent + offset as u64) % 251) as u8; + } + pending.push(&chunk[..count]); + sent += count as u64; + } + pending.flush(&mut peer).unwrap(); + largest = largest.max(pending.bytes.capacity()); + } + assert!(pending.is_empty()); + assert_eq!(peer.received, total); + // Delivered bytes are dropped: storage stays within a small multiple of the limit. + assert!( + largest <= 4 * (TO_PROVIDER_LIMIT + CHUNK), + "buffer grew to {largest} bytes for a {TO_PROVIDER_LIMIT}-byte limit" + ); +} diff --git a/packages/runtime-core/src/provider/lifecycle.rs b/packages/runtime-core/src/provider/lifecycle.rs index 2b0f272d7..d48908d6b 100644 --- a/packages/runtime-core/src/provider/lifecycle.rs +++ b/packages/runtime-core/src/provider/lifecycle.rs @@ -1384,7 +1384,7 @@ fn start_pool( // Disk rejection must not keep an allocation alive. Process authority is // independent of disk adoption; this is an unclean stop, never reconciliation. let process = owner.process.as_ref().expect("process just recorded"); - let cleanup = stop_failed_boot(process, &binary(candidate)); + let cleanup = stop_verified_provider(process, &binary(candidate)); state::write( &root(candidate).join("failed-boot.json"), &json!({ @@ -1429,7 +1429,8 @@ fn start_machine( .map(drop) } -fn stop_failed_boot( +/// SIGTERM a provider whose identity is verified at the signal, then wait for it to exit. +fn stop_verified_provider( process: &identity::ProcessIdentity, binary: &Path, ) -> Result<(), CandidateError> { @@ -1442,7 +1443,7 @@ fn stop_failed_boot( if Instant::now() >= deadline { return Err(CandidateError::new( "stop_uncertain", - "Verified failed-boot provider did not stop after SIGTERM.", + "Verified provider did not stop after SIGTERM.", )); } std::thread::sleep(Duration::from_millis(50)); @@ -1851,6 +1852,90 @@ pub fn down(candidate: &Candidate) -> Result { return status(candidate); } verify_live(candidate, &owner)?; + match guest_stop(candidate, &mut owner) { + Ok(()) => status(candidate), + Err(error) + if error.code == "agent_protocol_uncertain" && agent_silent(candidate, &owner)? => + { + stop_silent_guest(candidate, &mut owner)?; + status(candidate) + } + Err(error) => Err(error), + } +} + +/// How many pings, a few seconds apart, a guest agent must leave unanswered before +/// `down` or `recover` treats it as stopped responding. +const SILENT_PINGS: u32 = 3; +#[cfg(not(test))] +const SILENT_PING_TIMEOUT: Duration = Duration::from_secs(3); +#[cfg(not(test))] +const SILENT_PING_INTERVAL: Duration = Duration::from_secs(2); +#[cfg(test)] +const SILENT_PING_TIMEOUT: Duration = Duration::from_millis(200); +#[cfg(test)] +const SILENT_PING_INTERVAL: Duration = Duration::from_millis(50); + +/// `true` when the owned guest agent answered none of [`SILENT_PINGS`] pings. +fn agent_silent(candidate: &Candidate, owner: &Owner) -> Result { + Ok(agent_silent_at(&socket(candidate, owner, "agent.sock")?)) +} + +/// Any reply, even an unexpected one, means the guest is alive and keeps the normal stop +/// path. +fn agent_silent_at(path: &Path) -> bool { + for attempt in 0..SILENT_PINGS { + if attempt > 0 { + std::thread::sleep(SILENT_PING_INTERVAL); + } + if agent::answers_ping(path, SILENT_PING_TIMEOUT) { + return false; + } + } + true +} + +/// Stop the verified provider of a guest whose agent stopped responding. Nothing in the +/// guest is quiesced or flushed, so the pool is recorded `recovered-unclean`, as after a +/// provider exit; the next start checks its retained disks as usual. The signal reaches +/// only this home's recorded provider, rechecked by PID, start time, user and executable +/// immediately before it is sent. A provider that already exited is only recorded. +fn stop_silent_guest(candidate: &Candidate, owner: &mut Owner) -> Result<(), CandidateError> { + // Checked again at the effect: the caller holds the operation lease, but a running + // build holds only its shared fence while its client runs. + require_no_build_for_silent_stop(candidate)?; + let process = owner.process.clone().ok_or_else(|| { + CandidateError::new("process_identity_unavailable", "No recorded provider.") + })?; + if identity::alive(process.pid)? { + verify_live(candidate, owner)?; + stop_verified_provider(&process, &binary(candidate))?; + } + finish_stopped(candidate, owner, "recovered-unclean") +} + +/// Refuse to stop a silent guest while a native image build is admitted. A build releases +/// the operation lease while its client runs and keeps only the shared build fence +/// (`run/image-builds/active.lock`), so the lease alone does not prove the engine is idle. +/// The fence is a `flock`: it is released when the build's processes exit, including on +/// SIGKILL, so a refusal lasts only while a build process is alive. Callers hold the +/// operation lease, under which no new build can take the fence. +fn require_no_build_for_silent_stop(candidate: &Candidate) -> Result<(), CandidateError> { + super::image_build::require_no_active_build(candidate).map_err(|error| { + if error.code == "provider_busy" { + CandidateError::new( + "provider_busy", + "A native image build is running against this guest. Finish or cancel it first (Ctrl-C the `hack up` that started it); the next attempt then stops the guest that stopped responding.", + ) + } else { + error + } + }) +} + +/// `down`'s guest-facing part: quiesce the engine, ask the guest to flush and shut down, +/// stop the provider and record the stop. +fn guest_stop(candidate: &Candidate, owner: &mut Owner) -> Result<(), CandidateError> { let before_engine = engine_was_not_started( &owner.phase, owner.guest_boot_id.as_deref(), @@ -1862,7 +1947,7 @@ pub fn down(candidate: &Candidate) -> Result { if failed_engine { guest( candidate, - &owner, + owner, r#"set -eu test "$(cat /storage/.hack-local-owner)" = "$1" test "$(cat /proc/sys/kernel/random/boot_id)" = "$2" @@ -1895,10 +1980,10 @@ sync"#, } else { "initial" }; - phase(candidate, &mut owner, "quiescing")?; + phase(candidate, owner, "quiescing")?; let quiescence = guest( candidate, - &owner, + owner, include_str!("guest-quiesce.sh"), &[ &owner.token, @@ -1917,9 +2002,9 @@ sync"#, }), )?; } - phase(candidate, &mut owner, "stopping")?; + phase(candidate, owner, "stopping")?; let response = agent::request( - &socket(candidate, &owner, "agent.sock")?, + &socket(candidate, owner, "agent.sock")?, json!({"method":"shutdown"}), Duration::from_secs(15), )?; @@ -1929,7 +2014,7 @@ sync"#, "Guest did not acknowledge flush/unmount.", )); } - verify_live(candidate, &owner)?; + verify_live(candidate, owner)?; let process = owner.process.as_ref().expect("live identity verified"); // Revalidate immediately before signalling. macOS has no Linux pidfd equivalent; // this is cooperative same-user ownership, not hostile-process containment. @@ -1945,13 +2030,13 @@ sync"#, )); } if before_engine { - finish_stopped(candidate, &mut owner, "stopped-before-engine")?; + finish_stopped(candidate, owner, "stopped-before-engine")?; } else if failed_engine { - finish_stopped(candidate, &mut owner, "stopped-after-engine-failure")?; + finish_stopped(candidate, owner, "stopped-after-engine-failure")?; } else { - finish_quiesced_stop(candidate, &mut owner)?; + finish_quiesced_stop(candidate, owner)?; } - status(candidate) + Ok(()) } /// `down`'s completion after the guest acknowledged flush and shutdown of a quiesced engine @@ -2151,10 +2236,20 @@ pub fn recover(candidate: &Candidate) -> Result { return status(candidate); }; if identity::alive(process.pid)? { - return Err(CandidateError::new( - "recovery_required", - "Recovery only accepts a confirmed dead recorded process. Use down for a verified live guest.", - )); + // A live provider is stopped here only when its identity verifies and its guest + // agent stopped responding; a responsive guest is stopped cleanly by `down`. + // Refuse first, without probing, while a build is admitted. + require_no_build_for_silent_stop(candidate)?; + verify_live(candidate, &owner)?; + if !agent_silent(candidate, &owner)? { + return Err(CandidateError::new( + "recovery_required", + "Recovery only accepts a confirmed dead recorded process or a guest that stopped responding. Use down for a responsive guest.", + )); + } + // The operation lease taken above stays held through the stop and its receipt. + stop_silent_guest(candidate, &mut owner)?; + return status(candidate); } finish_stopped(candidate, &mut owner, "recovered-unclean")?; status(candidate) @@ -2721,6 +2816,112 @@ printf 'verified-cache\n' ); } + /// A short private directory for a fake agent socket (`sockaddr_un` is small). + struct ShortDirectory(std::path::PathBuf); + impl ShortDirectory { + fn new() -> Self { + let mut bytes = [0_u8; 6]; + File::open("/dev/urandom") + .unwrap() + .read_exact(&mut bytes) + .unwrap(); + let name: String = bytes.iter().map(|b| format!("{b:02x}")).collect(); + let path = Path::new("/tmp").join(format!("hka-{name}")); + fs::create_dir(&path).unwrap(); + Self(path) + } + } + impl Drop for ShortDirectory { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } + } + + #[test] + fn only_a_guest_agent_that_never_starts_a_reply_counts_as_silent() { + use std::os::unix::net::UnixListener; + let directory = ShortDirectory::new(); + let path = directory.0.join("agent.sock"); + // No listener: refused connections are silence. + assert!(agent_silent_at(&path)); + let listener = UnixListener::bind(&path).unwrap(); + // Accepted but never answered, as when the guest's vsock path is saturated. + let held = std::thread::spawn(move || { + // Keep every connection open and unanswered until the probe is over. + let kept: Vec<_> = listener + .incoming() + .take(SILENT_PINGS as usize) + .map(Result::unwrap) + .collect(); + kept.len() + }); + let started = Instant::now(); + assert!(agent_silent_at(&path)); + // Each probe waits out its deadline (the last sees the fake end the connection). + assert!(started.elapsed() >= SILENT_PING_TIMEOUT * (SILENT_PINGS - 1)); + assert_eq!(held.join().unwrap(), SILENT_PINGS as usize); + fs::remove_file(&path).unwrap(); + // Any reply byte, even of an unexpected response, is an answer. + let listener = UnixListener::bind(&path).unwrap(); + let answering = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + stream.write_all(b"?").unwrap(); + }); + assert!(!agent_silent_at(&path)); + answering.join().unwrap(); + fs::remove_file(&path).unwrap(); + // A connection the guest side ends before replying is silence. + let listener = UnixListener::bind(&path).unwrap(); + let closing = std::thread::spawn(move || { + for stream in listener.incoming().take(SILENT_PINGS as usize) { + drop(stream.unwrap()); + } + }); + assert!(agent_silent_at(&path)); + closing.join().unwrap(); + } + + #[cfg(target_os = "macos")] + #[test] + fn a_silent_guest_is_not_stopped_under_an_admitted_build() { + let root = crate::test_root::private_test_root( + &std::env::temp_dir().canonicalize().unwrap(), + "hack-silent-stop-build", + ); + let checkout = root.join("checkout"); + state::private_directory(&checkout).unwrap(); + let candidate = Candidate::discover(&checkout).unwrap(); + // A live process stands in for the recorded provider; it must not be signalled. + let mut provider = std::process::Command::new("/bin/sleep") + .arg("30") + .spawn() + .unwrap(); + let mut owner: Owner = serde_json::from_value(json!({ + "version":1,"checkout":"/fixture","token":"fixture","machine":"fixture", + "short_home":"/fixture","created":true,"phase":"running","process":null, + "storage":null,"overlay":null,"guest_boot_id":null, + "daemon_pid":null,"daemon_start":null,"rootfs_digest":null + })) + .unwrap(); + owner.process = Some(identity::observe(provider.id() as i32).unwrap()); + // An admitted build holds the shared fence, as `image_build::run` does. + let fence = super::super::image_build::hold_build_fence_for_test(&candidate); + let refused = stop_silent_guest(&candidate, &mut owner).unwrap_err(); + assert_eq!(refused.code, "provider_busy"); + assert!(refused.message.contains("Ctrl-C")); + assert!( + provider.try_wait().unwrap().is_none(), + "provider was signalled" + ); + assert_eq!(owner.phase, "running"); + drop(fence); + // Without a build, the check passes and the stop proceeds to identity checks. + assert!(require_no_build_for_silent_stop(&candidate).is_ok()); + provider.kill().unwrap(); + provider.wait().unwrap(); + let _ = fs::remove_dir_all(&root); + } + #[cfg(target_os = "macos")] #[test] fn failed_boot_cleanup_stops_only_the_verified_process() { @@ -2734,7 +2935,7 @@ printf 'verified-cache\n' .unwrap(); let observed = identity::observe(child.id() as i32).unwrap(); let reaper = std::thread::spawn(move || child.wait().unwrap()); - stop_failed_boot(&observed, &observed.executable).unwrap(); + stop_verified_provider(&observed, &observed.executable).unwrap(); reaper.join().unwrap(); assert!(!identity::alive(observed.pid).unwrap()); assert!(sentinel.try_wait().unwrap().is_none()); diff --git a/packages/runtime-core/src/provider/lifecycle/interrupted.rs b/packages/runtime-core/src/provider/lifecycle/interrupted.rs index c3bf4f4e4..84676a909 100644 --- a/packages/runtime-core/src/provider/lifecycle/interrupted.rs +++ b/packages/runtime-core/src/provider/lifecycle/interrupted.rs @@ -25,7 +25,8 @@ //! Nothing unowned is adopted, stopped or removed. use super::super::{config_audit, disk_template::Stage, identity, state}; use super::{ - Owner, binary, finish_absent, invoke, recorded_process, stop_failed_boot, verify_templates, + Owner, binary, finish_absent, invoke, recorded_process, stop_verified_provider, + verify_templates, }; use crate::{Candidate, CandidateError}; use std::fs; @@ -135,7 +136,7 @@ fn recover_start( &binary(candidate), unsafe { libc::geteuid() }, )?; - stop_failed_boot(&process, &binary(candidate))?; + stop_verified_provider(&process, &binary(candidate))?; } // The record may be stale (a previous boot's) while an orphaned start is still running. provider_idle(candidate)?;