From d53662647d3b472e766548fa4e6633e8eb34a863 Mon Sep 17 00:00:00 2001 From: Clovis Muneza Date: Fri, 9 Oct 2026 20:52:06 -0400 Subject: [PATCH 1/6] feat(platform,state): add monitor watch, paths, reservation, and log open ProcessExitWatch::with_exec reports a watched process's exec as well as its exit, without reaping it, and exposes its kqueue descriptor so a Tokio AsyncFd can await it. PvPaths gains run/m/<12 hex> monitor directories and their reservation locks, sharing the 12-digit SHA-256 digest worker admin sockets use. MonitorReservation takes the exclusive lock, hands the same locked file to a child as stdin, and lets that child adopt and verify it. open_private_log_file opens a runtime log without following symlinks and checks owner, type and link count on the opened descriptor. --- crates/platform/src/lib.rs | 2 +- crates/platform/src/process.rs | 9 +++ crates/platform/src/process/macos.rs | 82 +++++++++++++++++++++- crates/platform/src/process/unsupported.rs | 8 +++ crates/platform/tests/process_identity.rs | 60 +++++++++++++++- crates/state/src/fs.rs | 40 +++++++++++ crates/state/src/lib.rs | 4 +- crates/state/src/paths.rs | 30 +++++++- crates/state/src/update_lock.rs | 78 +++++++++++++++++++- 9 files changed, 303 insertions(+), 10 deletions(-) diff --git a/crates/platform/src/lib.rs b/crates/platform/src/lib.rs index fb87c928..8efca91d 100644 --- a/crates/platform/src/lib.rs +++ b/crates/platform/src/lib.rs @@ -41,7 +41,7 @@ pub use pf::{ }; pub use port::loopback_tcp_port_available; pub use process::{ - BootSessionId, ProcessExitWatch, ProcessIdentity, ProcessStartIdentity, + BootSessionId, ProcessEvents, ProcessExitWatch, ProcessIdentity, ProcessStartIdentity, current_boot_session_id, exec_replace, exec_replace_with_env, inspect_process_identity, inspect_process_start_identity, process_group_has_live_members, process_is_zombie, }; diff --git a/crates/platform/src/process.rs b/crates/platform/src/process.rs index 049c80e6..93a68fd6 100644 --- a/crates/platform/src/process.rs +++ b/crates/platform/src/process.rs @@ -38,6 +38,15 @@ pub struct ProcessStartIdentity { pub microseconds: u64, } +/// What a watched process did since [`ProcessExitWatch::try_events`] last collected its events. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct ProcessEvents { + /// The process replaced its program with `exec`. Its pid and birth identity are unchanged. + pub exec: bool, + /// The exact status of an exit the watch observed without reaping the process. + pub exit: Option, +} + /// Identifies one kernel boot, independently of wall-clock changes. #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] #[serde(try_from = "String", into = "String")] diff --git a/crates/platform/src/process/macos.rs b/crates/platform/src/process/macos.rs index a66e32d1..79738f18 100644 --- a/crates/platform/src/process/macos.rs +++ b/crates/platform/src/process/macos.rs @@ -1,5 +1,6 @@ use std::io; use std::mem::{self, MaybeUninit}; +use std::os::fd::{AsFd, AsRawFd, RawFd}; use std::os::unix::process::ExitStatusExt; use std::process::ExitStatus; use std::ptr; @@ -116,19 +117,98 @@ impl ProcessExitWatch { } Ok(Some(ExitStatus::from_raw(data))) } + + /// Watches `pid`'s exec as well as its exit, for a child that cannot exec until the watch + /// exists. Events accumulate until [`ProcessExitWatch::try_events`] collects them, and the + /// watch's descriptor becomes readable while any are pending. + pub fn with_exec(pid: u32) -> Result { + if pid == 0 || i32::try_from(pid).is_err() { + return Err(exit_watch_error(io::Error::new( + io::ErrorKind::InvalidInput, + "invalid process id", + ))); + } + let queue = Kqueue::new().map_err(exit_watch_error)?; + let change = process_event( + pid, + EvFlags::EV_ADD | EvFlags::EV_CLEAR, + FilterFlag::NOTE_EXEC | FilterFlag::NOTE_EXIT | FilterFlag::NOTE_EXITSTATUS, + ); + queue + .kevent(&[change], &mut [], Some(zero_timeout())) + .map_err(exit_watch_error)?; + Ok(Self { queue, pid }) + } + + /// Collects the exec and exit events of a watch made by [`ProcessExitWatch::with_exec`]. + pub fn try_events(&mut self) -> Result { + let mut events = [process_event( + self.pid, + EvFlags::empty(), + FilterFlag::empty(), + )]; + let count = self + .queue + .kevent(&[], &mut events, Some(zero_timeout())) + .map_err(exit_watch_error)?; + if count == 0 { + return Ok(super::ProcessEvents::default()); + } + let event = events[0]; + let data = i32::try_from(event.data()).map_err(exit_watch_error)?; + if event.flags().contains(EvFlags::EV_ERROR) { + return Err(exit_watch_error(io::Error::from_raw_os_error(data))); + } + let flags = event.fflags(); + let exit = if flags.contains(FilterFlag::NOTE_EXIT) { + if !flags.contains(FilterFlag::NOTE_EXITSTATUS) { + return Err(exit_watch_error(io::Error::other( + "exit event omitted process status", + ))); + } + Some(ExitStatus::from_raw(data)) + } else { + None + }; + Ok(super::ProcessEvents { + exec: flags.contains(FilterFlag::NOTE_EXEC), + exit, + }) + } +} + +impl AsRawFd for ProcessExitWatch { + fn as_raw_fd(&self) -> RawFd { + self.queue.as_fd().as_raw_fd() + } } fn exit_event(pid: u32, flags: EvFlags) -> KEvent { + process_event( + pid, + flags, + FilterFlag::NOTE_EXIT | FilterFlag::NOTE_EXITSTATUS, + ) +} + +fn process_event(pid: u32, flags: EvFlags, filter_flags: FilterFlag) -> KEvent { KEvent::new( pid as usize, EventFilter::EVFILT_PROC, flags, - FilterFlag::NOTE_EXIT | FilterFlag::NOTE_EXITSTATUS, + filter_flags, 0, 0, ) } +const fn zero_timeout() -> libc::timespec { + libc::timespec { + tv_sec: 0, + tv_nsec: 0, + } +} + fn exit_watch_error(source: impl std::error::Error + Send + Sync + 'static) -> PlatformError { PlatformError::ProcessIdentityInspection { source: Box::new(source), diff --git a/crates/platform/src/process/unsupported.rs b/crates/platform/src/process/unsupported.rs index 6ded0a27..458bf7b6 100644 --- a/crates/platform/src/process/unsupported.rs +++ b/crates/platform/src/process/unsupported.rs @@ -18,6 +18,14 @@ impl ProcessExitWatch { pub fn try_exit_status(&mut self) -> Result, PlatformError> { Err(unsupported(PlatformCapability::ProcessInspection)?) } + + pub fn with_exec(_pid: u32) -> Result { + Err(unsupported(PlatformCapability::ProcessInspection)?) + } + + pub fn try_events(&mut self) -> Result { + Err(unsupported(PlatformCapability::ProcessInspection)?) + } } pub(super) fn inspect_process_identity( diff --git a/crates/platform/tests/process_identity.rs b/crates/platform/tests/process_identity.rs index 9608b7c1..ddb35d9e 100644 --- a/crates/platform/tests/process_identity.rs +++ b/crates/platform/tests/process_identity.rs @@ -1,13 +1,18 @@ #![cfg(target_os = "macos")] -use std::process::Child; +use std::io::Write as _; +use std::os::unix::process::ExitStatusExt; +use std::process::{Child, Stdio}; use std::thread; use std::time::Duration; use anyhow::{Result, anyhow}; use camino::Utf8Path; use camino_tempfile::tempdir; -use platform::{inspect_process_identity, inspect_process_start_identity}; +use platform::{ + ProcessEvents, ProcessExitWatch, inspect_process_identity, inspect_process_start_identity, + process_is_zombie, +}; #[expect( clippy::disallowed_types, @@ -67,6 +72,57 @@ fn native_process_identity_reports_shebang_script_and_ordered_arguments() -> Res Ok(()) } +#[test] +fn process_watch_reports_exec_then_exit_without_reaping() -> Result<()> { + // `/bin/sh` would re-exec itself as bash, so start bash directly: its only exec is the one the + // test releases by writing a line. + let mut child = ChildGuard( + TestCommand::new("/bin/bash") + .args(["-c", "read line; exec /bin/sleep 30"]) + .stdin(Stdio::piped()) + .spawn()?, + ); + let pid = child.0.id(); + let birth = inspect_process_start_identity(pid)?; + let mut watch = ProcessExitWatch::with_exec(pid)?; + + assert_eq!(watch.try_events()?, ProcessEvents::default()); + child + .0 + .stdin + .take() + .ok_or_else(|| anyhow!("bash stdin was not piped"))? + .write_all(b"go\n")?; + let exec = wait_for_events(&mut watch, |events| events.exec)?; + let identity = inspect_child(&mut child)?; + child.0.kill()?; + let exit = wait_for_events(&mut watch, |events| events.exit.is_some())?; + + assert!(exec.exec); + assert_eq!(identity.executable, Utf8Path::new("/bin/sleep")); + assert_eq!(Some(identity.start_identity), birth); + assert_eq!(exit.exit.and_then(|status| status.signal()), Some(9)); + assert!(process_is_zombie(pid)?); + assert_eq!(child.0.wait()?.signal(), Some(9)); + + Ok(()) +} + +fn wait_for_events( + watch: &mut ProcessExitWatch, + done: impl Fn(&ProcessEvents) -> bool, +) -> Result { + for _attempt in 0..500 { + let events = watch.try_events()?; + if done(&events) { + return Ok(events); + } + thread::sleep(Duration::from_millis(10)); + } + + Err(anyhow!("process events never arrived")) +} + fn inspect_child(child: &mut ChildGuard) -> Result { let pid = child.0.id(); diff --git a/crates/state/src/fs.rs b/crates/state/src/fs.rs index 8bad019b..0563787c 100644 --- a/crates/state/src/fs.rs +++ b/crates/state/src/fs.rs @@ -155,6 +155,46 @@ pub fn open_append_file(path: &Utf8Path) -> Result { Ok(file) } +/// Opens a runtime log for reading and appending without following a symlink at `path`, then +/// proves through the open descriptor that it is a regular file with one link that this user owns. +#[cfg(unix)] +#[expect( + clippy::disallowed_types, + reason = "PV filesystem helper owns direct file handles" +)] +pub fn open_private_log_file(path: &Utf8Path) -> Result { + use std::os::unix::fs::OpenOptionsExt; + + ensure_parent_dir(path)?; + let no_follow = i32::try_from(rustix::fs::OFlags::NOFOLLOW.bits()) + .map_err(|error| StateError::filesystem(path.to_path_buf(), io::Error::other(error)))?; + let file = std::fs::OpenOptions::new() + .read(true) + .append(true) + .create(true) + .mode(SENSITIVE_FILE_MODE) + .custom_flags(no_follow) + .open(path) + .map_err(|source| StateError::filesystem(path.to_path_buf(), source))?; + let metadata = file + .metadata() + .map_err(|source| StateError::filesystem(path.to_path_buf(), source))?; + if !metadata.file_type().is_file() || metadata.nlink() != 1 || metadata.uid() != current_uid()? + { + return Err(StateError::filesystem( + path.to_path_buf(), + io::Error::new( + io::ErrorKind::InvalidInput, + "log is not a private regular file", + ), + )); + } + file.set_permissions(std::fs::Permissions::from_mode(SENSITIVE_FILE_MODE)) + .map_err(|source| StateError::filesystem(path.to_path_buf(), source))?; + + Ok(file) +} + #[expect( clippy::disallowed_types, reason = "PV filesystem helper owns direct file handles" diff --git a/crates/state/src/lib.rs b/crates/state/src/lib.rs index 60ff410f..94997605 100644 --- a/crates/state/src/lib.rs +++ b/crates/state/src/lib.rs @@ -25,7 +25,9 @@ pub use database::{ }; pub use error::{StateCapability, StateError}; pub use paths::{PathSummaryEntry, PvPaths}; -pub use update_lock::{HelperLifecycleLock, JobsLock, RuntimeLifecycleLock, UpdateLock}; +pub use update_lock::{ + HelperLifecycleLock, JobsLock, MonitorReservation, RuntimeLifecycleLock, UpdateLock, +}; #[doc(hidden)] pub mod testing { diff --git a/crates/state/src/paths.rs b/crates/state/src/paths.rs index e1acd958..74f71e05 100644 --- a/crates/state/src/paths.rs +++ b/crates/state/src/paths.rs @@ -124,10 +124,26 @@ impl PvPaths { } pub fn worker_admin_socket(&self, php_runtime: &str) -> Utf8PathBuf { - let digest = format!("{:x}", Sha256::digest(php_runtime.as_bytes())); - self.run() - .join(format!("worker-admin-{}.sock", &digest[..12])) + .join(format!("worker-admin-{}.sock", short_digest(php_runtime))) + } + + /// The directories of per-runtime monitors and their reservations. + pub fn monitors(&self) -> Utf8PathBuf { + self.run().join("m") + } + + /// The control socket and recovery record of the monitor that owns `subject`'s runtime. The + /// short digest keeps the socket path within macOS's Unix socket limit. + pub fn monitor_dir(&self, subject: &str) -> Utf8PathBuf { + self.monitors().join(short_digest(subject)) + } + + /// The reservation for `subject`'s monitor. It sits beside the monitor's directory, so removing + /// that directory for a replacement never unlinks a lock another process still holds. + pub fn monitor_lock(&self, subject: &str) -> Utf8PathBuf { + self.monitors() + .join(format!("{}.lock", short_digest(subject))) } pub fn daemon_startup_error(&self) -> Utf8PathBuf { @@ -363,3 +379,11 @@ impl PvPaths { ] } } + +/// The first 12 hex digits of `value`'s SHA-256: short enough for socket paths, and unique across +/// the few runtimes one PV home runs. +fn short_digest(value: &str) -> String { + let digest = format!("{:x}", Sha256::digest(value.as_bytes())); + + digest[..12].to_owned() +} diff --git a/crates/state/src/update_lock.rs b/crates/state/src/update_lock.rs index 5a04751e..42333a90 100644 --- a/crates/state/src/update_lock.rs +++ b/crates/state/src/update_lock.rs @@ -1,9 +1,9 @@ #[cfg(unix)] use std::io; #[cfg(unix)] -use std::os::fd::AsFd; +use std::os::fd::{AsFd, OwnedFd}; -use camino::Utf8Path; +use camino::{Utf8Path, Utf8PathBuf}; #[cfg(unix)] use rustix::fs::FlockOperation; @@ -48,6 +48,80 @@ pub struct RuntimeLifecycleLock { _file: std::fs::File, } +/// Reserves one runtime subject for one monitor. A controller acquires it before starting the +/// monitor and passes the same locked file to the monitor as stdin, so the reservation stays held, +/// with no gap, until the monitor process dies. +#[derive(Debug)] +#[must_use = "dropping the reservation lets another monitor start for this subject"] +#[expect( + clippy::disallowed_types, + reason = "monitor reservation owns the OS-locked file handle" +)] +pub struct MonitorReservation { + file: std::fs::File, + path: Utf8PathBuf, +} + +impl MonitorReservation { + pub fn acquire(paths: &PvPaths, subject: &str) -> Result { + require_file_locking()?; + fs::ensure_user_dir(paths.run())?; + fs::ensure_user_dir(&paths.monitors())?; + let path = paths.monitor_lock(subject); + let file = open_lock_file(&path)?; + fs::secure_sensitive_file(&path)?; + lock_exclusively(&file, &path)?; + + Ok(Self { file, path }) + } + + /// Another descriptor for the same locked file, for a child process's stdin. The lock stays + /// held while any descriptor for the file is open. + pub fn child_stdio(&self) -> Result { + let file = self + .file + .try_clone() + .map_err(|source| StateError::filesystem(self.path.clone(), source))?; + + Ok(file.into()) + } +} + +#[cfg(unix)] +impl MonitorReservation { + /// Adopts a reservation inherited as `descriptor`. It must be `subject`'s lock file, and it + /// must hold the lock: locking again succeeds on the open file that already holds it. + pub fn adopt(paths: &PvPaths, subject: &str, descriptor: OwnedFd) -> Result { + let path = paths.monitor_lock(subject); + let held = rustix::fs::fstat(&descriptor) + .map_err(|error| StateError::filesystem(path.clone(), io::Error::from(error)))?; + let expected = rustix::fs::statat( + rustix::fs::CWD, + path.as_std_path(), + rustix::fs::AtFlags::SYMLINK_NOFOLLOW, + ) + .map_err(|error| StateError::filesystem(path.clone(), io::Error::from(error)))?; + if rustix::fs::FileType::from_raw_mode(held.st_mode) != rustix::fs::FileType::RegularFile + || held.st_dev != expected.st_dev + || held.st_ino != expected.st_ino + { + return Err(StateError::filesystem( + path, + io::Error::new( + io::ErrorKind::InvalidInput, + "inherited descriptor is not this monitor's reservation", + ), + )); + } + lock_exclusively(&descriptor, &path)?; + + Ok(Self { + file: descriptor.into(), + path, + }) + } +} + impl RuntimeLifecycleLock { pub fn acquire_shared(paths: &PvPaths) -> Result { Self::acquire(paths, false) From 611fdffd9aa2b89f4223ec800e4c71ff755fa249 Mon Sep 17 00:00:00 2001 From: Clovis Muneza Date: Fri, 9 Oct 2026 20:52:18 -0400 Subject: [PATCH 2/6] feat(daemon): add per-service runtime monitor engine A monitor process owns one runtime for its whole life. It starts the runtime through a gate that execs only once the monitor has written its recovery record, so the runtime keeps the gate's pid and birth identity. The monitor learns the exit status from kqueue without reaping, so the zombie leader pins the process group until cleanup is proven, and it reaps the leader exactly once after that. Controllers use start_monitor, monitor_state, stop_monitor, release_monitor and recover_monitor over a bounded NDJSON protocol (version 1) that checks the peer and the instance on both sides. Release removes records only once the reservation proves the monitor exited. Recovery stops a dead monitor's runtime only by its exact identity from this boot. The monitor rotates its runtime's log by copy and truncate on the retained descriptor and keeps five archives that `pv logs` lists. Signalling a group whose only members are zombies now counts as done, because macOS answers EPERM instead of ESRCH for one. Nothing in production starts a monitor yet. --- crates/cli/src/commands/logs.rs | 48 + crates/daemon/examples/pv-monitor.rs | 44 + crates/daemon/src/error.rs | 24 + crates/daemon/src/lib.rs | 8 + crates/daemon/src/monitor.rs | 204 +++ crates/daemon/src/monitor/macos.rs | 1185 +++++++++++++++++ crates/daemon/src/monitor/rotation.rs | 196 +++ crates/daemon/src/supervisor.rs | 40 +- crates/daemon/tests/monitor.rs | 769 +++++++++++ ...r_requests_follow_the_frozen_contract.snap | 21 + ...ns_a_runtime_through_stop_and_release.snap | 31 + ...issing_command_and_leaves_no_instance.snap | 5 + crates/pv-fake/src/install.rs | 17 +- crates/pv-fake/src/lib.rs | 5 +- 14 files changed, 2581 insertions(+), 16 deletions(-) create mode 100644 crates/daemon/examples/pv-monitor.rs create mode 100644 crates/daemon/src/monitor.rs create mode 100644 crates/daemon/src/monitor/macos.rs create mode 100644 crates/daemon/src/monitor/rotation.rs create mode 100644 crates/daemon/tests/monitor.rs create mode 100644 crates/daemon/tests/snapshots/monitor__monitor_requests_follow_the_frozen_contract.snap create mode 100644 crates/daemon/tests/snapshots/monitor__monitor_runs_a_runtime_through_stop_and_release.snap create mode 100644 crates/daemon/tests/snapshots/monitor__monitor_start_reports_a_missing_command_and_leaves_no_instance.snap diff --git a/crates/cli/src/commands/logs.rs b/crates/cli/src/commands/logs.rs index 27c6b8b9..2e46830c 100644 --- a/crates/cli/src/commands/logs.rs +++ b/crates/cli/src/commands/logs.rs @@ -570,6 +570,54 @@ mod tests { }) } + #[test] + fn follow_sources_continues_after_a_rotation_truncates_the_log() -> anyhow::Result<()> { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + + runtime.block_on(async { + let tempdir = tempdir()?; + let gateway_log = tempdir.path().join("gateway.log"); + write_test_file(&gateway_log, "")?; + let sources = vec![LogSource { + label: "gateway".to_string(), + active_path: gateway_log.clone(), + }]; + let writer = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(250)).await; + append_test_file(&gateway_log, "before rotation\n")?; + tokio::time::sleep(Duration::from_millis(250)).await; + // A monitor rotates by truncating the log the runtime keeps appending to. + truncate_test_file(&gateway_log)?; + append_test_file(&gateway_log, "after\n")?; + + anyhow::Ok(()) + }); + let mut stdout = Vec::new(); + + follow_sources_async(&sources, false, &mut stdout, Some(2)).await?; + writer.await??; + + assert_eq!(String::from_utf8(stdout)?, "before rotation\nafter\n"); + + anyhow::Ok(()) + }) + } + + #[expect( + clippy::disallowed_types, + reason = "logs tests truncate fixture logs as rotation does" + )] + fn truncate_test_file(path: &Utf8Path) -> anyhow::Result<()> { + std::fs::OpenOptions::new() + .write(true) + .open(path)? + .set_len(0)?; + + Ok(()) + } + #[expect(clippy::disallowed_methods, reason = "logs tests create fixture files")] fn write_test_file(path: &Utf8Path, contents: &str) -> anyhow::Result<()> { std::fs::write(path, contents)?; diff --git a/crates/daemon/examples/pv-monitor.rs b/crates/daemon/examples/pv-monitor.rs new file mode 100644 index 00000000..8118c506 --- /dev/null +++ b/crates/daemon/examples/pv-monitor.rs @@ -0,0 +1,44 @@ +//! Builds a runtime monitor with test hooks along with the daemon's tests; see +//! `pv_fake::example_binary`. It runs the entry points of `pv monitor:run` and `pv monitor:gate` +//! with the [`daemon::MonitorHooks`] a test writes to `pv-monitor-hooks.json` in its PV home. + +use std::io::{self, Write as _}; +use std::process::ExitCode; + +use daemon::{DaemonError, MONITOR_GATE_COMMAND, MONITOR_RUN_COMMAND, MonitorHooks}; +use state::PvPaths; + +const HOOKS_FILE: &str = "pv-monitor-hooks.json"; + +fn main() -> ExitCode { + let command = std::env::args().nth(1); + let result = PvPaths::default_home() + .map_err(DaemonError::from) + .and_then(|paths| { + let hooks = hooks(&paths); + match command.as_deref() { + Some(MONITOR_RUN_COMMAND) => daemon::run_monitor_blocking(paths, hooks), + Some(MONITOR_GATE_COMMAND) => daemon::run_monitor_gate(&hooks), + _ => Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("usage: pv-monitor {MONITOR_RUN_COMMAND}|{MONITOR_GATE_COMMAND}"), + ) + .into()), + } + }); + + match result { + Ok(()) => ExitCode::SUCCESS, + Err(error) => { + let _write_result = writeln!(io::stderr(), "pv-monitor: {error}"); + ExitCode::FAILURE + } + } +} + +fn hooks(paths: &PvPaths) -> MonitorHooks { + state::fs::read_to_string(&paths.home().join(HOOKS_FILE)) + .ok() + .and_then(|hooks| serde_json::from_str(&hooks).ok()) + .unwrap_or_default() +} diff --git a/crates/daemon/src/error.rs b/crates/daemon/src/error.rs index f5597922..7327ef71 100644 --- a/crates/daemon/src/error.rs +++ b/crates/daemon/src/error.rs @@ -254,6 +254,30 @@ pub enum DaemonError { #[error("cleanup of process {pid} is unproven: {reason}")] RuntimeCleanupUnproven { pid: u32, reason: String }, + #[error("monitor socket path {path} is longer than the {limit}-byte Unix socket limit")] + MonitorSocketPathTooLong { + path: camino::Utf8PathBuf, + limit: usize, + }, + + #[error( + "the monitor of `{subject}` left records for instance {instance}; recover them before starting it again" + )] + MonitorUnresolved { subject: String, instance: String }, + + #[error("the monitor of `{subject}` failed to start: {message}")] + MonitorStartupFailed { subject: String, message: String }, + + #[error("the monitor of `{subject}` is unavailable: {reason}")] + MonitorUnavailable { subject: String, reason: String }, + + #[error("the monitor of `{subject}` rejected the request ({kind}): {message}")] + MonitorRejected { + subject: String, + kind: crate::monitor::MonitorErrorKind, + message: String, + }, + #[error("runtime record is incomplete or invalid: {path}")] InvalidRuntimeRecord { path: camino::Utf8PathBuf }, diff --git a/crates/daemon/src/lib.rs b/crates/daemon/src/lib.rs index b3502d4c..bc07a2dd 100644 --- a/crates/daemon/src/lib.rs +++ b/crates/daemon/src/lib.rs @@ -8,6 +8,7 @@ mod health; mod ipc; mod jobs; mod managed_resources; +mod monitor; mod project_env; mod reconciliation; mod runtime_stop; @@ -40,6 +41,13 @@ pub use client::{ }; pub use dns::{dns_port_available, response_bytes}; pub use error::{DaemonError, ManagedResourceProjectFailure, RuntimeReconciliationFailure}; +pub use monitor::{ + MONITOR_GATE_COMMAND, MONITOR_RUN_COMMAND, MonitorCleanup, MonitorErrorKind, MonitorExit, + MonitorHooks, MonitorPause, MonitorStart, MonitorState, MonitorStop, run_monitor_blocking, + run_monitor_gate, +}; +#[cfg(target_os = "macos")] +pub use monitor::{monitor_state, recover_monitor, release_monitor, start_monitor, stop_monitor}; pub use protocol::PROTOCOL_VERSION; pub use reconciliation::{ EnqueueResult, QueuedReconciliation, ReconciliationDebouncer, ReconciliationJob, diff --git a/crates/daemon/src/monitor.rs b/crates/daemon/src/monitor.rs new file mode 100644 index 00000000..ad39ebcf --- /dev/null +++ b/crates/daemon/src/monitor.rs @@ -0,0 +1,204 @@ +//! Per-runtime monitors. +//! +//! A monitor is a small `pv monitor:run` process that owns one runtime for the runtime's whole +//! life, whether or not the daemon is running. It starts the runtime through `pv monitor:gate`, +//! which execs the runtime only once the monitor has recorded it, so the runtime keeps the gate's +//! pid and birth identity. The monitor learns how the runtime exited from a kqueue watch without +//! reaping it, so the exited leader keeps its process group pinned until the group's cleanup is +//! proven. Controllers talk to the monitor over a private socket, and the monitor's reservation +//! lock proves when it has exited. + +#[cfg(target_os = "macos")] +mod macos; +#[cfg(target_os = "macos")] +mod rotation; + +use std::collections::BTreeMap; +use std::fmt; +#[cfg(target_os = "macos")] +use std::time::Duration; + +use camino::Utf8PathBuf; +use serde::{Deserialize, Serialize}; +#[cfg(not(target_os = "macos"))] +use state::PvPaths; + +#[cfg(not(target_os = "macos"))] +use crate::DaemonError; +use crate::StopSignal; +use crate::supervisor::PrivateEnvironmentDebug; + +#[cfg(target_os = "macos")] +pub use macos::{ + monitor_state, recover_monitor, release_monitor, run_monitor_blocking, run_monitor_gate, + start_monitor, stop_monitor, +}; + +/// The hidden command that runs a monitor. +pub const MONITOR_RUN_COMMAND: &str = "monitor:run"; +/// The hidden command a monitor starts its runtime through. +pub const MONITOR_GATE_COMMAND: &str = "monitor:gate"; + +/// What a controller asks a new monitor to run. +#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)] +pub struct MonitorStart { + /// Names the runtime, such as `gateway`; one monitor runs per subject. + pub subject: String, + pub command: Utf8PathBuf, + pub arguments: Vec, + /// Reaches the runtime only through its environment; never recorded. + pub private_environment: BTreeMap, + /// Where the runtime's stdout and stderr go. + pub log_path: Utf8PathBuf, + /// How the monitor stops what is left of the runtime's process group when no controller asks: + /// after the runtime exits on its own, and once the lifeline closes. + pub fallback_stop: MonitorStop, + /// An inherited pipe read end. Once every write end closes, the monitor stops the runtime + /// and exits. Tests pass their lifeline; production passes `None`. + pub lifeline_fd: Option, +} + +impl fmt::Debug for MonitorStart { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("MonitorStart") + .field("subject", &self.subject) + .field("command", &self.command) + .field("arguments", &self.arguments) + .field( + "private_environment", + &PrivateEnvironmentDebug(&self.private_environment), + ) + .field("log_path", &self.log_path) + .field("fallback_stop", &self.fallback_stop) + .field("lifeline_fd", &self.lifeline_fd) + .finish() + } +} + +/// A graceful signal for the runtime's process group, then SIGKILL once `grace_ms` passes. +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct MonitorStop { + pub signal: StopSignal, + pub grace_ms: u64, +} + +#[cfg(target_os = "macos")] +impl MonitorStop { + fn grace(self) -> Duration { + Duration::from_millis(self.grace_ms) + } +} + +/// What a monitor knows about its runtime. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct MonitorState { + pub subject: String, + pub monitor_pid: u32, + pub runtime_pid: u32, + /// The first stop a controller requested. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stop: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub exit: Option, + pub cleanup: MonitorCleanup, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub log_rotation_error: Option, +} + +/// How the runtime exited. +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum MonitorExit { + Code(i32), + Signal(i32), +} + +/// Whether the runtime's whole process group is proven gone. +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum MonitorCleanup { + /// The runtime is running, or its group is being stopped. + Pending, + /// No live member remains and the runtime leader is reaped. + Complete, + /// The group could not be proven empty; the monitor keeps its records. + Unproven, +} + +/// Why a monitor refused a request. +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum MonitorErrorKind { + UnsupportedVersion, + InstanceMismatch, + NotReleasable, + InvalidRequest, + StartupFailed, +} + +impl fmt::Display for MonitorErrorKind { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(match self { + Self::UnsupportedVersion => "unsupported_version", + Self::InstanceMismatch => "instance_mismatch", + Self::NotReleasable => "not_releasable", + Self::InvalidRequest => "invalid_request", + Self::StartupFailed => "startup_failed", + }) + } +} + +/// Test-only behavior for the `pv-monitor` example. Production `pv` uses the defaults. +#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] +#[serde(default)] +pub struct MonitorHooks { + /// Where a paused point writes `.reached`, holding its pid, then waits for + /// `.continue`. + pub pause_dir: Option, + pub pauses: Vec, + /// Makes the final process-group check report an unknown outcome. + pub unknown_group_check: bool, + pub rotation_bytes: Option, + pub rotation_interval_ms: Option, +} + +/// A point where the monitor or its gate can be held for a test. +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum MonitorPause { + /// Before the monitor writes its recovery record. + BeforePublication, + /// After the record, before the gate may exec the runtime. + BeforePermission, + /// In the gate, after permission and before exec. + GateAfterPermission, + /// During a log rotation, before the copy. + RotationCopy, +} + +#[cfg(target_os = "macos")] +impl MonitorPause { + const fn name(self) -> &'static str { + match self { + Self::BeforePublication => "before_publication", + Self::BeforePermission => "before_permission", + Self::GateAfterPermission => "gate_after_permission", + Self::RotationCopy => "rotation_copy", + } + } +} + +#[cfg(not(target_os = "macos"))] +pub fn run_monitor_blocking(_paths: PvPaths, _hooks: MonitorHooks) -> Result<(), DaemonError> { + platform::require_capability(platform::PlatformCapability::ProcessContainment)?; + + Ok(()) +} + +#[cfg(not(target_os = "macos"))] +pub fn run_monitor_gate(_hooks: &MonitorHooks) -> Result<(), DaemonError> { + platform::require_capability(platform::PlatformCapability::ProcessContainment)?; + + Ok(()) +} diff --git a/crates/daemon/src/monitor/macos.rs b/crates/daemon/src/monitor/macos.rs new file mode 100644 index 00000000..d9e67947 --- /dev/null +++ b/crates/daemon/src/monitor/macos.rs @@ -0,0 +1,1185 @@ +use std::fmt::Write as _; +use std::io::{self, BufRead as _, BufReader, Write as _}; +use std::os::fd::{AsFd as _, OwnedFd}; +use std::os::unix::net::UnixStream as StdUnixStream; +use std::os::unix::process::{CommandExt as _, ExitStatusExt as _}; +use std::process::{ExitStatus, Stdio}; +use std::time::Duration; + +use camino::{Utf8Path, Utf8PathBuf}; +use futures_util::StreamExt as _; +use platform::{BootSessionId, ProcessEvents, ProcessExitWatch, ProcessStartIdentity}; +use protocol::{DaemonTransport, ProtocolError, transport, write_line}; +use rustix::fs::FileType; +use serde::{Deserialize, Serialize}; +use state::{MonitorReservation, PvPaths, StateError, fs}; +use tokio::io::Interest; +use tokio::io::unix::AsyncFd; +use tokio::net::{UnixListener, UnixStream}; +use tokio::sync::{mpsc, oneshot}; +use tokio::task::JoinHandle; +use tokio::time::{Instant, MissedTickBehavior, interval_at, sleep, timeout}; + +use super::{ + MONITOR_GATE_COMMAND, MONITOR_RUN_COMMAND, MonitorCleanup, MonitorErrorKind, MonitorExit, + MonitorHooks, MonitorPause, MonitorStart, MonitorState, MonitorStop, rotation, +}; +use crate::DaemonError; +use crate::supervisor::{ + PHP_INI_ENVIRONMENT_KEYS, reap_process_if_child, stop_process_group, + wait_for_process_group_exit, +}; +use crate::{StopSignal, build_runtime}; + +const PROTOCOL_VERSION: u32 = 1; +/// macOS keeps a Unix socket path in 104 bytes, including its terminating NUL. +const SOCKET_PATH_LIMIT: usize = 103; +const RECORD_FILE: &str = "monitor.json"; +const SOCKET_FILE: &str = "sock"; +const STARTUP_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); +const STARTUP_REPLY_TIMEOUT: Duration = Duration::from_secs(30); +const EXEC_TIMEOUT: Duration = Duration::from_secs(10); +const REQUEST_TIMEOUT: Duration = Duration::from_secs(5); +const RELEASE_TIMEOUT: Duration = Duration::from_secs(5); +const STOP_MARGIN: Duration = Duration::from_secs(5); +const GATE_FAILURE_TIMEOUT: Duration = Duration::from_millis(100); +const EXITING_LEADER_TIMEOUT: Duration = Duration::from_secs(1); +const POLL_INTERVAL: Duration = Duration::from_millis(10); +const ROTATION_INTERVAL: Duration = Duration::from_secs(60); +const ROTATION_BYTES: u64 = 10 * 1024 * 1024; + +/// The startup request a controller writes to a new monitor's stdout socket. +#[derive(Deserialize, Serialize)] +struct StartRequest { + version: u32, + instance: String, + #[serde(flatten)] + start: MonitorStart, +} + +#[derive(Debug, Deserialize, Serialize)] +struct RequestHeader { + version: u32, + instance: String, +} + +#[derive(Debug, Deserialize, Serialize)] +struct Request { + version: u32, + instance: String, + #[serde(flatten)] + operation: Operation, +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(tag = "op", rename_all = "snake_case")] +enum Operation { + State, + Stop(MonitorStop), + Release, +} + +#[derive(Debug, Deserialize, Serialize)] +struct Reply { + version: u32, + instance: String, + #[serde(flatten)] + result: ReplyResult, +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(tag = "result", rename_all = "snake_case")] +enum ReplyResult { + State(MonitorState), + Accepted, + Error { + kind: MonitorErrorKind, + message: String, + }, +} + +impl Reply { + fn new(instance: &str, result: ReplyResult) -> Self { + Self { + version: PROTOCOL_VERSION, + instance: instance.to_owned(), + result, + } + } +} + +/// What a monitor records in `monitor.json` before its runtime may start: enough for a later +/// controller to authenticate the monitor, and to stop the runtime if the monitor dies. It holds +/// no secrets. +#[derive(Debug, Deserialize, Serialize)] +struct MonitorRecord { + version: u32, + subject: String, + instance: String, + boot_session_id: BootSessionId, + monitor: RecordedProcess, + runtime: RecordedProcess, + command: Utf8PathBuf, + arguments: Vec, + log_path: Utf8PathBuf, + fallback_stop: MonitorStop, +} + +#[derive(Clone, Copy, Debug, Deserialize, Serialize)] +struct RecordedProcess { + pid: u32, + start_identity: ProcessStartIdentity, +} + +#[derive(Deserialize, Serialize)] +struct GatePermission { + command: Utf8PathBuf, + arguments: Vec, +} + +#[derive(Deserialize, Serialize)] +struct GateFailure { + message: String, +} + +/// Starts a monitor for `start.subject` with `executable`, which is `pv` or a build of the +/// `pv-monitor` example, and returns its first state once the runtime has exec'd. +/// +/// The reservation passes to the monitor as its stdin, so it stays held with no gap. Any failure +/// kills the monitor and recovers what it left. +pub async fn start_monitor( + paths: &PvPaths, + executable: &Utf8Path, + start: MonitorStart, +) -> Result { + let subject = start.subject.clone(); + monitor_socket(paths, &subject)?; + let reservation = MonitorReservation::acquire(paths, &subject) + .map_err(|error| reservation_error(&subject, error))?; + if let Some(record) = read_record(paths, &subject)? { + return Err(DaemonError::MonitorUnresolved { + subject, + instance: record.instance, + }); + } + remove_monitor_dir(paths, &subject)?; + let request = StartRequest { + version: PROTOCOL_VERSION, + instance: new_instance()?, + start, + }; + let (controller_end, monitor_end) = StdUnixStream::pair()?; + let mut monitor = spawn_monitor(paths, executable, &reservation, monitor_end)?; + + match complete_startup(controller_end, &request).await { + // Tokio reaps the monitor whenever it exits. + Ok(state) => Ok(state), + Err(error) => { + let _kill_result = monitor.start_kill(); + let _wait_result = monitor.wait().await; + match recover_reserved(paths, &subject).await { + Ok(()) => Err(error), + Err(cleanup) => Err(DaemonError::StartupCleanupFailed { + source: Box::new(error), + cleanup: Box::new(cleanup), + }), + } + } + } +} + +/// Asks the monitor of `subject` for its state. +pub async fn monitor_state(paths: &PvPaths, subject: &str) -> Result { + match request(paths, subject, Operation::State).await?.1 { + ReplyResult::State(state) => Ok(state), + ReplyResult::Accepted | ReplyResult::Error { .. } => Err(unavailable( + subject, + "the monitor replied without its state", + )), + } +} + +/// Asks the monitor of `subject` to stop its runtime, then waits until cleanup is decided. +pub async fn stop_monitor( + paths: &PvPaths, + subject: &str, + stop: MonitorStop, +) -> Result { + request(paths, subject, Operation::Stop(stop)).await?; + let deadline = Instant::now().checked_add(stop.grace().saturating_add(STOP_MARGIN)); + + loop { + let state = monitor_state(paths, subject).await?; + if state.cleanup != MonitorCleanup::Pending { + return Ok(state); + } + if deadline.is_some_and(|deadline| Instant::now() >= deadline) { + return Err(unavailable(subject, "the runtime's cleanup did not finish")); + } + sleep(POLL_INTERVAL).await; + } +} + +/// Releases the monitor of `subject` once its runtime has exited and its cleanup is complete, +/// then removes the monitor's records once its reservation proves it has exited. +pub async fn release_monitor(paths: &PvPaths, subject: &str) -> Result<(), DaemonError> { + let (record, _accepted) = request(paths, subject, Operation::Release).await?; + let reservation = wait_for_reservation(paths, subject).await?; + // Only this instance's records: a replacement that failed may have left its own. + if read_record(paths, subject)?.is_some_and(|current| current.instance == record.instance) { + remove_monitor_dir(paths, subject)?; + } + drop(reservation); + + Ok(()) +} + +/// Recovers after the monitor of `subject` died: stops its runtime if the runtime is exactly the +/// recorded process, then removes the records. A live monitor holds its reservation, so this +/// refuses to touch it. +pub async fn recover_monitor(paths: &PvPaths, subject: &str) -> Result<(), DaemonError> { + let reservation = MonitorReservation::acquire(paths, subject) + .map_err(|error| reservation_error(subject, error))?; + let result = recover_reserved(paths, subject).await; + drop(reservation); + + result +} + +async fn recover_reserved(paths: &PvPaths, subject: &str) -> Result<(), DaemonError> { + let Some(record) = read_record(paths, subject)? else { + return remove_monitor_dir(paths, subject); + }; + // After a reboot, nothing the record names can still be running. A runtime leader that + // exited is at most a zombie, which no live member keeps company. + if record.boot_session_id == platform::current_boot_session_id()? + && platform::process_group_has_live_members(record.runtime.pid)? + { + let runtime = record.runtime; + match platform::inspect_process_start_identity(runtime.pid)? { + Some(identity) if identity == runtime.start_identity => { + stop_process_group( + runtime.pid, + record.fallback_stop.signal, + record.fallback_stop.grace(), + || runtime_matches(runtime), + ) + .await?; + } + // A leader that is exiting has no identity yet still counts as live: it may finish. + None if wait_for_process_group_exit(runtime.pid, EXITING_LEADER_TIMEOUT).await? => {} + Some(_) | None => { + return Err(DaemonError::RuntimeProcessIdentityChanged { pid: runtime.pid }); + } + } + } + + remove_monitor_dir(paths, subject) +} + +fn runtime_matches(runtime: RecordedProcess) -> Result { + Ok(platform::inspect_process_start_identity(runtime.pid)? == Some(runtime.start_identity)) +} + +#[expect( + clippy::disallowed_types, + reason = "the monitor controller owns spawning monitor processes" +)] +fn spawn_monitor( + paths: &PvPaths, + executable: &Utf8Path, + reservation: &MonitorReservation, + monitor_end: StdUnixStream, +) -> Result { + let mut command = tokio::process::Command::new(executable); + command + .arg(MONITOR_RUN_COMMAND) + .env("HOME", paths.home()) + .stdin(reservation.child_stdio()?) + .stdout(OwnedFd::from(monitor_end)) + .stderr(Stdio::null()); + + // Dropping `command` closes this process's copies of the reservation and the socket end. + Ok(command.spawn()?) +} + +async fn complete_startup( + stream: StdUnixStream, + request: &StartRequest, +) -> Result { + let subject = &request.start.subject; + stream.set_nonblocking(true)?; + let mut startup = transport(UnixStream::from_std(stream)?); + write_line(&mut startup, request).await?; + let reply = timeout(STARTUP_REPLY_TIMEOUT, read_reply(&mut startup)) + .await + .map_err(|_elapsed| DaemonError::ProtocolTimedOut { + phase: "monitor startup", + })?? + .ok_or_else(|| startup_failed(subject, "the monitor exited before it started"))?; + if reply.version != PROTOCOL_VERSION || reply.instance != request.instance { + return Err(startup_failed( + subject, + "the monitor replied for another instance", + )); + } + + match reply.result { + ReplyResult::State(state) => Ok(state), + ReplyResult::Error { message, .. } => Err(startup_failed(subject, message)), + ReplyResult::Accepted => Err(startup_failed( + subject, + "the monitor replied without its state", + )), + } +} + +/// Sends `operation` to the monitor of `subject` after proving the socket is served by the +/// recorded monitor process, from this boot, and returns its record with the reply. +async fn request( + paths: &PvPaths, + subject: &str, + operation: Operation, +) -> Result<(MonitorRecord, ReplyResult), DaemonError> { + let record = read_record(paths, subject)? + .ok_or_else(|| unavailable(subject, "no monitor is recorded"))?; + let socket = monitor_socket(paths, subject)?; + let stream = timeout(REQUEST_TIMEOUT, UnixStream::connect(&socket)) + .await + .map_err(|_elapsed| unavailable(subject, "connecting timed out"))? + .map_err(|error| unavailable(subject, &format!("cannot connect: {error}")))?; + if !peer_is_monitor(&stream, &record)? { + return Err(unavailable( + subject, + "the socket is not served by the recorded monitor", + )); + } + let mut connection = transport(stream); + let request = Request { + version: PROTOCOL_VERSION, + instance: record.instance.clone(), + operation, + }; + write_line(&mut connection, &request).await?; + let reply = timeout(REQUEST_TIMEOUT, read_reply(&mut connection)) + .await + .map_err(|_elapsed| unavailable(subject, "the monitor did not reply"))?? + .ok_or_else(|| unavailable(subject, "the monitor closed the connection"))?; + if reply.instance != record.instance { + return Err(unavailable(subject, "the reply came from another instance")); + } + if let ReplyResult::Error { kind, message } = reply.result { + return Err(DaemonError::MonitorRejected { + subject: subject.to_owned(), + kind, + message, + }); + } + + Ok((record, reply.result)) +} + +fn peer_is_monitor(stream: &UnixStream, record: &MonitorRecord) -> Result { + let credentials = stream.peer_cred()?; + let peer_pid = credentials.pid().and_then(|pid| u32::try_from(pid).ok()); + + Ok(credentials.uid() == rustix::process::getuid().as_raw() + && peer_pid == Some(record.monitor.pid) + && platform::inspect_process_start_identity(record.monitor.pid)? + == Some(record.monitor.start_identity) + && platform::current_boot_session_id()? == record.boot_session_id) +} + +async fn read_reply( + connection: &mut DaemonTransport, +) -> Result, DaemonError> { + let Some(line) = connection.next().await else { + return Ok(None); + }; + + Ok(Some(serde_json::from_str( + &line.map_err(ProtocolError::from)?, + )?)) +} + +async fn wait_for_reservation( + paths: &PvPaths, + subject: &str, +) -> Result { + let deadline = Instant::now() + RELEASE_TIMEOUT; + + loop { + match MonitorReservation::acquire(paths, subject) { + Ok(reservation) => return Ok(reservation), + Err(StateError::CoordinationLockHeld { .. }) if Instant::now() < deadline => { + sleep(POLL_INTERVAL).await; + } + Err(error) => return Err(reservation_error(subject, error)), + } + } +} + +/// Runs a monitor: `pv monitor:run`. Its stdin is the reservation and its stdout the startup +/// socket. Returns once a controller releases it, or the lifeline closes and the runtime is +/// cleaned up. +pub fn run_monitor_blocking(paths: PvPaths, hooks: MonitorHooks) -> Result<(), DaemonError> { + platform::require_capability(platform::PlatformCapability::ProcessContainment)?; + // A session of its own, so a signal for the controller's group or terminal never reaches it. + rustix::process::setsid().map_err(io::Error::from)?; + let runtime = build_runtime()?; + let result = runtime.block_on(run_monitor(&paths, &hooks)); + // A held rotation or the lifeline reader must not keep the monitor alive. + runtime.shutdown_background(); + + result +} + +async fn run_monitor(paths: &PvPaths, hooks: &MonitorHooks) -> Result<(), DaemonError> { + let startup = StdUnixStream::from(io::stdout().as_fd().try_clone_to_owned()?); + startup.set_nonblocking(true)?; + let mut startup = transport(UnixStream::from_std(startup)?); + let line = timeout(STARTUP_REQUEST_TIMEOUT, startup.next()) + .await + .map_err(|_elapsed| DaemonError::ProtocolTimedOut { + phase: "monitor startup request", + })? + .ok_or_else(|| { + io::Error::new( + io::ErrorKind::UnexpectedEof, + "the controller closed before sending a startup request", + ) + })?; + let request: StartRequest = serde_json::from_str(&line.map_err(ProtocolError::from)?)?; + + let monitor = match start_runtime(paths, hooks, &request).await { + Ok(monitor) => monitor, + Err(error) => { + let message = match &error { + DaemonError::MonitorStartupFailed { message, .. } => message.clone(), + error => error.to_string(), + }; + let reply = Reply::new( + &request.instance, + ReplyResult::Error { + kind: MonitorErrorKind::StartupFailed, + message, + }, + ); + let _reply_result = write_line(&mut startup, &reply).await; + return Err(error); + } + }; + let reply = Reply::new(&request.instance, ReplyResult::State(monitor.state())); + // The runtime is running: keep supervising it even if the controller has gone. + let _reply_result = write_line(&mut startup, &reply).await; + drop(startup); + + monitor.run(hooks).await +} + +async fn start_runtime( + paths: &PvPaths, + hooks: &MonitorHooks, + request: &StartRequest, +) -> Result { + let start = &request.start; + let subject = &start.subject; + if request.version != PROTOCOL_VERSION { + return Err(startup_failed( + subject, + format!( + "this monitor speaks protocol version {PROTOCOL_VERSION}, not {}", + request.version + ), + )); + } + let reservation = + MonitorReservation::adopt(paths, subject, io::stdin().as_fd().try_clone_to_owned()?)?; + let socket = monitor_socket(paths, subject)?; + let log = fs::open_private_log_file(&start.log_path)?; + let lifeline = start.lifeline_fd.map(lifeline_path).transpose()?; + fs::ensure_user_dir(&paths.monitor_dir(subject))?; + let (gate, gate_end) = StdUnixStream::pair()?; + let pid = spawn_gate(start, &log, gate_end)?; + + match supervise_gate(paths, hooks, request, pid, gate, &socket).await { + Ok((watch, exit, record, listener)) => Ok(Monitor { + paths: paths.clone(), + record, + watch, + listener, + log, + lifeline, + _reservation: reservation, + stop: None, + exit, + cleanup: MonitorCleanup::Pending, + stopped: false, + executor: None, + rotation: None, + log_rotation_error: None, + }), + Err(error) => { + abandon_gate(paths, subject, pid).await; + Err(error) + } + } +} + +type SupervisedGate = ( + Option>, + Option, + MonitorRecord, + UnixListener, +); + +/// Records the gate, lets it exec the runtime, and confirms the exec. +async fn supervise_gate( + paths: &PvPaths, + hooks: &MonitorHooks, + request: &StartRequest, + pid: u32, + gate: StdUnixStream, + socket: &Utf8Path, +) -> Result { + let start = &request.start; + let subject = &start.subject; + // A kqueue descriptor is only ever readable; registering it as writable fails. + let mut watch = AsyncFd::with_interest(ProcessExitWatch::with_exec(pid)?, Interest::READABLE)?; + let record = MonitorRecord { + version: PROTOCOL_VERSION, + subject: subject.clone(), + instance: request.instance.clone(), + boot_session_id: platform::current_boot_session_id()?, + monitor: recorded_process(std::process::id())?, + runtime: recorded_process(pid)?, + command: start.command.clone(), + arguments: start.arguments.clone(), + log_path: start.log_path.clone(), + fallback_stop: start.fallback_stop, + }; + pause(hooks, MonitorPause::BeforePublication).await; + fs::write_sensitive_file( + &record_path(paths, subject), + &serde_json::to_string_pretty(&record)?, + )?; + pause(hooks, MonitorPause::BeforePermission).await; + gate.set_nonblocking(true)?; + let mut gate = transport(UnixStream::from_std(gate)?); + let permission = GatePermission { + command: start.command.clone(), + arguments: start.arguments.clone(), + }; + write_line(&mut gate, &permission).await?; + let events = timeout(EXEC_TIMEOUT, confirm_exec(&mut watch, &mut gate, subject)) + .await + .map_err(|_elapsed| startup_failed(subject, "the runtime did not start in time"))??; + let exit = events.exit.map(MonitorExit::from); + // An exit event ends the kqueue watch. + let watch = exit.is_none().then_some(watch); + fs::remove_file_if_exists(socket)?; + let listener = UnixListener::bind(socket)?; + fs::secure_sensitive_file(socket)?; + + Ok((watch, exit, record, listener)) +} + +/// Waits for the gate's exec. The gate's socket closing proves nothing: only the kqueue exec +/// event does, or the gate's own report that the exec failed. +async fn confirm_exec( + watch: &mut AsyncFd, + gate: &mut DaemonTransport, + subject: &str, +) -> Result { + let mut gate_open = true; + + loop { + tokio::select! { + events = next_events(watch) => { + let events = events?; + if events.exec { + return Ok(events); + } + if let Some(status) = events.exit { + let message = timeout(GATE_FAILURE_TIMEOUT, gate_failure(gate)) + .await + .ok() + .flatten() + .unwrap_or_else(|| format!("the gate exited before the runtime started ({status})")); + return Err(startup_failed(subject, message)); + } + } + line = gate.next(), if gate_open => match line { + Some(line) => { + let line = line.map_err(ProtocolError::from)?; + return Err(startup_failed(subject, failure_message(&line))); + } + None => gate_open = false, + }, + } + } +} + +async fn gate_failure(gate: &mut DaemonTransport) -> Option { + let line = gate.next().await?.ok()?; + + Some(failure_message(&line)) +} + +fn failure_message(line: &str) -> String { + serde_json::from_str::(line).map_or_else( + |_error| "the gate reported an unreadable failure".to_owned(), + |failure| failure.message, + ) +} + +async fn next_events(watch: &mut AsyncFd) -> Result { + loop { + let mut ready = watch.readable_mut().await?; + let events = ready.get_inner_mut().try_events()?; + if events != ProcessEvents::default() { + return Ok(events); + } + ready.clear_ready(); + } +} + +/// Kills a gate whose startup failed, reaps it, and removes the records once its group is gone. +async fn abandon_gate(paths: &PvPaths, subject: &str, pid: u32) { + let stopped = stop_process_group(pid, StopSignal::Terminate, Duration::ZERO, || Ok(true)).await; + if stopped.is_ok() && reap_process_if_child(pid).is_ok() { + let _remove_result = remove_monitor_dir(paths, subject); + } +} + +#[expect( + clippy::disallowed_types, + reason = "the monitor spawns its gate with std, so no Tokio waiter can reap the runtime" +)] +#[expect( + clippy::disallowed_methods, + reason = "the gate is this same monitor executable" +)] +fn spawn_gate( + start: &MonitorStart, + log: &std::fs::File, + gate_end: StdUnixStream, +) -> Result { + let mut command = std::process::Command::new(std::env::current_exe()?); + command + .arg(MONITOR_GATE_COMMAND) + .process_group(0) + .stdin(OwnedFd::from(gate_end)) + .stdout(log.try_clone()?) + .stderr(log.try_clone()?); + for key in PHP_INI_ENVIRONMENT_KEYS { + command.env_remove(key); + } + command.envs(&start.private_environment); + + // Dropping the std child neither waits for nor kills it. + Ok(command.spawn()?.id()) +} + +/// Starts the runtime: `pv monitor:gate`. Its stdin is the monitor's gate socket. It waits for +/// the monitor's permission, then execs the runtime, so the runtime keeps this process's pid and +/// birth identity. Returns only if it cannot. +pub fn run_monitor_gate(hooks: &MonitorHooks) -> Result<(), DaemonError> { + platform::require_capability(platform::PlatformCapability::ProcessContainment)?; + let mut socket = StdUnixStream::from(io::stdin().as_fd().try_clone_to_owned()?); + let mut line = String::new(); + BufReader::new(&socket).read_line(&mut line)?; + if line.is_empty() { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "the monitor closed before permitting the runtime to start", + ) + .into()); + } + let permission: GatePermission = serde_json::from_str(&line)?; + pause_blocking(hooks, MonitorPause::GateAfterPermission); + + let error = runtime_command(&permission).exec(); + let failure = GateFailure { + message: format!("cannot start {}: {error}", permission.command), + }; + let _report_result = writeln!(socket, "{}", serde_json::to_string(&failure)?); + + Err(error.into()) +} + +#[expect( + clippy::disallowed_types, + reason = "the gate replaces itself with the runtime" +)] +fn runtime_command(permission: &GatePermission) -> std::process::Command { + let mut command = std::process::Command::new(&permission.command); + command.args(&permission.arguments).stdin(Stdio::null()); + + command +} + +/// A running monitor's view of its runtime. +#[expect( + clippy::disallowed_types, + reason = "the monitor retains the runtime log's append descriptor for rotation" +)] +struct Monitor { + paths: PvPaths, + record: MonitorRecord, + /// The runtime's kqueue watch until its exit is known. + watch: Option>, + listener: UnixListener, + log: std::fs::File, + lifeline: Option, + _reservation: MonitorReservation, + stop: Option, + exit: Option, + cleanup: MonitorCleanup, + /// Whether the stop executor proved the group empty, before the exit event arrived. + stopped: bool, + executor: Option>>, + rotation: Option>>, + log_rotation_error: Option, +} + +enum Command { + State(oneshot::Sender), + Stop(MonitorStop), + Release(oneshot::Sender>), + Exit, +} + +impl Monitor { + fn state(&self) -> MonitorState { + MonitorState { + subject: self.record.subject.clone(), + monitor_pid: self.record.monitor.pid, + runtime_pid: self.record.runtime.pid, + stop: self.stop, + exit: self.exit, + cleanup: self.cleanup, + log_rotation_error: self.log_rotation_error.clone(), + } + } + + async fn run(mut self, hooks: &MonitorHooks) -> Result<(), DaemonError> { + let (commands, mut requests) = mpsc::channel(16); + let mut lifeline = self.lifeline.take().map(watch_lifeline).transpose()?; + let mut self_release = false; + let rotation_period = hooks + .rotation_interval_ms + .map_or(ROTATION_INTERVAL, Duration::from_millis); + let mut rotation_tick = interval_at(Instant::now() + rotation_period, rotation_period); + rotation_tick.set_missed_tick_behavior(MissedTickBehavior::Delay); + if self.exit.is_some() { + self.start_cleanup(self.record.fallback_stop); + } + + loop { + if self_release { + match self.cleanup { + MonitorCleanup::Complete => { + return remove_monitor_dir(&self.paths, &self.record.subject); + } + // Keep the records as evidence for recovery. + MonitorCleanup::Unproven => return Ok(()), + MonitorCleanup::Pending => {} + } + } + + tokio::select! { + events = next_optional_events(&mut self.watch) => self.observe(events?, hooks), + accepted = self.listener.accept() => { + if let Ok((stream, _address)) = accepted { + let instance = self.record.instance.clone(); + tokio::spawn(serve_connection(stream, instance, commands.clone())); + } + } + Some(command) = requests.recv() => match command { + Command::State(reply) => { + let _send_result = reply.send(self.state()); + } + Command::Stop(stop) => { + self.stop.get_or_insert(stop); + self.start_cleanup(stop); + } + Command::Release(reply) => { + let _send_result = reply.send(self.releasable()); + } + Command::Exit => return Ok(()), + }, + stopped = join_optional(&mut self.executor) => { + self.executor = None; + self.finish_stop(stopped, hooks); + } + () = lifeline_closed(&mut lifeline) => { + lifeline = None; + self_release = true; + self.start_cleanup(self.record.fallback_stop); + } + _tick = rotation_tick.tick(), if self.rotation.is_none() => { + match self.start_rotation(hooks) { + Ok(rotation) => self.rotation = Some(rotation), + Err(error) => self.log_rotation_error = Some(error.to_string()), + } + } + rotated = join_optional(&mut self.rotation) => { + self.rotation = None; + self.log_rotation_error = rotated.err().map(|error| error.to_string()); + } + } + } + } + + fn observe(&mut self, events: ProcessEvents, hooks: &MonitorHooks) { + let Some(status) = events.exit else { + return; + }; + self.watch = None; + self.exit = Some(MonitorExit::from(status)); + if self.stopped { + self.complete_cleanup(hooks); + } else { + // The runtime exited on its own: stop whatever it left in its group. + self.start_cleanup(self.record.fallback_stop); + } + } + + /// Starts the one stop executor of this instance, unless it already ran. A repeated stop + /// never resets its deadline, and a client disconnecting never cancels it. + fn start_cleanup(&mut self, stop: MonitorStop) { + if self.executor.is_some() || self.stopped || self.cleanup != MonitorCleanup::Pending { + return; + } + let pid = self.record.runtime.pid; + // The unreaped leader pins the group's id, so the group is still the runtime's. + self.executor = Some(tokio::spawn(async move { + stop_process_group(pid, stop.signal, stop.grace(), || Ok(true)).await + })); + } + + fn finish_stop(&mut self, stopped: Result<(), DaemonError>, hooks: &MonitorHooks) { + if stopped.is_err() { + self.cleanup = MonitorCleanup::Unproven; + return; + } + self.stopped = true; + // The leader is a zombie now, so its exit event is already queued if not yet observed. + if self.exit.is_some() { + self.complete_cleanup(hooks); + } + } + + /// Proves the group empty one final time, then reaps the leader, exactly once. + fn complete_cleanup(&mut self, hooks: &MonitorHooks) { + let pid = self.record.runtime.pid; + let group_empty = !hooks.unknown_group_check + && matches!(platform::process_group_has_live_members(pid), Ok(false)); + self.cleanup = if group_empty && reap_process_if_child(pid).is_ok() { + MonitorCleanup::Complete + } else { + MonitorCleanup::Unproven + }; + } + + fn releasable(&self) -> Result<(), String> { + if self.exit.is_some() && self.cleanup == MonitorCleanup::Complete { + return Ok(()); + } + + Err("the runtime has not exited with its cleanup complete".to_owned()) + } + + fn start_rotation( + &self, + hooks: &MonitorHooks, + ) -> Result>, DaemonError> { + let log = self.log.try_clone()?; + let path = self.record.log_path.clone(); + let threshold = hooks.rotation_bytes.unwrap_or(ROTATION_BYTES); + let hooks = hooks.clone(); + + Ok(tokio::task::spawn_blocking(move || { + rotation::rotate_if_needed(&log, &path, threshold, || { + pause_blocking(&hooks, MonitorPause::RotationCopy); + }) + })) + } +} + +async fn serve_connection(stream: UnixStream, instance: String, commands: mpsc::Sender) { + let _answer_result = answer(stream, &instance, &commands).await; +} + +async fn answer( + stream: UnixStream, + instance: &str, + commands: &mpsc::Sender, +) -> Result<(), DaemonError> { + if stream.peer_cred()?.uid() != rustix::process::getuid().as_raw() { + return Ok(()); + } + let mut connection = transport(stream); + let Some(line) = timeout(REQUEST_TIMEOUT, connection.next()) + .await + .map_err(|_elapsed| DaemonError::ProtocolTimedOut { + phase: "monitor request", + })? + else { + return Ok(()); + }; + let line = line.map_err(ProtocolError::from)?; + let mut released = false; + let result = match parse_request(&line, instance) { + Err((kind, message)) => ReplyResult::Error { kind, message }, + Ok(Operation::State) => { + let (reply, state) = oneshot::channel(); + send(commands, Command::State(reply)).await?; + ReplyResult::State(state.await.map_err(|_closed| monitor_closing())?) + } + Ok(Operation::Stop(stop)) => { + send(commands, Command::Stop(stop)).await?; + ReplyResult::Accepted + } + Ok(Operation::Release) => { + let (reply, releasable) = oneshot::channel(); + send(commands, Command::Release(reply)).await?; + match releasable.await.map_err(|_closed| monitor_closing())? { + Ok(()) => { + released = true; + ReplyResult::Accepted + } + Err(message) => ReplyResult::Error { + kind: MonitorErrorKind::NotReleasable, + message, + }, + } + } + }; + let written = write_line(&mut connection, &Reply::new(instance, result)).await; + if released { + send(commands, Command::Exit).await?; + } + written?; + + Ok(()) +} + +fn parse_request(line: &str, instance: &str) -> Result { + let invalid = |error: serde_json::Error| (MonitorErrorKind::InvalidRequest, error.to_string()); + let header: RequestHeader = serde_json::from_str(line).map_err(invalid)?; + if header.version != PROTOCOL_VERSION { + return Err(( + MonitorErrorKind::UnsupportedVersion, + format!( + "this monitor speaks protocol version {PROTOCOL_VERSION}, not {}", + header.version + ), + )); + } + if header.instance != instance { + return Err(( + MonitorErrorKind::InstanceMismatch, + "the request names another monitor instance".to_owned(), + )); + } + let request: Request = serde_json::from_str(line).map_err(invalid)?; + + Ok(request.operation) +} + +async fn send(commands: &mpsc::Sender, command: Command) -> Result<(), DaemonError> { + commands + .send(command) + .await + .map_err(|_closed| monitor_closing()) +} + +fn monitor_closing() -> DaemonError { + io::Error::other("the monitor is shutting down").into() +} + +async fn next_optional_events( + watch: &mut Option>, +) -> Result { + match watch { + Some(watch) => next_events(watch).await, + None => std::future::pending().await, + } +} + +async fn join_optional( + handle: &mut Option>>, +) -> Result<(), DaemonError> { + match handle { + Some(handle) => handle.await?, + None => std::future::pending().await, + } +} + +async fn lifeline_closed(lifeline: &mut Option>) { + match lifeline { + Some(closed) => { + let _closed_result = closed.await; + } + None => std::future::pending().await, + } +} + +fn lifeline_path(fd: i32) -> Result { + // Opening /dev/fd/N duplicates descriptor N, so the monitor never owns a raw descriptor. + let path = Utf8PathBuf::from(format!("/dev/fd/{fd}")); + let is_pipe = fd > 2 + && rustix::fs::stat(path.as_std_path()) + .is_ok_and(|stat| FileType::from_raw_mode(stat.st_mode) == FileType::Fifo); + if !is_pipe { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("lifeline descriptor {fd} is not an inherited pipe"), + ) + .into()); + } + + Ok(path) +} + +fn watch_lifeline(path: Utf8PathBuf) -> Result, DaemonError> { + let (closed, receiver) = oneshot::channel(); + std::thread::Builder::new() + .name("pv-monitor-lifeline".to_owned()) + .spawn(move || { + // Returns once every write end is closed: the controller holding it is gone. + let _read_result = fs::read_to_string(&path); + let _send_result = closed.send(()); + })?; + + Ok(receiver) +} + +async fn pause(hooks: &MonitorHooks, point: MonitorPause) { + if let Some((reached, resume)) = pause_files(hooks, point) { + announce(&reached); + while !fs::path_exists(&resume) { + sleep(POLL_INTERVAL).await; + } + } +} + +fn pause_blocking(hooks: &MonitorHooks, point: MonitorPause) { + if let Some((reached, resume)) = pause_files(hooks, point) { + announce(&reached); + while !fs::path_exists(&resume) { + std::thread::sleep(POLL_INTERVAL); + } + } +} + +fn pause_files(hooks: &MonitorHooks, point: MonitorPause) -> Option<(Utf8PathBuf, Utf8PathBuf)> { + let dir = hooks.pause_dir.as_ref()?; + if !hooks.pauses.contains(&point) { + return None; + } + let name = point.name(); + + Some(( + dir.join(format!("{name}.reached")), + dir.join(format!("{name}.continue")), + )) +} + +fn announce(reached: &Utf8Path) { + let _write_result = fs::write_sensitive_file(reached, &format!("{}\n", std::process::id())); +} + +impl From for MonitorExit { + fn from(status: ExitStatus) -> Self { + match status.code() { + Some(code) => Self::Code(code), + None => Self::Signal(status.signal().unwrap_or_default()), + } + } +} + +fn recorded_process(pid: u32) -> Result { + let start_identity = platform::inspect_process_start_identity(pid)?.ok_or_else(|| { + DaemonError::MissingProcessIdentity { + name: "monitor".to_owned(), + pid, + } + })?; + + Ok(RecordedProcess { + pid, + start_identity, + }) +} + +fn monitor_socket(paths: &PvPaths, subject: &str) -> Result { + let path = paths.monitor_dir(subject).join(SOCKET_FILE); + if path.as_str().len() > SOCKET_PATH_LIMIT { + return Err(DaemonError::MonitorSocketPathTooLong { + path, + limit: SOCKET_PATH_LIMIT, + }); + } + + Ok(path) +} + +fn record_path(paths: &PvPaths, subject: &str) -> Utf8PathBuf { + paths.monitor_dir(subject).join(RECORD_FILE) +} + +fn read_record(paths: &PvPaths, subject: &str) -> Result, DaemonError> { + match fs::read_to_string(&record_path(paths, subject)) { + Ok(content) => Ok(Some(serde_json::from_str(&content)?)), + Err(StateError::Filesystem { source, .. }) if source.kind() == io::ErrorKind::NotFound => { + Ok(None) + } + Err(error) => Err(error.into()), + } +} + +fn remove_monitor_dir(paths: &PvPaths, subject: &str) -> Result<(), DaemonError> { + match fs::delete_dir_all(&paths.monitor_dir(subject)) { + Err(StateError::Filesystem { source, .. }) if source.kind() == io::ErrorKind::NotFound => { + Ok(()) + } + result => Ok(result?), + } +} + +fn new_instance() -> Result { + let mut bytes = [0_u8; 16]; + getrandom::fill(&mut bytes).map_err(io::Error::other)?; + let mut instance = String::with_capacity(32); + for byte in bytes { + let _write_result = write!(instance, "{byte:02x}"); + } + + Ok(instance) +} + +fn reservation_error(subject: &str, error: StateError) -> DaemonError { + match error { + StateError::CoordinationLockHeld { .. } => { + unavailable(subject, "a monitor or controller holds its reservation") + } + error => error.into(), + } +} + +fn startup_failed(subject: &str, message: impl Into) -> DaemonError { + DaemonError::MonitorStartupFailed { + subject: subject.to_owned(), + message: message.into(), + } +} + +fn unavailable(subject: &str, reason: &str) -> DaemonError { + DaemonError::MonitorUnavailable { + subject: subject.to_owned(), + reason: reason.to_owned(), + } +} diff --git a/crates/daemon/src/monitor/rotation.rs b/crates/daemon/src/monitor/rotation.rs new file mode 100644 index 00000000..c934ad6d --- /dev/null +++ b/crates/daemon/src/monitor/rotation.rs @@ -0,0 +1,196 @@ +//! Rotates a runtime log the monitor captured. +//! +//! The runtime appends to the same open file the monitor retains, so rotation copies and +//! truncates that file in place rather than renaming it. Output written between the copy and the +//! truncation is lost; that window is accepted. + +use std::io::{self, Write as _}; +use std::os::unix::fs::FileExt as _; + +use camino::{Utf8Path, Utf8PathBuf}; +use state::fs; +use time::OffsetDateTime; +use time::macros::format_description; + +use crate::DaemonError; + +/// How many archives a log keeps. +const KEEP: usize = 5; +const COPY_BUFFER_BYTES: usize = 64 * 1024; +/// `YYYYMMDD-HHMMSS-mmm`. +const STAMP_LENGTH: usize = 19; + +/// Once `log` reaches `threshold` bytes, archives exactly those bytes as +/// `-YYYYMMDD-HHMMSS-mmm.log` in UTC, beside `path`, truncates `log`, and keeps the +/// newest archives. A failure leaves the active log untouched. `before_copy` runs once the +/// archive's temporary file exists. +#[expect( + clippy::disallowed_types, + reason = "rotation copies from the monitor's retained log descriptor" +)] +pub(super) fn rotate_if_needed( + log: &std::fs::File, + path: &Utf8Path, + threshold: u64, + before_copy: impl FnOnce(), +) -> Result<(), DaemonError> { + let size = log.metadata()?.len(); + if size < threshold { + return Ok(()); + } + let (Some(parent), Some(name)) = (path.parent(), path.file_name()) else { + return Err( + io::Error::new(io::ErrorKind::InvalidInput, "log path has no file name").into(), + ); + }; + let stem = name.strip_suffix(".log").unwrap_or(name); + // A dot file, so `pv logs` never shows a partial archive. + let temporary = parent.join(format!(".{name}.rotating")); + fs::remove_file_if_exists(&temporary)?; + let mut archive = fs::create_new_file(&temporary)?; + before_copy(); + if let Err(error) = copy_prefix(log, &mut archive, size) { + let _remove_result = fs::remove_file_if_exists(&temporary); + return Err(error); + } + let stamp = OffsetDateTime::now_utc().format(format_description!( + "[year][month][day]-[hour][minute][second]-[subsecond digits:3]" + ))?; + fs::rename(&temporary, &parent.join(format!("{stem}-{stamp}.log")))?; + log.set_len(0)?; + + prune(parent, stem) +} + +#[expect( + clippy::disallowed_types, + reason = "rotation copies from the monitor's retained log descriptor" +)] +fn copy_prefix( + log: &std::fs::File, + archive: &mut std::fs::File, + size: u64, +) -> Result<(), DaemonError> { + let mut buffer = vec![0; COPY_BUFFER_BYTES]; + let mut offset = 0; + while offset < size { + let wanted = + usize::try_from(size - offset).map_or(buffer.len(), |left| left.min(buffer.len())); + let read = log.read_at(&mut buffer[..wanted], offset)?; + if read == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "the log shrank while it was rotated", + ) + .into()); + } + archive.write_all(&buffer[..read])?; + offset += u64::try_from(read).map_err(io::Error::other)?; + } + archive.sync_all()?; + + Ok(()) +} + +fn prune(parent: &Utf8Path, stem: &str) -> Result<(), DaemonError> { + let prefix = format!("{stem}-"); + let mut archives = fs::read_dir_paths(parent)? + .into_iter() + .filter(|path| { + path.file_name() + .is_some_and(|name| is_archive(name, &prefix)) + }) + .collect::>(); + // Stamps sort in time order. + archives.sort(); + let excess = archives.len().saturating_sub(KEEP); + for archive in archives.iter().take(excess) { + fs::remove_file_if_exists(archive)?; + } + + Ok(()) +} + +fn is_archive(name: &str, prefix: &str) -> bool { + name.strip_prefix(prefix) + .and_then(|rest| rest.strip_suffix(".log")) + .is_some_and(|stamp| { + stamp.len() == STAMP_LENGTH + && stamp + .bytes() + .all(|byte| byte.is_ascii_digit() || byte == b'-') + }) +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use camino_tempfile::tempdir; + + use super::*; + + #[test] + fn rotation_keeps_the_newest_archives_and_appends_after_truncation() -> anyhow::Result<()> { + let tempdir = tempdir()?; + let path = tempdir.path().join("php-8.3.log"); + let log = fs::open_private_log_file(&path)?; + let unrelated = tempdir.path().join("php-8.4-20260101-000000-000.log"); + fs::write_sensitive_file(&unrelated, "another runtime\n")?; + + for round in 0..7 { + writeln!(&log, "round {round}")?; + rotate_if_needed(&log, &path, 1, || {})?; + // Archive names have millisecond stamps. + std::thread::sleep(Duration::from_millis(2)); + } + writeln!(&log, "after rotation")?; + rotate_if_needed(&log, &path, 1024, || {})?; + + let mut archives = fs::read_dir_paths(tempdir.path())? + .into_iter() + .filter(|archive| { + archive + .file_name() + .is_some_and(|name| is_archive(name, "php-8.3-")) + }) + .collect::>(); + archives.sort(); + let contents = archives + .iter() + .map(|archive| fs::read_to_string(archive)) + .collect::, _>>()?; + + assert_eq!( + contents, + [ + "round 2\n", + "round 3\n", + "round 4\n", + "round 5\n", + "round 6\n" + ] + ); + assert_eq!(fs::read_to_string(&path)?, "after rotation\n"); + assert!(fs::path_exists(&unrelated)); + + Ok(()) + } + + #[test] + fn failed_rotation_keeps_the_active_log() -> anyhow::Result<()> { + let tempdir = tempdir()?; + let path = tempdir.path().join("gateway.log"); + let log = fs::open_private_log_file(&path)?; + writeln!(&log, "kept")?; + // A directory where the archive's temporary file must go. + fs::ensure_user_dir(&tempdir.path().join(".gateway.log.rotating"))?; + + let rotated = rotate_if_needed(&log, &path, 1, || {}); + + assert!(rotated.is_err()); + assert_eq!(fs::read_to_string(&path)?, "kept\n"); + + Ok(()) + } +} diff --git a/crates/daemon/src/supervisor.rs b/crates/daemon/src/supervisor.rs index 29498333..f1e91ef4 100644 --- a/crates/daemon/src/supervisor.rs +++ b/crates/daemon/src/supervisor.rs @@ -31,12 +31,13 @@ const PROCESS_IDENTITY_TIMEOUT: Duration = Duration::from_secs(5); const SCRIPT_IDENTITY_STABILIZATION: Duration = Duration::from_millis(250); const PRIVATE_ENVIRONMENT_REDACTION: &str = ""; const PRIVATE_ENVIRONMENT_FINGERPRINT_PREFIX: &str = "sha256:v1:"; -const PHP_INI_ENVIRONMENT_KEYS: [&str; 2] = ["PHPRC", "PHP_INI_SCAN_DIR"]; +pub(crate) const PHP_INI_ENVIRONMENT_KEYS: [&str; 2] = ["PHPRC", "PHP_INI_SCAN_DIR"]; const POSTGRES_RECOVERY: &str = "Restart your Mac, then run this command again."; pub(crate) const RUNTIME_READINESS_CONCURRENCY_LIMIT: usize = 4; /// The signal that asks a runtime to shut down gracefully, before the grace period ends in SIGKILL. -#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] pub enum StopSignal { /// SIGTERM. Terminate, @@ -92,7 +93,7 @@ impl fmt::Debug for ProcessSpec { } } -struct PrivateEnvironmentDebug<'a>(&'a BTreeMap); +pub(crate) struct PrivateEnvironmentDebug<'a>(pub(crate) &'a BTreeMap); impl fmt::Debug for PrivateEnvironmentDebug<'_> { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { @@ -1114,7 +1115,11 @@ fn signal_process_group(pid: u32, signal: ProcessSignal) -> Result<(), DaemonErr Ok(()) => Ok(()), Err(source) => { let error = io::Error::from(source); - if process_not_found(&error) { + // macOS answers EPERM, not ESRCH, for a group whose only members are unreaped zombies. + if process_not_found(&error) + || (error.kind() == io::ErrorKind::PermissionDenied + && process_group_has_exited(pid)?) + { return Ok(()); } @@ -1219,17 +1224,28 @@ async fn stop_process_group_by_pid( signal: StopSignal, grace_period: Duration, ) -> Result<(), DaemonError> { - let pid = owned.pid; + // An adopted leader can be reaped during the grace period, by launchd once the + // daemon that started it has exited. Do not signal a group whose recorded + // leader no longer matches. + stop_process_group(owned.pid, signal, grace_period, || owned.matches_live()).await +} + +/// Signals `pid`'s process group with `signal`, then SIGKILL once `grace_period` passes, and +/// waits until no live member remains. Escalates only while `leader_still_owned` proves the +/// group still belongs to the runtime. +pub(crate) async fn stop_process_group( + pid: u32, + signal: StopSignal, + grace_period: Duration, + leader_still_owned: impl Fn() -> Result, +) -> Result<(), DaemonError> { signal_process_group(pid, ProcessSignal::Stop(signal))?; if wait_for_process_group_exit(pid, grace_period).await? { return Ok(()); } - // An adopted leader can be reaped during the grace period, by launchd once the - // daemon that started it has exited. Do not signal a group whose recorded - // leader no longer matches. - if !owned.matches_live()? { + if !leader_still_owned()? { return Err(DaemonError::RuntimeCleanupUnproven { pid, reason: "runtime leader changed before escalation; group ownership is unproven" @@ -1262,7 +1278,7 @@ async fn wait_for_managed_process_group_exit( Ok(()) } -async fn wait_for_process_group_exit( +pub(crate) async fn wait_for_process_group_exit( pid: u32, readiness_timeout: Duration, ) -> Result { @@ -1501,7 +1517,7 @@ fn process_and_group_are_absent(pid: u32) -> Result { } #[cfg(target_os = "macos")] -fn reap_process_if_child(pid: u32) -> Result<(), DaemonError> { +pub(crate) fn reap_process_if_child(pid: u32) -> Result<(), DaemonError> { let process = process_group_pid(pid)?; match waitpid(Some(process), WaitOptions::NOHANG) { Ok(_) | Err(rustix::io::Errno::CHILD) => Ok(()), @@ -1531,7 +1547,7 @@ fn process_and_group_are_absent(_pid: u32) -> Result { } #[cfg(any(target_os = "linux", target_os = "windows"))] -fn reap_process_if_child(_pid: u32) -> Result<(), DaemonError> { +pub(crate) fn reap_process_if_child(_pid: u32) -> Result<(), DaemonError> { require_process_containment() } diff --git a/crates/daemon/tests/monitor.rs b/crates/daemon/tests/monitor.rs new file mode 100644 index 00000000..b1238f4f --- /dev/null +++ b/crates/daemon/tests/monitor.rs @@ -0,0 +1,769 @@ +//! Process contract tests for per-runtime monitors, run through the `pv-monitor` example. +//! +//! Each monitor watches this test process's lifeline, so a test that fails part-way leaves no +//! monitor or runtime behind: when the test process exits, the monitor stops the runtime. + +#![cfg(target_os = "macos")] + +use std::collections::BTreeMap; +use std::os::fd::{AsRawFd as _, OwnedFd}; +use std::os::unix::process::CommandExt as _; +use std::time::Duration; + +use anyhow::{Result, anyhow, bail}; +use camino::{Utf8Path, Utf8PathBuf}; +use camino_tempfile::{Utf8TempDir, tempdir}; +use daemon::{ + DaemonError, MonitorCleanup, MonitorErrorKind, MonitorExit, MonitorHooks, MonitorPause, + MonitorStart, MonitorState, MonitorStop, StopSignal, +}; +use insta::{Settings, assert_snapshot}; +use pv_fake::{EventKind, FakeSettings, InstalledFake, Persona, Scenario}; +use rustix::io::{FdFlags, fcntl_setfd}; +use rustix::process::{Pid, Signal, kill_process, test_kill_process}; +use state::{MonitorReservation, PvPaths, StateError}; +use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader}; +use tokio::net::UnixStream; +use tokio::time::sleep; + +#[expect( + clippy::disallowed_types, + reason = "monitor tests spawn unrelated processes to prove what monitors never touch" +)] +type TestCommand = std::process::Command; + +const SUBJECT: &str = "gateway"; +const POLL_INTERVAL: Duration = Duration::from_millis(10); +const WAIT_ATTEMPTS: usize = 1_000; +const FALLBACK: MonitorStop = MonitorStop { + signal: StopSignal::Terminate, + grace_ms: 1_000, +}; +const STOP: MonitorStop = MonitorStop { + signal: StopSignal::Terminate, + grace_ms: 5_000, +}; + +#[tokio::test] +async fn monitor_runs_a_runtime_through_stop_and_release() -> Result<()> { + let harness = Harness::new()?; + let fake = harness.fake(Persona::LongRunning)?; + + let started = harness + .start(harness.start_for(fake.executable(), &[])?) + .await?; + let fake_pid = wait_for_fake_start(&fake).await?; + let record = harness.record()?; + let runtime_birth = platform::inspect_process_start_identity(started.runtime_pid)?; + let stopped = daemon::stop_monitor(&harness.paths, SUBJECT, STOP).await?; + daemon::release_monitor(&harness.paths, SUBJECT).await?; + + assert_eq!(started.cleanup, MonitorCleanup::Pending); + assert_eq!(started.exit, None); + // The runtime is the gate after exec: same pid, same birth identity. + assert_eq!(fake_pid, started.runtime_pid); + assert_eq!( + runtime_birth.map(|birth| serde_json::json!(birth)), + Some(record["runtime"]["start_identity"].clone()) + ); + assert_eq!(stopped.stop, Some(STOP)); + assert_eq!(stopped.exit, Some(MonitorExit::Code(0))); + assert_eq!(stopped.cleanup, MonitorCleanup::Complete); + assert!(!state::fs::path_exists(&harness.monitor_dir())); + assert!(harness.reservation_is_free()?); + let record = serde_json::to_string_pretty(&record)?; + harness.snapshot(|| assert_snapshot!(record)); + + Ok(()) +} + +#[tokio::test] +async fn monitor_requests_follow_the_frozen_contract() -> Result<()> { + let harness = Harness::new()?; + let fake = harness.fake(Persona::LongRunning)?; + harness + .start(harness.start_for(fake.executable(), &[])?) + .await?; + let instance = harness.record()?["instance"] + .as_str() + .ok_or_else(|| anyhow!("record has no instance"))? + .to_owned(); + + let mut replies = Vec::new(); + for request in [ + format!(r#"{{"version":1,"instance":"{instance}","op":"state","added_later":true}}"#), + format!(r#"{{"version":2,"instance":"{instance}","op":"state"}}"#), + r#"{"version":1,"instance":"another","op":"state"}"#.to_owned(), + format!(r#"{{"version":1,"instance":"{instance}","op":"restart"}}"#), + format!(r#"{{"version":1,"instance":"{instance}","op":"release"}}"#), + "not json".to_owned(), + ] { + let reply = harness.raw_request(&request).await?; + replies.push(format!("{request}\n{reply}")); + } + daemon::stop_monitor(&harness.paths, SUBJECT, STOP).await?; + daemon::release_monitor(&harness.paths, SUBJECT).await?; + + let mut settings = Settings::clone_current(); + settings.add_filter(&instance, ""); + settings.add_filter(r#""(monitor_pid|runtime_pid)":\d+"#, r#""$1":"#); + settings.bind(|| assert_snapshot!(replies.join("\n\n"))); + + Ok(()) +} + +#[tokio::test] +async fn monitor_start_reports_a_missing_command_and_leaves_no_instance() -> Result<()> { + let harness = Harness::new()?; + let missing = harness.paths.root().join("runtime/bin/missing"); + + let started = harness.start(harness.start_for(&missing, &[])?).await; + + let Err(DaemonError::MonitorStartupFailed { message, .. }) = started else { + bail!("expected a startup failure, got {started:?}"); + }; + harness.snapshot(|| assert_snapshot!(message)); + assert!(!state::fs::path_exists(&harness.monitor_dir())); + assert!(harness.reservation_is_free()?); + + Ok(()) +} + +#[tokio::test] +async fn monitor_retains_an_exit_nobody_was_connected_for() -> Result<()> { + let harness = Harness::new()?; + let shell = Utf8Path::new("/bin/sh"); + + harness + .start(harness.start_for(shell, &["-c", "exit 7"])?) + .await?; + let exited = harness + .wait_for_state(|state| state.cleanup != MonitorCleanup::Pending) + .await?; + daemon::release_monitor(&harness.paths, SUBJECT).await?; + + assert_eq!(exited.exit, Some(MonitorExit::Code(7))); + assert_eq!(exited.stop, None); + assert_eq!(exited.cleanup, MonitorCleanup::Complete); + assert!(harness.reservation_is_free()?); + + Ok(()) +} + +#[tokio::test] +async fn monitor_stop_outlives_its_client_and_keeps_its_first_deadline() -> Result<()> { + let harness = Harness::new()?; + let shell = Utf8Path::new("/bin/sh"); + harness + .start(harness.start_for( + shell, + &[ + "-c", + "trap '' TERM; echo ready; while :; do sleep 0.1; done", + ], + )?) + .await?; + harness.wait_for_log("ready").await?; + let instance = harness.instance()?; + + // Each raw request disconnects as soon as it is accepted. + harness + .raw_request(&format!( + r#"{{"version":1,"instance":"{instance}","op":"stop","signal":"terminate","grace_ms":200}}"# + )) + .await?; + harness + .raw_request(&format!( + r#"{{"version":1,"instance":"{instance}","op":"stop","signal":"interrupt","grace_ms":60000}}"# + )) + .await?; + let stopped = harness + .wait_for_state(|state| state.cleanup != MonitorCleanup::Pending) + .await?; + daemon::release_monitor(&harness.paths, SUBJECT).await?; + + assert_eq!( + stopped.stop, + Some(MonitorStop { + signal: StopSignal::Terminate, + grace_ms: 200, + }) + ); + assert_eq!(stopped.exit, Some(MonitorExit::Signal(9))); + assert_eq!(stopped.cleanup, MonitorCleanup::Complete); + + Ok(()) +} + +#[tokio::test] +async fn monitor_reports_a_leader_exit_while_its_group_lives_on() -> Result<()> { + let harness = Harness::new()?; + let shell = Utf8Path::new("/bin/sh"); + let mut start = harness.start_for(shell, &["-c", "trap '' TERM; sleep 30 & exit 3"])?; + start.fallback_stop.grace_ms = 3_000; + harness.start(start.clone()).await?; + + let exited = harness.wait_for_state(|state| state.exit.is_some()).await?; + let release = daemon::release_monitor(&harness.paths, SUBJECT).await; + let restart = harness.start(start).await; + let cleaned = harness + .wait_for_state(|state| state.cleanup != MonitorCleanup::Pending) + .await?; + daemon::release_monitor(&harness.paths, SUBJECT).await?; + + assert_eq!(exited.exit, Some(MonitorExit::Code(3))); + assert_eq!(exited.cleanup, MonitorCleanup::Pending); + assert!(matches!( + release, + Err(DaemonError::MonitorRejected { + kind: MonitorErrorKind::NotReleasable, + .. + }) + )); + assert!(matches!( + restart, + Err(DaemonError::MonitorUnavailable { .. }) + )); + assert_eq!(cleaned.cleanup, MonitorCleanup::Complete); + + Ok(()) +} + +#[tokio::test] +async fn monitor_keeps_its_records_when_group_cleanup_is_unproven() -> Result<()> { + let harness = Harness::with_hooks(MonitorHooks { + unknown_group_check: true, + ..MonitorHooks::default() + })?; + let fake = harness.fake(Persona::LongRunning)?; + let started = harness + .start(harness.start_for(fake.executable(), &[])?) + .await?; + wait_for_fake_start(&fake).await?; + + let stopped = daemon::stop_monitor(&harness.paths, SUBJECT, STOP).await?; + let release = daemon::release_monitor(&harness.paths, SUBJECT).await; + let records_kept = state::fs::path_exists(&harness.monitor_dir().join("monitor.json")); + kill(started.monitor_pid, Signal::KILL)?; + wait_until(|| harness.reservation_is_free()).await?; + daemon::recover_monitor(&harness.paths, SUBJECT).await?; + + assert_eq!(stopped.exit, Some(MonitorExit::Code(0))); + assert_eq!(stopped.cleanup, MonitorCleanup::Unproven); + assert!(matches!( + release, + Err(DaemonError::MonitorRejected { + kind: MonitorErrorKind::NotReleasable, + .. + }) + )); + assert!(records_kept); + assert!(!state::fs::path_exists(&harness.monitor_dir())); + + Ok(()) +} + +#[tokio::test] +async fn monitor_death_before_publication_starts_nothing() -> Result<()> { + let (harness, fake, started) = kill_monitor_at(MonitorPause::BeforePublication).await?; + + assert!( + matches!(started, Err(DaemonError::MonitorStartupFailed { .. })), + "{started:?}" + ); + assert!(!state::fs::path_exists(&harness.monitor_dir())); + assert!(fake.events()?.is_empty()); + + Ok(()) +} + +#[tokio::test] +async fn monitor_death_after_publication_is_recovered() -> Result<()> { + let (harness, fake, started) = kill_monitor_at(MonitorPause::BeforePermission).await?; + + assert!( + matches!(started, Err(DaemonError::MonitorStartupFailed { .. })), + "{started:?}" + ); + assert!(!state::fs::path_exists(&harness.monitor_dir())); + assert!(fake.events()?.is_empty()); + + Ok(()) +} + +#[tokio::test] +async fn monitor_death_after_permission_is_recovered() -> Result<()> { + let (harness, fake, started) = kill_monitor_at(MonitorPause::GateAfterPermission).await?; + + assert!( + matches!(started, Err(DaemonError::MonitorStartupFailed { .. })), + "{started:?}" + ); + assert!(!state::fs::path_exists(&harness.monitor_dir())); + assert!(fake.events()?.is_empty()); + + Ok(()) +} + +#[tokio::test] +async fn recovery_refuses_a_live_monitor_and_stops_a_killed_monitors_runtime() -> Result<()> { + let harness = Harness::new()?; + let fake = harness.fake(Persona::LongRunning)?; + let started = harness + .start(harness.start_for(fake.executable(), &[])?) + .await?; + wait_for_fake_start(&fake).await?; + + kill(started.monitor_pid, Signal::STOP)?; + let refused = daemon::recover_monitor(&harness.paths, SUBJECT).await; + kill(started.monitor_pid, Signal::KILL)?; + wait_until(|| harness.reservation_is_free()).await?; + daemon::recover_monitor(&harness.paths, SUBJECT).await?; + + assert!(matches!( + refused, + Err(DaemonError::MonitorUnavailable { .. }) + )); + // launchd reaps the orphaned runtime. + wait_until(|| Ok(!process_exists(started.runtime_pid)?)).await?; + + assert!( + fake.events()?.iter().any( + |event| matches!(&event.kind, EventKind::Signal { signal } if signal == "SIGTERM") + ) + ); + assert!(!state::fs::path_exists(&harness.monitor_dir())); + + Ok(()) +} + +#[tokio::test] +async fn recovery_never_signals_a_changed_or_previous_boot_runtime() -> Result<()> { + let harness = Harness::new()?; + let mut bystander = ChildGuard( + TestCommand::new("/bin/sleep") + .arg("30") + .process_group(0) + .spawn()?, + ); + let pid = bystander.0.id(); + let birth = platform::inspect_process_start_identity(pid)? + .ok_or_else(|| anyhow!("bystander {pid} has no identity"))?; + let current_boot = String::from(platform::current_boot_session_id()?); + let changed_birth = serde_json::json!({ + "seconds": 1, + "microseconds": 0, + }); + + // The exact runtime identity, but recorded during another boot. + harness.write_record( + pid, + &serde_json::json!(birth), + "00000000-0000-0000-0000-000000000001", + )?; + let previous_boot_recovery = daemon::recover_monitor(&harness.paths, SUBJECT).await; + let previous_boot_records = state::fs::path_exists(&harness.monitor_dir()); + harness.write_record(pid, &changed_birth, ¤t_boot)?; + let changed_recovery = daemon::recover_monitor(&harness.paths, SUBJECT).await; + + assert!(previous_boot_recovery.is_ok()); + assert!(!previous_boot_records); + assert!(matches!( + changed_recovery, + Err(DaemonError::RuntimeProcessIdentityChanged { pid: refused }) if refused == pid + )); + assert!(bystander.0.try_wait()?.is_none()); + + Ok(()) +} + +#[tokio::test] +async fn concurrent_spawns_never_inherit_the_reservation() -> Result<()> { + let harness = Harness::with_hooks(MonitorHooks { + pauses: vec![MonitorPause::BeforePublication], + ..MonitorHooks::default() + })?; + let fake = harness.fake(Persona::LongRunning)?; + let start = harness.start_for(fake.executable(), &[])?; + + let (started, bystander) = tokio::join!(harness.start(start), async { + harness.reached(MonitorPause::BeforePublication).await?; + // Spawned while the controller holds the reservation and the monitor's startup is open. + let bystander = ChildGuard(TestCommand::new("/bin/sleep").arg("30").spawn()?); + harness.resume(MonitorPause::BeforePublication)?; + anyhow::Ok(bystander) + }); + let _bystander = bystander?; + started?; + daemon::stop_monitor(&harness.paths, SUBJECT, STOP).await?; + // Release waits for the reservation, which a bystander holding it would block. + daemon::release_monitor(&harness.paths, SUBJECT).await?; + + assert!(harness.reservation_is_free()?); + + Ok(()) +} + +#[tokio::test] +async fn closed_lifeline_stops_the_runtime_and_releases_the_monitor() -> Result<()> { + let harness = Harness::new()?; + let executable = harness.paths.root().join("runtime/bin/runtime"); + // The fake's own lifeline is off, so only the monitor can stop it. + let fake = pv_fake::install_with( + &pv_fake::binary()?, + &executable, + &Scenario { + persona: Persona::LongRunning, + lifeline_fd: None, + settings: FakeSettings::default(), + }, + )?; + let (read, write) = rustix::pipe::pipe()?; + fcntl_setfd(&write, FdFlags::CLOEXEC)?; + let mut start = harness.start_for(fake.executable(), &[])?; + start.lifeline_fd = Some(read.as_raw_fd()); + let started = harness.start(start).await?; + drop(read); + wait_for_fake_start(&fake).await?; + + // The controller dies. + drop::(write); + wait_until(|| Ok(!state::fs::path_exists(&harness.monitor_dir()))).await?; + wait_until(|| harness.reservation_is_free()).await?; + + assert!( + fake.events()?.iter().any( + |event| matches!(&event.kind, EventKind::Signal { signal } if signal == "SIGTERM") + ) + ); + assert!(!process_exists(started.runtime_pid)?); + + Ok(()) +} + +#[tokio::test] +async fn overlong_socket_paths_are_rejected_before_anything_starts() -> Result<()> { + let tempdir = tempdir()?; + let paths = PvPaths::for_home(tempdir.path().join("a".repeat(80))); + let start = Harness::new()?.start_for(Utf8Path::new("/bin/sh"), &[])?; + + let started = daemon::start_monitor(&paths, Utf8Path::new("/nonexistent"), start).await; + + assert!(matches!( + started, + Err(DaemonError::MonitorSocketPathTooLong { limit: 103, .. }) + )); + assert!(!state::fs::path_exists(&paths.monitors())); + + Ok(()) +} + +#[tokio::test] +async fn symlinked_logs_are_rejected_before_the_runtime_starts() -> Result<()> { + let harness = Harness::new()?; + let fake = harness.fake(Persona::LongRunning)?; + let target = harness.tempdir.path().join("elsewhere.log"); + state::fs::write_sensitive_file(&target, "")?; + let start = harness.start_for(fake.executable(), &[])?; + symlink(&target, &start.log_path)?; + + let started = harness.start(start).await; + + assert!(matches!( + started, + Err(DaemonError::MonitorStartupFailed { .. }) + )); + assert_eq!(state::fs::read_to_string(&target)?, ""); + assert!(fake.events()?.is_empty()); + assert!(!state::fs::path_exists(&harness.monitor_dir())); + + Ok(()) +} + +#[tokio::test] +async fn held_rotation_still_lets_the_monitor_exit() -> Result<()> { + let harness = Harness::with_hooks(MonitorHooks { + pauses: vec![MonitorPause::RotationCopy], + rotation_bytes: Some(1), + rotation_interval_ms: Some(20), + ..MonitorHooks::default() + })?; + let shell = Utf8Path::new("/bin/sh"); + harness + .start(harness.start_for(shell, &["-c", "echo started; while :; do sleep 0.1; done"])?) + .await?; + + harness.reached(MonitorPause::RotationCopy).await?; + daemon::stop_monitor(&harness.paths, SUBJECT, STOP).await?; + daemon::release_monitor(&harness.paths, SUBJECT).await?; + + assert!(harness.reservation_is_free()?); + assert!(!state::fs::path_exists(&harness.monitor_dir())); + + Ok(()) +} + +/// Kills the monitor once it pauses at `point`, then returns what the controller saw. +async fn kill_monitor_at( + point: MonitorPause, +) -> Result<(Harness, InstalledFake, Result)> { + let harness = Harness::with_hooks(MonitorHooks { + pauses: vec![point], + ..MonitorHooks::default() + })?; + let fake = harness.fake(Persona::LongRunning)?; + let start = harness.start_for(fake.executable(), &[])?; + + let (started, killed) = tokio::join!(harness.start(start), async { + let paused = harness.reached(point).await?; + let monitor = if point == MonitorPause::GateAfterPermission { + harness.record()?["monitor"]["pid"] + .as_u64() + .and_then(|pid| u32::try_from(pid).ok()) + .ok_or_else(|| anyhow!("record has no monitor pid"))? + } else { + paused + }; + kill(monitor, Signal::KILL) + }); + killed?; + + Ok((harness, fake, started)) +} + +struct Harness { + tempdir: Utf8TempDir, + paths: PvPaths, + monitor: Utf8PathBuf, +} + +impl Harness { + fn new() -> Result { + Self::with_hooks(MonitorHooks::default()) + } + + fn with_hooks(mut hooks: MonitorHooks) -> Result { + let tempdir = tempdir()?; + let paths = PvPaths::for_home(tempdir.path().join("home")); + state::fs::ensure_layout(&paths)?; + let pauses = tempdir.path().join("pauses"); + state::fs::ensure_user_dir(&pauses)?; + hooks.pause_dir = Some(pauses); + state::fs::write_sensitive_file( + &paths.home().join("pv-monitor-hooks.json"), + &serde_json::to_string(&hooks)?, + )?; + + Ok(Self { + tempdir, + paths, + monitor: pv_fake::example_binary("pv-monitor")?, + }) + } + + fn start_for(&self, command: &Utf8Path, arguments: &[&str]) -> Result { + Ok(MonitorStart { + subject: SUBJECT.to_owned(), + command: command.to_owned(), + arguments: arguments.iter().map(ToString::to_string).collect(), + private_environment: BTreeMap::new(), + log_path: self.paths.logs().join("runtime.log"), + fallback_stop: FALLBACK, + lifeline_fd: Some(pv_fake::lifeline_fd()?), + }) + } + + fn fake(&self, persona: Persona) -> Result { + pv_fake::install(&self.paths.root().join("runtime/bin/runtime"), persona) + } + + async fn start(&self, start: MonitorStart) -> Result { + daemon::start_monitor(&self.paths, &self.monitor, start).await + } + + fn monitor_dir(&self) -> Utf8PathBuf { + self.paths.monitor_dir(SUBJECT) + } + + fn record(&self) -> Result { + let record = state::fs::read_to_string(&self.monitor_dir().join("monitor.json"))?; + + Ok(serde_json::from_str(&record)?) + } + + fn instance(&self) -> Result { + self.record()?["instance"] + .as_str() + .map(ToOwned::to_owned) + .ok_or_else(|| anyhow!("record has no instance")) + } + + /// Writes a record as a dead monitor would have left it, naming `pid` as its runtime. + fn write_record( + &self, + pid: u32, + start_identity: &serde_json::Value, + boot_session_id: &str, + ) -> Result<()> { + let record = serde_json::json!({ + "version": 1, + "subject": SUBJECT, + "instance": "0123456789abcdef0123456789abcdef", + "boot_session_id": boot_session_id, + "monitor": { "pid": pid, "start_identity": start_identity }, + "runtime": { "pid": pid, "start_identity": start_identity }, + "command": "/bin/sleep", + "arguments": ["30"], + "log_path": self.paths.logs().join("runtime.log"), + "fallback_stop": { "signal": "terminate", "grace_ms": 1_000 }, + }); + state::fs::write_sensitive_file( + &self.monitor_dir().join("monitor.json"), + &serde_json::to_string_pretty(&record)?, + )?; + + Ok(()) + } + + async fn raw_request(&self, line: &str) -> Result { + let stream = UnixStream::connect(self.monitor_dir().join("sock")).await?; + let (read, mut write) = stream.into_split(); + write.write_all(format!("{line}\n").as_bytes()).await?; + let mut reply = String::new(); + BufReader::new(read).read_line(&mut reply).await?; + + Ok(reply.trim_end().to_owned()) + } + + async fn wait_for_state(&self, done: impl Fn(&MonitorState) -> bool) -> Result { + for _attempt in 0..WAIT_ATTEMPTS { + let state = daemon::monitor_state(&self.paths, SUBJECT).await?; + if done(&state) { + return Ok(state); + } + sleep(POLL_INTERVAL).await; + } + + Err(anyhow!("monitor never reached the expected state")) + } + + async fn wait_for_log(&self, line: &str) -> Result<()> { + let log = self.paths.logs().join("runtime.log"); + wait_until(|| { + Ok(state::fs::read_to_string(&log) + .is_ok_and(|content| content.lines().any(|logged| logged == line))) + }) + .await + } + + /// Waits for the monitor or gate to pause at `point`, and returns the paused pid. + async fn reached(&self, point: MonitorPause) -> Result { + let reached = self.pause_file(point, "reached")?; + wait_until(|| Ok(state::fs::path_exists(&reached))).await?; + + Ok(state::fs::read_to_string(&reached)?.trim().parse()?) + } + + fn resume(&self, point: MonitorPause) -> Result<()> { + state::fs::write_sensitive_file(&self.pause_file(point, "continue")?, "")?; + + Ok(()) + } + + fn pause_file(&self, point: MonitorPause, suffix: &str) -> Result { + let name = serde_json::to_value(point)? + .as_str() + .map(ToOwned::to_owned) + .ok_or_else(|| anyhow!("pause point is not a string"))?; + + Ok(self.tempdir.path().join(format!("pauses/{name}.{suffix}"))) + } + + fn reservation_is_free(&self) -> Result { + match MonitorReservation::acquire(&self.paths, SUBJECT) { + Ok(_reservation) => Ok(true), + Err(StateError::CoordinationLockHeld { .. }) => Ok(false), + Err(error) => Err(error.into()), + } + } + + fn snapshot(&self, assertion: impl FnOnce()) { + let mut settings = Settings::clone_current(); + settings.add_filter(self.tempdir.path().as_str(), ""); + settings.add_filter(r#""(pid|seconds|microseconds)": \d+"#, r#""$1": "#); + settings.add_filter( + r#""instance": "[0-9a-f]{32}""#, + r#""instance": """#, + ); + settings.add_filter( + r#""boot_session_id": "[0-9A-F-]{36}""#, + r#""boot_session_id": """#, + ); + + settings.bind(assertion) + } +} + +/// Waits for the fake to record its start, and returns its pid. +async fn wait_for_fake_start(fake: &InstalledFake) -> Result { + for _attempt in 0..WAIT_ATTEMPTS { + let started = fake + .events()? + .into_iter() + .find(|event| matches!(event.kind, EventKind::Started { .. })); + if let Some(event) = started { + return Ok(u32::try_from(event.pid)?); + } + sleep(POLL_INTERVAL).await; + } + + Err(anyhow!("fake {} never started", fake.executable())) +} + +async fn wait_until(done: impl Fn() -> Result) -> Result<()> { + for _attempt in 0..WAIT_ATTEMPTS { + if done()? { + return Ok(()); + } + sleep(POLL_INTERVAL).await; + } + + Err(anyhow!("condition never held")) +} + +fn kill(pid: u32, signal: Signal) -> Result<()> { + kill_process(pid_of(pid)?, signal)?; + + Ok(()) +} + +fn process_exists(pid: u32) -> Result { + match test_kill_process(pid_of(pid)?) { + Ok(()) => Ok(true), + Err(rustix::io::Errno::SRCH) => Ok(false), + Err(error) => Err(error.into()), + } +} + +fn pid_of(pid: u32) -> Result { + Pid::from_raw(i32::try_from(pid)?).ok_or_else(|| anyhow!("pid {pid} is not positive")) +} + +#[expect( + clippy::disallowed_methods, + reason = "monitor tests plant a symlinked log the monitor must refuse" +)] +fn symlink(target: &Utf8Path, link: &Utf8Path) -> Result<()> { + std::os::unix::fs::symlink(target, link)?; + + Ok(()) +} + +struct ChildGuard(std::process::Child); + +impl Drop for ChildGuard { + fn drop(&mut self) { + let _kill_result = self.0.kill(); + let _wait_result = self.0.wait(); + } +} diff --git a/crates/daemon/tests/snapshots/monitor__monitor_requests_follow_the_frozen_contract.snap b/crates/daemon/tests/snapshots/monitor__monitor_requests_follow_the_frozen_contract.snap new file mode 100644 index 00000000..fc098921 --- /dev/null +++ b/crates/daemon/tests/snapshots/monitor__monitor_requests_follow_the_frozen_contract.snap @@ -0,0 +1,21 @@ +--- +source: crates/daemon/tests/monitor.rs +expression: "replies.join(\"\\n\\n\")" +--- +{"version":1,"instance":"","op":"state","added_later":true} +{"version":1,"instance":"","result":"state","subject":"gateway","monitor_pid":,"runtime_pid":,"cleanup":"pending"} + +{"version":2,"instance":"","op":"state"} +{"version":1,"instance":"","result":"error","kind":"unsupported_version","message":"this monitor speaks protocol version 1, not 2"} + +{"version":1,"instance":"another","op":"state"} +{"version":1,"instance":"","result":"error","kind":"instance_mismatch","message":"the request names another monitor instance"} + +{"version":1,"instance":"","op":"restart"} +{"version":1,"instance":"","result":"error","kind":"invalid_request","message":"unknown variant `restart`, expected one of `state`, `stop`, `release` at line 1 column 74"} + +{"version":1,"instance":"","op":"release"} +{"version":1,"instance":"","result":"error","kind":"not_releasable","message":"the runtime has not exited with its cleanup complete"} + +not json +{"version":1,"instance":"","result":"error","kind":"invalid_request","message":"expected ident at line 1 column 2"} diff --git a/crates/daemon/tests/snapshots/monitor__monitor_runs_a_runtime_through_stop_and_release.snap b/crates/daemon/tests/snapshots/monitor__monitor_runs_a_runtime_through_stop_and_release.snap new file mode 100644 index 00000000..49af46d9 --- /dev/null +++ b/crates/daemon/tests/snapshots/monitor__monitor_runs_a_runtime_through_stop_and_release.snap @@ -0,0 +1,31 @@ +--- +source: crates/daemon/tests/monitor.rs +expression: record +--- +{ + "arguments": [], + "boot_session_id": "", + "command": "/home/.pv/runtime/bin/runtime", + "fallback_stop": { + "grace_ms": 1000, + "signal": "terminate" + }, + "instance": "", + "log_path": "/home/.pv/logs/runtime.log", + "monitor": { + "pid": , + "start_identity": { + "microseconds": , + "seconds": + } + }, + "runtime": { + "pid": , + "start_identity": { + "microseconds": , + "seconds": + } + }, + "subject": "gateway", + "version": 1 +} diff --git a/crates/daemon/tests/snapshots/monitor__monitor_start_reports_a_missing_command_and_leaves_no_instance.snap b/crates/daemon/tests/snapshots/monitor__monitor_start_reports_a_missing_command_and_leaves_no_instance.snap new file mode 100644 index 00000000..b10a6c98 --- /dev/null +++ b/crates/daemon/tests/snapshots/monitor__monitor_start_reports_a_missing_command_and_leaves_no_instance.snap @@ -0,0 +1,5 @@ +--- +source: crates/daemon/tests/monitor.rs +expression: message +--- +cannot start /home/.pv/runtime/bin/missing: No such file or directory (os error 2) diff --git a/crates/pv-fake/src/install.rs b/crates/pv-fake/src/install.rs index 112cc593..f6b28dcc 100644 --- a/crates/pv-fake/src/install.rs +++ b/crates/pv-fake/src/install.rs @@ -74,23 +74,34 @@ pub fn install_with( } /// The daemon's `pv-fake` example, which Cargo builds along with the daemon's tests. +pub fn binary() -> Result { + example_binary("pv-fake") +} + +/// The daemon's example `name`, which Cargo builds along with the daemon's tests. #[expect( clippy::disallowed_methods, reason = "pv-fake finds the example target next to the running test executable" )] -pub fn binary() -> Result { +pub fn example_binary(name: &str) -> Result { let test_executable = Utf8PathBuf::try_from(std::env::current_exe()?)?; // Test executables run from target//deps; examples live in target//examples. let Some(profile_dir) = test_executable.parent().and_then(Utf8Path::parent) else { bail!("test executable {test_executable} is not inside a Cargo target directory"); }; - let binary = profile_dir.join("examples/pv-fake"); + let binary = profile_dir.join("examples").join(name); if !state::fs::path_is_file(&binary)? { bail!( - "pv-fake binary {binary} is missing; `cargo nextest run -p daemon` builds the daemon's \ + "{name} binary {binary} is missing; `cargo nextest run -p daemon` builds the daemon's \ examples, `cargo test --lib` does not" ); } Ok(binary) } + +/// The inheritable read end of this test process's lifeline, for processes a test starts +/// outside a fake, such as a runtime monitor. +pub fn lifeline_fd() -> Result { + lifeline::test_process_read_fd() +} diff --git a/crates/pv-fake/src/lib.rs b/crates/pv-fake/src/lib.rs index 0c48b13a..a60e62ab 100644 --- a/crates/pv-fake/src/lib.rs +++ b/crates/pv-fake/src/lib.rs @@ -47,7 +47,10 @@ pub use events::{Event, EventKind}; #[cfg(unix)] pub use gateway::write_gateway_control; #[cfg(unix)] -pub use install::{InstalledFake, binary, install, install_with, install_with_settings}; +pub use install::{ + InstalledFake, binary, example_binary, install, install_with, install_with_settings, + lifeline_fd, +}; /// The argument a fake starts its descendant with, followed by the descendant's parent pipe. #[cfg(unix)] From 259602b0cfb1786bdd2d08eedfaa1d173213d172 Mon Sep 17 00:00:00 2001 From: Clovis Muneza Date: Fri, 9 Oct 2026 20:52:35 -0400 Subject: [PATCH 3/6] feat(cli): route hidden monitor entry points `pv monitor:run` and `pv monitor:gate` run the monitor engine. Both need process containment and skip runtime lifecycle admission, which a monitor holding it for its whole life would keep `daemon:disable` from taking. A smoke test drives the real binary through start, stop and release. --- Cargo.lock | 3 ++ Cargo.toml | 3 ++ crates/cli/src/args.rs | 14 +++++++ crates/cli/src/commands/daemon.rs | 16 ++++++++ crates/cli/src/commands/mod.rs | 15 +++++++ it/cli.rs | 40 +++++++++++++++++++ ...cli__completions_generate_bash_script.snap | 36 ++++++++++++++++- ...cli__completions_generate_fish_script.snap | 6 +++ .../cli__completions_generate_zsh_script.snap | 26 ++++++++++++ 9 files changed, 158 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index 9d6ca912..cc282875 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3097,10 +3097,13 @@ dependencies = [ "camino", "camino-tempfile", "cli", + "daemon", "insta", "portable-pty", + "pv-fake", "serde_json", "state", + "tokio", "vt100", ] diff --git a/Cargo.toml b/Cargo.toml index 8cfd4550..969eb006 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -114,10 +114,13 @@ anyhow = { workspace = true } assert_cmd = { workspace = true } camino = { workspace = true } camino-tempfile = { workspace = true } +daemon = { path = "crates/daemon" } insta = { workspace = true } portable-pty = { workspace = true } +pv-fake = { path = "crates/pv-fake" } serde_json = { workspace = true } state = { path = "crates/state" } +tokio = { workspace = true } vt100 = { workspace = true } [[test]] diff --git a/crates/cli/src/args.rs b/crates/cli/src/args.rs index 2178e289..cb9a329f 100644 --- a/crates/cli/src/args.rs +++ b/crates/cli/src/args.rs @@ -46,6 +46,20 @@ pub(crate) enum Command { #[command(name = "daemon:run", about = "Run the internal PV daemon", hide = true)] DaemonRun, + #[command( + name = "monitor:run", + about = "Run an internal PV runtime monitor", + hide = true + )] + MonitorRun, + + #[command( + name = "monitor:gate", + about = "Start a runtime for an internal PV runtime monitor", + hide = true + )] + MonitorGate, + #[command( name = "shim:php", about = "Run the internal PV PHP shim", diff --git a/crates/cli/src/commands/daemon.rs b/crates/cli/src/commands/daemon.rs index 7dc62585..a7b44486 100644 --- a/crates/cli/src/commands/daemon.rs +++ b/crates/cli/src/commands/daemon.rs @@ -1,6 +1,7 @@ use std::process::ExitCode; use std::time::Duration; +use ::daemon::MonitorHooks; use camino::Utf8PathBuf; use platform::{LaunchAgentConfig, LaunchAgentFileState}; use state::{PvPaths, StateError}; @@ -203,6 +204,21 @@ pub(crate) fn run() -> Result { Ok(ExitCode::SUCCESS) } +pub(crate) fn run_monitor() -> Result { + let paths = PvPaths::default_home()?; + + ::daemon::run_monitor_blocking(paths, MonitorHooks::default())?; + + Ok(ExitCode::SUCCESS) +} + +/// Returns only when the runtime could not start: on success, the runtime replaces this process. +pub(crate) fn run_monitor_gate() -> Result { + ::daemon::run_monitor_gate(&MonitorHooks::default())?; + + Ok(ExitCode::SUCCESS) +} + /// Returns success when PV may touch the LaunchAgent file; otherwise writes /// the refusal to stderr and returns failure, for a file PV does not own or /// cannot read. diff --git a/crates/cli/src/commands/mod.rs b/crates/cli/src/commands/mod.rs index c23212dd..ffe134f7 100644 --- a/crates/cli/src/commands/mod.rs +++ b/crates/cli/src/commands/mod.rs @@ -65,6 +65,8 @@ where Command::DaemonDisable => daemon::disable(environment, streams), Command::DaemonRestart => daemon::restart(environment, streams), Command::DaemonRun => daemon::run(), + Command::MonitorRun => daemon::run_monitor(), + Command::MonitorGate => daemon::run_monitor_gate(), Command::ShimPhp(args) => php::shim(args, environment, runtime_lifecycle_lock), Command::ShimComposer(args) => composer::shim(args, environment, runtime_lifecycle_lock), Command::DnsStatus => dns::status(environment, streams), @@ -145,10 +147,14 @@ fn acquire_runtime_lifecycle( command: &Command, environment: &impl Environment, ) -> Result, ExecuteError> { + // A monitor and its runtime outlive any command, so holding admission for their whole life + // would block `daemon:disable`; their controller holds it while starting them. if !cfg!(target_os = "macos") || matches!( command, Command::DaemonRun + | Command::MonitorRun + | Command::MonitorGate | Command::Uninstall(_) | Command::Env(_) | Command::Completions(_) @@ -198,6 +204,7 @@ fn required_capability(command: &Command) -> Option { Command::DaemonRun | Command::Link(_) | Command::Unlink(_) => { Some(PlatformCapability::DaemonIpc) } + Command::MonitorRun | Command::MonitorGate => Some(PlatformCapability::ProcessContainment), Command::PortsStatus(_) | Command::PortsInstall | Command::PortsUninstall => { Some(PlatformCapability::LowPortFrontend) } @@ -650,6 +657,14 @@ mod tests { ); } + #[test] + fn required_capability_maps_runtime_monitor_commands() { + assert_required_capability( + &[Command::MonitorRun, Command::MonitorGate], + Some(PlatformCapability::ProcessContainment), + ); + } + #[test] fn required_capability_maps_internal_update_like_update() { assert_required_capability( diff --git a/it/cli.rs b/it/cli.rs index f728c023..4a4454b4 100644 --- a/it/cli.rs +++ b/it/cli.rs @@ -1,10 +1,16 @@ +#[cfg(target_os = "macos")] +use std::collections::BTreeMap; use std::process::ExitStatus; use anyhow::Result; use assert_cmd::Command; use camino::Utf8Path; use camino_tempfile::tempdir; +#[cfg(target_os = "macos")] +use daemon::{MonitorCleanup, MonitorExit, MonitorStart, MonitorStop, StopSignal}; use insta::{assert_debug_snapshot, assert_snapshot}; +#[cfg(target_os = "macos")] +use state::MonitorReservation; use state::{ Database, ProjectEnvObservedStatus, ProjectEnvObservedWarningInput, PvPaths, UpdateLock, }; @@ -221,6 +227,40 @@ fn daemon_run_is_hidden_from_top_level_help() -> Result<()> { Ok(()) } +#[cfg(target_os = "macos")] +#[tokio::test] +async fn hidden_monitor_commands_run_a_runtime_until_released() -> Result<()> { + let tempdir = tempdir()?; + let paths = PvPaths::for_home(tempdir.path().join("home")); + state::fs::ensure_layout(&paths)?; + let stop = MonitorStop { + signal: StopSignal::Terminate, + grace_ms: 5_000, + }; + let start = MonitorStart { + subject: "smoke".to_owned(), + command: "/bin/sh".into(), + arguments: vec!["-c".to_owned(), "while :; do sleep 0.1; done".to_owned()], + private_environment: BTreeMap::new(), + log_path: paths.logs().join("smoke.log"), + fallback_stop: stop, + lifeline_fd: Some(pv_fake::lifeline_fd()?), + }; + + let started = + daemon::start_monitor(&paths, Utf8Path::new(env!("CARGO_BIN_EXE_pv")), start).await?; + let stopped = daemon::stop_monitor(&paths, "smoke", stop).await?; + daemon::release_monitor(&paths, "smoke").await?; + + assert_eq!(started.cleanup, MonitorCleanup::Pending); + assert_eq!(stopped.exit, Some(MonitorExit::Signal(15))); + assert_eq!(stopped.cleanup, MonitorCleanup::Complete); + // The reservation is free only once the monitor process has exited. + assert!(MonitorReservation::acquire(&paths, "smoke").is_ok()); + + Ok(()) +} + #[test] fn daemon_lifecycle_commands_are_documented_without_running_them() -> Result<()> { let output = [ diff --git a/it/snapshots/cli__completions_generate_bash_script.snap b/it/snapshots/cli__completions_generate_bash_script.snap index 474dba87..1ce04b2f 100644 --- a/it/snapshots/cli__completions_generate_bash_script.snap +++ b/it/snapshots/cli__completions_generate_bash_script.snap @@ -116,6 +116,12 @@ _pv() { pv,mailpit:update) cmd="pv__subcmd__mailpit:update" ;; + pv,monitor:gate) + cmd="pv__subcmd__monitor:gate" + ;; + pv,monitor:run) + cmd="pv__subcmd__monitor:run" + ;; pv,mysql:install) cmd="pv__subcmd__mysql:install" ;; @@ -252,7 +258,7 @@ _pv() { case "${cmd}" in pv) - opts="-h -V --no-color --help --version env completions setup uninstall daemon:enable daemon:disable daemon:restart daemon:run shim:php shim:composer dns:status dns:install dns:uninstall ports:status ports:install ports:uninstall ca:status ca:trust ca:untrust link init unlink open project:env status logs doctor jobs update internal:update-managed-resources list php:use php:install php:update php:uninstall php:list composer:install composer:update composer:uninstall mailpit:install mailpit:update mailpit:uninstall mailpit:list mailpit:open mail:install mail:update mail:uninstall mail:list mail:open redis:install redis:update redis:uninstall redis:list rustfs:install rustfs:update rustfs:uninstall rustfs:list rustfs:open s3:install s3:update s3:uninstall s3:list s3:open mysql:install mysql:update mysql:uninstall mysql:list postgres:install postgres:update postgres:uninstall postgres:list pg:install pg:update pg:uninstall pg:list" + opts="-h -V --no-color --help --version env completions setup uninstall daemon:enable daemon:disable daemon:restart daemon:run monitor:run monitor:gate shim:php shim:composer dns:status dns:install dns:uninstall ports:status ports:install ports:uninstall ca:status ca:trust ca:untrust link init unlink open project:env status logs doctor jobs update internal:update-managed-resources list php:use php:install php:update php:uninstall php:list composer:install composer:update composer:uninstall mailpit:install mailpit:update mailpit:uninstall mailpit:list mailpit:open mail:install mail:update mail:uninstall mail:list mail:open redis:install redis:update redis:uninstall redis:list rustfs:install rustfs:update rustfs:uninstall rustfs:list rustfs:open s3:install s3:update s3:uninstall s3:list s3:open mysql:install mysql:update mysql:uninstall mysql:list postgres:install postgres:update postgres:uninstall postgres:list pg:install pg:update pg:uninstall pg:list" if [[ ${cur} == -* || ${COMP_CWORD} -eq 1 ]] ; then COMPREPLY=( $(compgen -W "${opts}" -- "${cur}") ) return 0 @@ -741,6 +747,34 @@ _pv() { COMPREPLY=( $(compgen -W "${opts}" -- "${cur}") ) return 0 ;; + pv__subcmd__monitor:gate) + opts="-h --no-color --help" + if [[ ${cur} == -* || ${COMP_CWORD} -eq 2 ]] ; then + COMPREPLY=( $(compgen -W "${opts}" -- "${cur}") ) + return 0 + fi + case "${prev}" in + *) + COMPREPLY=() + ;; + esac + COMPREPLY=( $(compgen -W "${opts}" -- "${cur}") ) + return 0 + ;; + pv__subcmd__monitor:run) + opts="-h --no-color --help" + if [[ ${cur} == -* || ${COMP_CWORD} -eq 2 ]] ; then + COMPREPLY=( $(compgen -W "${opts}" -- "${cur}") ) + return 0 + fi + case "${prev}" in + *) + COMPREPLY=() + ;; + esac + COMPREPLY=( $(compgen -W "${opts}" -- "${cur}") ) + return 0 + ;; pv__subcmd__mysql:install) opts="-h --no-color --help [track]" if [[ ${cur} == -* || ${COMP_CWORD} -eq 2 ]] ; then diff --git a/it/snapshots/cli__completions_generate_fish_script.snap b/it/snapshots/cli__completions_generate_fish_script.snap index bc4e49e4..278f9556 100644 --- a/it/snapshots/cli__completions_generate_fish_script.snap +++ b/it/snapshots/cli__completions_generate_fish_script.snap @@ -39,6 +39,8 @@ complete -c pv -n "__fish_pv_needs_command" -f -a "daemon:enable" -d 'Enable the complete -c pv -n "__fish_pv_needs_command" -f -a "daemon:disable" -d 'Disable the PV login daemon' complete -c pv -n "__fish_pv_needs_command" -f -a "daemon:restart" -d 'Restart the PV login daemon' complete -c pv -n "__fish_pv_needs_command" -f -a "daemon:run" -d 'Run the internal PV daemon' +complete -c pv -n "__fish_pv_needs_command" -f -a "monitor:run" -d 'Run an internal PV runtime monitor' +complete -c pv -n "__fish_pv_needs_command" -f -a "monitor:gate" -d 'Start a runtime for an internal PV runtime monitor' complete -c pv -n "__fish_pv_needs_command" -f -a "shim:php" -d 'Run the internal PV PHP shim' complete -c pv -n "__fish_pv_needs_command" -f -a "shim:composer" -d 'Run the internal PV Composer shim' complete -c pv -n "__fish_pv_needs_command" -f -a "dns:status" -d 'Show PV .test resolver status' @@ -130,6 +132,10 @@ complete -c pv -n "__fish_pv_using_subcommand daemon:restart" -l no-color -d 'Di complete -c pv -n "__fish_pv_using_subcommand daemon:restart" -s h -l help -d 'Print help' complete -c pv -n "__fish_pv_using_subcommand daemon:run" -l no-color -d 'Disable colored output' complete -c pv -n "__fish_pv_using_subcommand daemon:run" -s h -l help -d 'Print help' +complete -c pv -n "__fish_pv_using_subcommand monitor:run" -l no-color -d 'Disable colored output' +complete -c pv -n "__fish_pv_using_subcommand monitor:run" -s h -l help -d 'Print help' +complete -c pv -n "__fish_pv_using_subcommand monitor:gate" -l no-color -d 'Disable colored output' +complete -c pv -n "__fish_pv_using_subcommand monitor:gate" -s h -l help -d 'Print help' complete -c pv -n "__fish_pv_using_subcommand shim:php" -l no-color -d 'Disable colored output' complete -c pv -n "__fish_pv_using_subcommand shim:composer" -l no-color -d 'Disable colored output' complete -c pv -n "__fish_pv_using_subcommand dns:status" -l no-color -d 'Disable colored output' diff --git a/it/snapshots/cli__completions_generate_zsh_script.snap b/it/snapshots/cli__completions_generate_zsh_script.snap index 9b45fc5f..c9520da3 100644 --- a/it/snapshots/cli__completions_generate_zsh_script.snap +++ b/it/snapshots/cli__completions_generate_zsh_script.snap @@ -96,6 +96,20 @@ _arguments "${_arguments_options[@]}" : \ '--help[Print help]' \ && ret=0 ;; +(monitor:run) +_arguments "${_arguments_options[@]}" : \ +'--no-color[Disable colored output]' \ +'-h[Print help]' \ +'--help[Print help]' \ +&& ret=0 +;; +(monitor:gate) +_arguments "${_arguments_options[@]}" : \ +'--no-color[Disable colored output]' \ +'-h[Print help]' \ +'--help[Print help]' \ +&& ret=0 +;; (shim:php) _arguments "${_arguments_options[@]}" : \ '--no-color[Disable colored output]' \ @@ -654,6 +668,8 @@ _pv_commands() { 'daemon:disable:Disable the PV login daemon' \ 'daemon:restart:Restart the PV login daemon' \ 'daemon:run:Run the internal PV daemon' \ +'monitor:run:Run an internal PV runtime monitor' \ +'monitor:gate:Start a runtime for an internal PV runtime monitor' \ 'shim:php:Run the internal PV PHP shim' \ 'shim:composer:Run the internal PV Composer shim' \ 'dns:status:Show PV .test resolver status' \ @@ -884,6 +900,16 @@ _pv__subcmd__mailpit:update_commands() { local commands; commands=() _describe -t commands 'pv mailpit:update commands' commands "$@" } +(( $+functions[_pv__subcmd__monitor:gate_commands] )) || +_pv__subcmd__monitor:gate_commands() { + local commands; commands=() + _describe -t commands 'pv monitor:gate commands' commands "$@" +} +(( $+functions[_pv__subcmd__monitor:run_commands] )) || +_pv__subcmd__monitor:run_commands() { + local commands; commands=() + _describe -t commands 'pv monitor:run commands' commands "$@" +} (( $+functions[_pv__subcmd__mysql:install_commands] )) || _pv__subcmd__mysql:install_commands() { local commands; commands=() From 049fcffc28e414543addef7ce2f733b40785423e Mon Sep 17 00:00:00 2001 From: Clovis Muneza Date: Fri, 9 Oct 2026 22:47:51 -0400 Subject: [PATCH 4/6] fix(daemon): address monitor review findings A monitor now removes its own records when it accepts a release, while it still holds the reservation, so a controller that disconnects after releasing no longer leaves records that block the next start. release_monitor only waits for the reservation. stop_monitor no longer times out on the caller's grace period. The monitor's stop is bounded by the stop it actually runs, and each state request has its own timeout. Controllers reject replies from another protocol version, as startup already did. The monitor backs off after an accept error instead of spinning, as the daemon's servers do. Test pause points give up after 150 s, so a test that dies at one leaves no paused monitor behind. --- crates/daemon/src/monitor/macos.rs | 58 ++++++++++++++++++------------ crates/daemon/tests/monitor.rs | 32 +++++++++++------ 2 files changed, 57 insertions(+), 33 deletions(-) diff --git a/crates/daemon/src/monitor/macos.rs b/crates/daemon/src/monitor/macos.rs index d9e67947..582aa758 100644 --- a/crates/daemon/src/monitor/macos.rs +++ b/crates/daemon/src/monitor/macos.rs @@ -41,10 +41,12 @@ const STARTUP_REPLY_TIMEOUT: Duration = Duration::from_secs(30); const EXEC_TIMEOUT: Duration = Duration::from_secs(10); const REQUEST_TIMEOUT: Duration = Duration::from_secs(5); const RELEASE_TIMEOUT: Duration = Duration::from_secs(5); -const STOP_MARGIN: Duration = Duration::from_secs(5); +const ACCEPT_ERROR_BACKOFF: Duration = Duration::from_millis(50); const GATE_FAILURE_TIMEOUT: Duration = Duration::from_millis(100); const EXITING_LEADER_TIMEOUT: Duration = Duration::from_secs(1); const POLL_INTERVAL: Duration = Duration::from_millis(10); +/// 150 s of [`POLL_INTERVAL`]s. +const PAUSE_POLLS: u32 = 15_000; const ROTATION_INTERVAL: Duration = Duration::from_secs(60); const ROTATION_BYTES: u64 = 10 * 1024 * 1024; @@ -200,36 +202,30 @@ pub async fn monitor_state(paths: &PvPaths, subject: &str) -> Result Result { request(paths, subject, Operation::Stop(stop)).await?; - let deadline = Instant::now().checked_add(stop.grace().saturating_add(STOP_MARGIN)); loop { let state = monitor_state(paths, subject).await?; if state.cleanup != MonitorCleanup::Pending { return Ok(state); } - if deadline.is_some_and(|deadline| Instant::now() >= deadline) { - return Err(unavailable(subject, "the runtime's cleanup did not finish")); - } sleep(POLL_INTERVAL).await; } } -/// Releases the monitor of `subject` once its runtime has exited and its cleanup is complete, -/// then removes the monitor's records once its reservation proves it has exited. +/// Releases the monitor of `subject` once its runtime has exited and its cleanup is complete. +/// The monitor removes its own records before it exits; its reservation proves it has exited. pub async fn release_monitor(paths: &PvPaths, subject: &str) -> Result<(), DaemonError> { - let (record, _accepted) = request(paths, subject, Operation::Release).await?; - let reservation = wait_for_reservation(paths, subject).await?; - // Only this instance's records: a replacement that failed may have left its own. - if read_record(paths, subject)?.is_some_and(|current| current.instance == record.instance) { - remove_monitor_dir(paths, subject)?; - } - drop(reservation); + request(paths, subject, Operation::Release).await?; + drop(wait_for_reservation(paths, subject).await?); Ok(()) } @@ -365,8 +361,11 @@ async fn request( .await .map_err(|_elapsed| unavailable(subject, "the monitor did not reply"))?? .ok_or_else(|| unavailable(subject, "the monitor closed the connection"))?; - if reply.instance != record.instance { - return Err(unavailable(subject, "the reply came from another instance")); + if reply.version != PROTOCOL_VERSION || reply.instance != record.instance { + return Err(unavailable( + subject, + "the reply came from another protocol version or instance", + )); } if let ReplyResult::Error { kind, message } = reply.result { return Err(DaemonError::MonitorRejected { @@ -790,12 +789,14 @@ impl Monitor { tokio::select! { events = next_optional_events(&mut self.watch) => self.observe(events?, hooks), - accepted = self.listener.accept() => { - if let Ok((stream, _address)) = accepted { + accepted = self.listener.accept() => match accepted { + Ok((stream, _address)) => { let instance = self.record.instance.clone(); tokio::spawn(serve_connection(stream, instance, commands.clone())); } - } + // A persistent failure such as EMFILE stays ready; don't spin on it. + Err(_error) => sleep(ACCEPT_ERROR_BACKOFF).await, + }, Some(command) = requests.recv() => match command { Command::State(reply) => { let _send_result = reply.send(self.state()); @@ -807,7 +808,10 @@ impl Monitor { Command::Release(reply) => { let _send_result = reply.send(self.releasable()); } - Command::Exit => return Ok(()), + // The reservation is still held, so no replacement can own these records. + Command::Exit => { + return remove_monitor_dir(&self.paths, &self.record.subject); + } }, stopped = join_optional(&mut self.executor) => { self.executor = None; @@ -1057,19 +1061,29 @@ fn watch_lifeline(path: Utf8PathBuf) -> Result, DaemonErro Ok(receiver) } +/// Holds at `point` until the test continues it, but for 150 s at most, so a test that dies while +/// it holds leaves nothing paused: the monitor then carries on and sees the closed lifeline. That +/// outlasts the CI profile's 120 s limit on a test, so a stalled test can't pass because of it. async fn pause(hooks: &MonitorHooks, point: MonitorPause) { if let Some((reached, resume)) = pause_files(hooks, point) { announce(&reached); - while !fs::path_exists(&resume) { + for _poll in 0..PAUSE_POLLS { + if fs::path_exists(&resume) { + return; + } sleep(POLL_INTERVAL).await; } } } +/// Holds like [`pause`], blocking the thread. fn pause_blocking(hooks: &MonitorHooks, point: MonitorPause) { if let Some((reached, resume)) = pause_files(hooks, point) { announce(&reached); - while !fs::path_exists(&resume) { + for _poll in 0..PAUSE_POLLS { + if fs::path_exists(&resume) { + return; + } std::thread::sleep(POLL_INTERVAL); } } diff --git a/crates/daemon/tests/monitor.rs b/crates/daemon/tests/monitor.rs index b1238f4f..26a1af60 100644 --- a/crates/daemon/tests/monitor.rs +++ b/crates/daemon/tests/monitor.rs @@ -140,12 +140,20 @@ async fn monitor_retains_an_exit_nobody_was_connected_for() -> Result<()> { let exited = harness .wait_for_state(|state| state.cleanup != MonitorCleanup::Pending) .await?; - daemon::release_monitor(&harness.paths, SUBJECT).await?; + // A bare release, as from a controller that disconnects right after: the monitor removes + // its own records. + let instance = harness.instance()?; + harness + .raw_request(&format!( + r#"{{"version":1,"instance":"{instance}","op":"release"}}"# + )) + .await?; + wait_until(|| harness.reservation_is_free()).await?; assert_eq!(exited.exit, Some(MonitorExit::Code(7))); assert_eq!(exited.stop, None); assert_eq!(exited.cleanup, MonitorCleanup::Complete); - assert!(harness.reservation_is_free()?); + assert!(!state::fs::path_exists(&harness.monitor_dir())); Ok(()) } @@ -166,20 +174,22 @@ async fn monitor_stop_outlives_its_client_and_keeps_its_first_deadline() -> Resu harness.wait_for_log("ready").await?; let instance = harness.instance()?; - // Each raw request disconnects as soon as it is accepted. + // The raw request disconnects as soon as it is accepted. harness .raw_request(&format!( r#"{{"version":1,"instance":"{instance}","op":"stop","signal":"terminate","grace_ms":200}}"# )) .await?; - harness - .raw_request(&format!( - r#"{{"version":1,"instance":"{instance}","op":"stop","signal":"interrupt","grace_ms":60000}}"# - )) - .await?; - let stopped = harness - .wait_for_state(|state| state.cleanup != MonitorCleanup::Pending) - .await?; + // A repeated stop waits for the stop the monitor runs, under the first request's deadline. + let stopped = daemon::stop_monitor( + &harness.paths, + SUBJECT, + MonitorStop { + signal: StopSignal::Interrupt, + grace_ms: 60_000, + }, + ) + .await?; daemon::release_monitor(&harness.paths, SUBJECT).await?; assert_eq!( From 388e7a97821f9ec94b786a110648401ca42ee0a0 Mon Sep 17 00:00:00 2001 From: Clovis Muneza Date: Fri, 9 Oct 2026 23:21:28 -0400 Subject: [PATCH 5/6] fix(daemon): refuse monitor records from another protocol version Starting, requesting and recovering a monitor all read its record through one loader, which now refuses a record written under another protocol version instead of trusting its identities and stop policy. --- crates/daemon/src/monitor/macos.rs | 14 ++++++++++++-- crates/daemon/tests/monitor.rs | 20 ++++++++++++++++---- 2 files changed, 28 insertions(+), 6 deletions(-) diff --git a/crates/daemon/src/monitor/macos.rs b/crates/daemon/src/monitor/macos.rs index 582aa758..4733551b 100644 --- a/crates/daemon/src/monitor/macos.rs +++ b/crates/daemon/src/monitor/macos.rs @@ -1145,9 +1145,19 @@ fn record_path(paths: &PvPaths, subject: &str) -> Utf8PathBuf { paths.monitor_dir(subject).join(RECORD_FILE) } +/// The monitor record of `subject`, refusing one written under another protocol version: its +/// identities and stop policy may not mean what this version reads them as. fn read_record(paths: &PvPaths, subject: &str) -> Result, DaemonError> { - match fs::read_to_string(&record_path(paths, subject)) { - Ok(content) => Ok(Some(serde_json::from_str(&content)?)), + let path = record_path(paths, subject); + match fs::read_to_string(&path) { + Ok(content) => { + let record: MonitorRecord = serde_json::from_str(&content)?; + if record.version != PROTOCOL_VERSION { + return Err(DaemonError::InvalidRuntimeRecord { path }); + } + + Ok(Some(record)) + } Err(StateError::Filesystem { source, .. }) if source.kind() == io::ErrorKind::NotFound => { Ok(None) } diff --git a/crates/daemon/tests/monitor.rs b/crates/daemon/tests/monitor.rs index 26a1af60..39a06c18 100644 --- a/crates/daemon/tests/monitor.rs +++ b/crates/daemon/tests/monitor.rs @@ -348,7 +348,7 @@ async fn recovery_refuses_a_live_monitor_and_stops_a_killed_monitors_runtime() - } #[tokio::test] -async fn recovery_never_signals_a_changed_or_previous_boot_runtime() -> Result<()> { +async fn recovery_never_signals_a_changed_foreign_or_previous_boot_runtime() -> Result<()> { let harness = Harness::new()?; let mut bystander = ChildGuard( TestCommand::new("/bin/sleep") @@ -367,17 +367,27 @@ async fn recovery_never_signals_a_changed_or_previous_boot_runtime() -> Result<( // The exact runtime identity, but recorded during another boot. harness.write_record( + 1, pid, &serde_json::json!(birth), "00000000-0000-0000-0000-000000000001", )?; let previous_boot_recovery = daemon::recover_monitor(&harness.paths, SUBJECT).await; let previous_boot_records = state::fs::path_exists(&harness.monitor_dir()); - harness.write_record(pid, &changed_birth, ¤t_boot)?; + // The exact runtime identity from this boot, but written by another protocol version. + harness.write_record(2, pid, &serde_json::json!(birth), ¤t_boot)?; + let foreign_recovery = daemon::recover_monitor(&harness.paths, SUBJECT).await; + let foreign_records = state::fs::path_exists(&harness.monitor_dir().join("monitor.json")); + harness.write_record(1, pid, &changed_birth, ¤t_boot)?; let changed_recovery = daemon::recover_monitor(&harness.paths, SUBJECT).await; assert!(previous_boot_recovery.is_ok()); assert!(!previous_boot_records); + assert!(matches!( + foreign_recovery, + Err(DaemonError::InvalidRuntimeRecord { .. }) + )); + assert!(foreign_records); assert!(matches!( changed_recovery, Err(DaemonError::RuntimeProcessIdentityChanged { pid: refused }) if refused == pid @@ -608,15 +618,17 @@ impl Harness { .ok_or_else(|| anyhow!("record has no instance")) } - /// Writes a record as a dead monitor would have left it, naming `pid` as its runtime. + /// Writes a record as a dead monitor of protocol `version` would have left it, naming `pid` + /// as its runtime. fn write_record( &self, + version: u32, pid: u32, start_identity: &serde_json::Value, boot_session_id: &str, ) -> Result<()> { let record = serde_json::json!({ - "version": 1, + "version": version, "subject": SUBJECT, "instance": "0123456789abcdef0123456789abcdef", "boot_session_id": boot_session_id, From f9a9f593c2ecf08a1fefb254bbafb343617c5299 Mon Sep 17 00:00:00 2001 From: Clovis Muneza Date: Sat, 10 Oct 2026 13:36:14 -0400 Subject: [PATCH 6/6] fix(daemon): let a group with exiting members finish without unproven cleanup On macOS, killpg answers EPERM when no member of a group can take a signal: each is an unreaped zombie or already exiting. A killed runtime's children are briefly in that state, often for far longer when they hold a lot of memory. signal_process_group accepted EPERM only once the group had already exited, so a monitor's fallback stop that started while a child was still tearing down failed, and the monitor marked its cleanup unproven, which blocks the runtime until a reboot. EPERM now leaves the decision to the group-exit wait that every caller already runs. A pv-fake descendant can now hold resident memory, which makes the test reproduce this every time. --- crates/daemon/src/supervisor.rs | 9 ++++--- crates/daemon/tests/monitor.rs | 40 ++++++++++++++++++++++++++++++- crates/pv-fake/src/events.rs | 2 ++ crates/pv-fake/src/fake.rs | 10 ++++++++ crates/pv-fake/src/lib.rs | 3 +++ crates/pv-fake/tests/contracts.rs | 1 + 6 files changed, 59 insertions(+), 6 deletions(-) diff --git a/crates/daemon/src/supervisor.rs b/crates/daemon/src/supervisor.rs index f1e91ef4..6ee9b535 100644 --- a/crates/daemon/src/supervisor.rs +++ b/crates/daemon/src/supervisor.rs @@ -1115,11 +1115,10 @@ fn signal_process_group(pid: u32, signal: ProcessSignal) -> Result<(), DaemonErr Ok(()) => Ok(()), Err(source) => { let error = io::Error::from(source); - // macOS answers EPERM, not ESRCH, for a group whose only members are unreaped zombies. - if process_not_found(&error) - || (error.kind() == io::ErrorKind::PermissionDenied - && process_group_has_exited(pid)?) - { + // macOS answers EPERM, not ESRCH, when no member can take a signal: each is an + // unreaped zombie or already exiting, as a killed runtime's children briefly are. + // Every caller then waits for the group to end, which decides the stop. + if process_not_found(&error) || error.kind() == io::ErrorKind::PermissionDenied { return Ok(()); } diff --git a/crates/daemon/tests/monitor.rs b/crates/daemon/tests/monitor.rs index 39a06c18..66510557 100644 --- a/crates/daemon/tests/monitor.rs +++ b/crates/daemon/tests/monitor.rs @@ -20,7 +20,7 @@ use daemon::{ use insta::{Settings, assert_snapshot}; use pv_fake::{EventKind, FakeSettings, InstalledFake, Persona, Scenario}; use rustix::io::{FdFlags, fcntl_setfd}; -use rustix::process::{Pid, Signal, kill_process, test_kill_process}; +use rustix::process::{Pid, Signal, kill_process, kill_process_group, test_kill_process}; use state::{MonitorReservation, PvPaths, StateError}; use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader}; use tokio::net::UnixStream; @@ -239,6 +239,44 @@ async fn monitor_reports_a_leader_exit_while_its_group_lives_on() -> Result<()> Ok(()) } +#[tokio::test] +async fn monitor_cleans_up_a_killed_group_whose_member_is_still_exiting() -> Result<()> { + let harness = Harness::new()?; + // Tearing down the descendant's resident memory outlasts the small leader's exit, so the + // fallback stop starts while a member can no longer take a signal yet is no zombie. + let fake = pv_fake::install_with_settings( + &harness.paths.root().join("runtime/bin/runtime"), + Persona::LongRunning, + FakeSettings { + descendant: true, + descendant_resident_mib: 256, + ..FakeSettings::default() + }, + )?; + let started = harness + .start(harness.start_for(fake.executable(), &[])?) + .await?; + wait_until(|| { + Ok(fake + .events()? + .iter() + .any(|event| matches!(event.kind, EventKind::DescendantResident))) + }) + .await?; + + kill_process_group(pid_of(started.runtime_pid)?, Signal::KILL)?; + let cleaned = harness + .wait_for_state(|state| state.cleanup != MonitorCleanup::Pending) + .await?; + let release = daemon::release_monitor(&harness.paths, SUBJECT).await; + + assert_eq!(cleaned.exit, Some(MonitorExit::Signal(9))); + assert_eq!(cleaned.cleanup, MonitorCleanup::Complete); + release?; + + Ok(()) +} + #[tokio::test] async fn monitor_keeps_its_records_when_group_cleanup_is_unproven() -> Result<()> { let harness = Harness::with_hooks(MonitorHooks { diff --git a/crates/pv-fake/src/events.rs b/crates/pv-fake/src/events.rs index f7bc8772..f8fccad0 100644 --- a/crates/pv-fake/src/events.rs +++ b/crates/pv-fake/src/events.rs @@ -44,6 +44,8 @@ pub enum EventKind { DescendantSpawned { descendant_pid: i32, }, + /// Recorded by a descendant once its `descendant_resident_mib` of memory is resident. + DescendantResident, /// Recorded by a descendant once its parent fake closes their pipe, on exit or by dying, just /// before the descendant exits too. ParentExited, diff --git a/crates/pv-fake/src/fake.rs b/crates/pv-fake/src/fake.rs index 1c0e63f0..a7a3393a 100644 --- a/crates/pv-fake/src/fake.rs +++ b/crates/pv-fake/src/fake.rs @@ -222,11 +222,21 @@ fn spawn_descendant(argv: &[String], events: &EventLog) -> Result { /// is gone, or it is signaled. async fn run_descendant(scenario: Scenario, parent_fd: i32, events: EventLog) -> Result { let mut signals = Signals::new()?; + let resident_mib = scenario.settings.descendant_resident_mib; + let mut resident = vec![0_u8; usize::try_from(resident_mib)? * 1024 * 1024]; + // Writing one byte per page makes every page resident. + for byte in resident.iter_mut().step_by(4096) { + *byte = 1; + } + if resident_mib > 0 { + events.record(EventKind::DescendantResident)?; + } if let Some(fd) = scenario.lifeline_fd { lifeline::arm(fd, events.clone(), EventKind::LifelineFired); } lifeline::arm(parent_fd, events, EventKind::ParentExited); signals.next().await; + std::hint::black_box(&resident); Ok(ExitCode::SUCCESS) } diff --git a/crates/pv-fake/src/lib.rs b/crates/pv-fake/src/lib.rs index a60e62ab..2c6569ca 100644 --- a/crates/pv-fake/src/lib.rs +++ b/crates/pv-fake/src/lib.rs @@ -109,6 +109,9 @@ pub struct FakeSettings { /// Starts one child process in the fake's process group, as runtimes start workers. It exits /// when the fake does. pub descendant: bool, + /// Makes the descendant hold this much resident memory, so its exit takes long enough to + /// overlap a stop that starts once the fake itself has exited. + pub descendant_resident_mib: u32, /// Makes a `pv_fake_mailpit` persona exit 0 once it has answered its first HTTP request, for /// tests of runtimes that exit after becoming ready. pub exit_after_first_http_response: bool, diff --git a/crates/pv-fake/tests/contracts.rs b/crates/pv-fake/tests/contracts.rs index 59fa5024..8790dde3 100644 --- a/crates/pv-fake/tests/contracts.rs +++ b/crates/pv-fake/tests/contracts.rs @@ -1690,6 +1690,7 @@ fn event_names(fake: &InstalledFake) -> Result> { EventKind::LifelineFired => "lifeline_fired".to_owned(), EventKind::Held { .. } => "held".to_owned(), EventKind::DescendantSpawned { .. } => "descendant_spawned".to_owned(), + EventKind::DescendantResident => "descendant_resident".to_owned(), EventKind::ParentExited => "parent_exited".to_owned(), EventKind::Exit { code } => format!("exit {code}"), })