From f38751ceaeca59bbd8c654bfc3986194cc0ff900 Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy <344842+roodboi@users.noreply.github.com> Date: Fri, 6 Nov 2026 14:26:03 -0500 Subject: [PATCH 1/7] fix(native): relay the build client so the VM stays responsive The pinned provider forwards the guest engine socket through its vsock muxer. On macOS, when a host client closes a forwarded connection while the guest is still sending, the muxer re-arms the dead stream with write interest only and then fails to delete it: one kevent change list stops at the unregistered read filter. The socket reports a hang-up on every wait, each one queues a reset to the guest and restarts the stream's reaper delay, so the stream is never removed. Guest vCPUs and the shared vsock queue stay saturated and later agent and engine requests time out. The BuildKit client closes many connections before reading the whole response and the rest at exit, so most builds reached that state. The build client now reaches the engine through a private relay that always drains guest bytes (discarding them once the client is gone), forwards a client's end of stream only as a half-close, and closes a provider connection only after the guest ended it, within a bounded wait. Refs HACK-1924 Co-Authored-By: Claude Opus 5.5 --- packages/runtime-core/README.md | 7 +- .../runtime-core/src/provider/image_build.rs | 25 +- .../src/provider/image_build/client_relay.rs | 536 ++++++++++++++++++ .../image_build/client_relay/tests.rs | 154 +++++ 4 files changed, 714 insertions(+), 8 deletions(-) create mode 100644 packages/runtime-core/src/provider/image_build/client_relay.rs create mode 100644 packages/runtime-core/src/provider/image_build/client_relay/tests.rs diff --git a/packages/runtime-core/README.md b/packages/runtime-core/README.md index 3f650f695..ba644c1a6 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 diff --git a/packages/runtime-core/src/provider/image_build.rs b/packages/runtime-core/src/provider/image_build.rs index 371484822..0db8ea8e3 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) => { @@ -1101,7 +1115,7 @@ fn command( scratch: &Path, metadata: &Path, inputs: &Inputs, - identity: &PoolIdentity, + engine: &Path, pins: &BTreeMap, labels: &BTreeMap<&'static str, String>, ) -> Result { @@ -1128,10 +1142,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..c536af042 --- /dev/null +++ b/packages/runtime-core/src/provider/image_build/client_relay.rs @@ -0,0 +1,536 @@ +//! 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 { + let directory = private_directory()?; + 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) +} + +/// A new mode-0700 directory with a short path, owned by this user. +fn private_directory() -> Result { + let parent = if cfg!(target_os = "macos") { + Path::new("/private/tmp") + } else { + Path::new("/tmp") + }; + 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!("hkb-{name}")); + 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.")) +} + +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. +#[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]) { + if self.start == self.bytes.len() { + self.bytes.clear(); + 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 UnixStream) -> 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(()), + } + } + 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..4fa1681d3 --- /dev/null +++ b/packages/runtime-core/src/provider/image_build/client_relay/tests.rs @@ -0,0 +1,154 @@ +//! 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().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(); + client + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + 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()); +} From ffe2b3fd5fa1bb41975abed77bfa204e89a85e92 Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy <344842+roodboi@users.noreply.github.com> Date: Fri, 6 Nov 2026 14:26:15 -0500 Subject: [PATCH 2/7] fix(native): stop a VM whose guest agent stopped responding `runtime down` failed with agent_protocol_uncertain and `runtime recover` refused while the provider process was alive, so a guest whose agent stopped answering could only be stopped by signalling the provider by hand. When a guest step of `down` fails that way, or `recover` finds the recorded provider alive, both now ping the agent three times over about 13 seconds. If no ping starts a reply, they stop this home's recorded provider after verifying its PID, start time, user and executable, with no guest quiescence, and record the pool `recovered-unclean`, which the next start accepts after its usual disk checks. A guest that answers keeps the clean stop, and `recover` still refuses it. Refs HACK-1924 Co-Authored-By: Claude Opus 5.5 --- packages/runtime-core/README.md | 4 + packages/runtime-core/src/provider/agent.rs | 53 ++++++ .../runtime-core/src/provider/lifecycle.rs | 172 ++++++++++++++++-- .../src/provider/lifecycle/interrupted.rs | 5 +- 4 files changed, 214 insertions(+), 20 deletions(-) diff --git a/packages/runtime-core/README.md b/packages/runtime-core/README.md index ba644c1a6..89a35ec40 100644 --- a/packages/runtime-core/README.md +++ b/packages/runtime-core/README.md @@ -505,6 +505,10 @@ 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. 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/lifecycle.rs b/packages/runtime-core/src/provider/lifecycle.rs index 2b0f272d7..dc2e1db61 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,68 @@ 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> { + 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") +} + +/// `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 +1925,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 +1958,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 +1980,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 +1992,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 +2008,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 +2214,18 @@ 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`. + 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 +2792,71 @@ 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 failed_boot_cleanup_stops_only_the_verified_process() { @@ -2734,7 +2870,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)?; From 5643b8c9c34ebc7d47971478185183a7761341e6 Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy <344842+roodboi@users.noreply.github.com> Date: Fri, 6 Nov 2026 15:16:16 -0500 Subject: [PATCH 3/7] fix(native): remove build relay sockets a killed process left behind A build whose process is killed cannot remove its private relay socket directory. Relay directories now carry the creating PID, and each new relay removes those whose process no longer exists, but only a mode-0700 directory owned by this user that holds nothing except the relay socket. Refs HACK-1924 Co-Authored-By: Claude Opus 5.5 --- .../src/provider/image_build/client_relay.rs | 61 +++++++++++++++++-- .../image_build/client_relay/tests.rs | 36 +++++++++++ 2 files changed, 92 insertions(+), 5 deletions(-) diff --git a/packages/runtime-core/src/provider/image_build/client_relay.rs b/packages/runtime-core/src/provider/image_build/client_relay.rs index c536af042..05cbdbfca 100644 --- a/packages/runtime-core/src/provider/image_build/client_relay.rs +++ b/packages/runtime-core/src/provider/image_build/client_relay.rs @@ -188,20 +188,28 @@ fn failure(message: &str) -> CandidateError { CandidateError::new("native_build_client", message) } -/// A new mode-0700 directory with a short path, owned by this user. -fn private_directory() -> Result { - let parent = if cfg!(target_os = "macos") { +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, 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() -> Result { + let parent = temporary_root(); + 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!("hkb-{name}")); + 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, @@ -223,6 +231,49 @@ fn private_directory() -> Result { 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() { 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 index 4fa1681d3..e04c1bb16 100644 --- a/packages/runtime-core/src/provider/image_build/client_relay/tests.rs +++ b/packages/runtime-core/src/provider/image_build/client_relay/tests.rs @@ -152,3 +152,39 @@ fn the_listening_socket_is_short_private_and_removed_when_the_relay_finishes() { 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(); + let make = |pid: u32| { + let directory = temporary_root().join(format!("{DIRECTORY_PREFIX}{pid}-0123456789ab")); + fs::DirBuilder::new() + .mode(0o700) + .create(&directory) + .unwrap(); + drop(UnixListener::bind(directory.join(SOCKET_NAME)).unwrap()); + directory + }; + let abandoned = make(dead); + let live = make(running.id()); + // A directory holding anything besides the socket is never removed. + let kept = make(dead + 1_000_000); + fs::write(kept.join("other"), "x").unwrap(); + let relay = ClientRelay::start(Path::new("/nonexistent/docker.sock")).unwrap(); + assert!(!abandoned.exists()); + assert!(live.join(SOCKET_NAME).exists()); + assert!(kept.join("other").exists()); + relay.finish(FINISH_GRACE); + running.kill().unwrap(); + running.wait().unwrap(); + for directory in [live, kept] { + let _ = fs::remove_dir_all(directory); + } +} From 66d25b8381e1f18cfe54d115a56615b06820effe Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy <344842+roodboi@users.noreply.github.com> Date: Fri, 6 Nov 2026 15:19:01 -0500 Subject: [PATCH 4/7] fix(native): bound build relay buffers by pending bytes Under sustained backpressure, partial writes advanced the start offset but the delivered prefix stayed in the buffer until the queue drained completely, so the limits bounded only the unread suffix while storage grew with the whole transfer (128 MiB for a 1 MiB limit in the new test). The delivered prefix is now dropped once it reaches half the buffer, keeping storage within about twice the pending bytes plus one read at linear cost. Refs HACK-1924 Co-Authored-By: Claude Opus 5.5 --- .../src/provider/image_build/client_relay.rs | 15 +++-- .../image_build/client_relay/tests.rs | 58 +++++++++++++++++++ 2 files changed, 69 insertions(+), 4 deletions(-) diff --git a/packages/runtime-core/src/provider/image_build/client_relay.rs b/packages/runtime-core/src/provider/image_build/client_relay.rs index 05cbdbfca..564f38761 100644 --- a/packages/runtime-core/src/provider/image_build/client_relay.rs +++ b/packages/runtime-core/src/provider/image_build/client_relay.rs @@ -322,7 +322,9 @@ fn connect(provider: &Path) -> std::io::Result { Ok(stream) } -/// Pending bytes for one direction. +/// 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, @@ -337,8 +339,10 @@ impl Pending { self.len() == 0 } fn push(&mut self, bytes: &[u8]) { - if self.start == self.bytes.len() { - self.bytes.clear(); + // 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); @@ -348,7 +352,7 @@ impl Pending { self.start = 0; } /// Write what `stream` accepts now. `Err` means the peer can take no more. - fn flush(&mut self, stream: &mut UnixStream) -> Result<(), ()> { + fn flush(&mut self, stream: &mut impl Write) -> Result<(), ()> { while !self.is_empty() { match stream.write(&self.bytes[self.start..]) { Ok(0) => return Err(()), @@ -358,6 +362,9 @@ impl Pending { Err(_) => return Err(()), } } + if self.start == self.bytes.len() { + self.clear(); + } Ok(()) } } 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 index e04c1bb16..2cc86a6fa 100644 --- a/packages/runtime-core/src/provider/image_build/client_relay/tests.rs +++ b/packages/runtime-core/src/provider/image_build/client_relay/tests.rs @@ -188,3 +188,61 @@ fn a_relay_left_by_a_killed_process_is_removed_and_a_live_one_is_kept() { let _ = fs::remove_dir_all(directory); } } + +/// 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" + ); +} From a86b3cf718420ef3c7f8480b290cc8f59db8cd7e Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy <344842+roodboi@users.noreply.github.com> Date: Fri, 6 Nov 2026 15:33:39 -0500 Subject: [PATCH 5/7] fix(native): refuse a silent-guest stop while a build is admitted A build releases the operation lease while its client runs and keeps only the shared build fence, so the lease alone did not prove that no build client was using the engine. `runtime recover` could stop the provider of a silent guest underneath a live build client. Both silent-guest paths now check the fence under the operation lease (`recover` before probing, and both again just before signalling) and refuse with provider_busy, telling the user to finish or cancel the build first; the next attempt then stops the silent guest. The fence is a flock released when the build's processes exit, including on SIGKILL, as a_killed_coordinator_leaves_its_client_fenced already proves. Also writes the abandoned-relay fixture's extra file before its socket, so a relay starting in a concurrent test cannot remove it first. Refs HACK-1924 Co-Authored-By: Claude Opus 5.5 --- packages/runtime-core/README.md | 4 +- .../runtime-core/src/provider/image_build.rs | 13 ++++ .../image_build/client_relay/tests.rs | 14 ++-- .../runtime-core/src/provider/lifecycle.rs | 65 +++++++++++++++++++ 4 files changed, 90 insertions(+), 6 deletions(-) diff --git a/packages/runtime-core/README.md b/packages/runtime-core/README.md index 89a35ec40..1af11cad9 100644 --- a/packages/runtime-core/README.md +++ b/packages/runtime-core/README.md @@ -508,7 +508,9 @@ retain their phase and receipts. Recovery preserves disks and labels an unclean 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. +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/image_build.rs b/packages/runtime-core/src/provider/image_build.rs index 0db8ea8e3..2cc9c1ece 100644 --- a/packages/runtime-core/src/provider/image_build.rs +++ b/packages/runtime-core/src/provider/image_build.rs @@ -987,6 +987,19 @@ pub(super) fn run( }) } +/// Hold the shared build fence as an admitted build does, for lifecycle tests. +#[cfg(test)] +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(|| { 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 index 2cc86a6fa..79edcaa26 100644 --- a/packages/runtime-core/src/provider/image_build/client_relay/tests.rs +++ b/packages/runtime-core/src/provider/image_build/client_relay/tests.rs @@ -163,20 +163,24 @@ fn a_relay_left_by_a_killed_process_is_removed_and_a_live_one_is_kept() { .arg("30") .spawn() .unwrap(); - let make = |pid: u32| { + // `other` is written before the socket, so a relay starting in a concurrent test never + // sees that directory holding only a socket. + let make = |pid: u32, other: bool| { let directory = temporary_root().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); - let live = make(running.id()); + 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); - fs::write(kept.join("other"), "x").unwrap(); + let kept = make(dead + 1_000_000, true); let relay = ClientRelay::start(Path::new("/nonexistent/docker.sock")).unwrap(); assert!(!abandoned.exists()); assert!(live.join(SOCKET_NAME).exists()); diff --git a/packages/runtime-core/src/provider/lifecycle.rs b/packages/runtime-core/src/provider/lifecycle.rs index dc2e1db61..d48908d6b 100644 --- a/packages/runtime-core/src/provider/lifecycle.rs +++ b/packages/runtime-core/src/provider/lifecycle.rs @@ -1901,6 +1901,9 @@ fn agent_silent_at(path: &Path) -> bool { /// 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.") })?; @@ -1911,6 +1914,25 @@ fn stop_silent_guest(candidate: &Candidate, owner: &mut Owner) -> Result<(), Can 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> { @@ -2216,6 +2238,8 @@ pub fn recover(candidate: &Candidate) -> Result { if identity::alive(process.pid)? { // 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( @@ -2857,6 +2881,47 @@ printf 'verified-cache\n' 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() { From f4fd82761ba02a415da204691d1d4378c555b493 Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy <344842+roodboi@users.noreply.github.com> Date: Fri, 6 Nov 2026 15:45:53 -0500 Subject: [PATCH 6/7] fix(native): gate the build fence test helper to macOS Its only caller is a macOS-only lifecycle test, so Linux test builds flagged the helper as dead code and failed the candidate-core lint. Refs HACK-1924 Co-Authored-By: Claude Opus 5.5 --- packages/runtime-core/src/provider/image_build.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/runtime-core/src/provider/image_build.rs b/packages/runtime-core/src/provider/image_build.rs index 2cc9c1ece..38a781259 100644 --- a/packages/runtime-core/src/provider/image_build.rs +++ b/packages/runtime-core/src/provider/image_build.rs @@ -987,8 +987,9 @@ pub(super) fn run( }) } -/// Hold the shared build fence as an admitted build does, for lifecycle tests. -#[cfg(test)] +/// 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(); From a0e3e797b182edc67e1f66046943e0e84fede96a Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy <344842+roodboi@users.noreply.github.com> Date: Fri, 6 Nov 2026 19:35:24 -0500 Subject: [PATCH 7/7] test(native): make the build client relay tests independent of concurrent relays The abandoned-relay test built its fixtures in the shared temporary directory, where any concurrently starting relay's sweep could remove a fixture between its creation and its socket bind. Fixtures now live under a per-test root that no other relay sweeps, and the relay under test starts there (ClientRelay::start_in, which start uses with the temporary directory). The test also checks that the abandoned fixture existed before the sweep. The refused-provider test treats EINVAL from setting the read timeout as the closed connection it expects: macOS refuses socket options once the relay has already closed the connection. Co-Authored-By: Claude Opus 5.5 --- .../src/provider/image_build/client_relay.rs | 16 +++++-- .../image_build/client_relay/tests.rs | 47 +++++++++++++++---- 2 files changed, 48 insertions(+), 15 deletions(-) diff --git a/packages/runtime-core/src/provider/image_build/client_relay.rs b/packages/runtime-core/src/provider/image_build/client_relay.rs index 564f38761..446d42897 100644 --- a/packages/runtime-core/src/provider/image_build/client_relay.rs +++ b/packages/runtime-core/src/provider/image_build/client_relay.rs @@ -102,7 +102,13 @@ impl ClientRelay { /// Connecting to `provider` happens per client connection, so starting never touches /// the engine. pub(super) fn start(provider: &Path) -> Result { - let directory = private_directory()?; + 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), @@ -198,10 +204,10 @@ fn temporary_root() -> &'static Path { } } -/// A new mode-0700 directory with a short path, 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() -> Result { - let parent = temporary_root(); +/// 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]; 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 index 79edcaa26..9c6acafd0 100644 --- a/packages/runtime-core/src/provider/image_build/client_relay/tests.rs +++ b/packages/runtime-core/src/provider/image_build/client_relay/tests.rs @@ -11,7 +11,7 @@ struct Provider { } impl Provider { fn new() -> Self { - let directory = private_directory().unwrap(); + let directory = private_directory(temporary_root()).unwrap(); let listener = UnixListener::bind(directory.join("provider.sock")).unwrap(); Self { directory, @@ -121,9 +121,14 @@ fn a_refused_provider_connection_ends_only_that_client_connection() { drop(provider); let relay = ClientRelay::start(&path).unwrap(); let mut client = UnixStream::connect(relay.socket()).unwrap(); - client - .set_read_timeout(Some(Duration::from_secs(10))) - .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); @@ -163,10 +168,12 @@ fn a_relay_left_by_a_killed_process_is_removed_and_a_live_one_is_kept() { .arg("30") .spawn() .unwrap(); - // `other` is written before the socket, so a relay starting in a concurrent test never - // sees that directory holding only a socket. + // 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 = temporary_root().join(format!("{DIRECTORY_PREFIX}{pid}-0123456789ab")); + let directory = root.0.join(format!("{DIRECTORY_PREFIX}{pid}-0123456789ab")); fs::DirBuilder::new() .mode(0o700) .create(&directory) @@ -181,15 +188,35 @@ fn a_relay_left_by_a_killed_process_is_removed_and_a_live_one_is_kept() { let live = make(running.id(), false); // A directory holding anything besides the socket is never removed. let kept = make(dead + 1_000_000, true); - let relay = ClientRelay::start(Path::new("/nonexistent/docker.sock")).unwrap(); + 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(); - for directory in [live, kept] { - let _ = fs::remove_dir_all(directory); +} + +/// 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); } }