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/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/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/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..4733551b --- /dev/null +++ b/crates/daemon/src/monitor/macos.rs @@ -0,0 +1,1209 @@ +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 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; + +/// 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. +/// +/// The monitor bounds the wait: its stop, whichever request started it, ends within that stop's +/// grace period plus the SIGKILL wait, and each state request times out on its own. +pub async fn stop_monitor( + paths: &PvPaths, + subject: &str, + stop: MonitorStop, +) -> Result { + request(paths, subject, Operation::Stop(stop)).await?; + + loop { + let state = monitor_state(paths, subject).await?; + if state.cleanup != MonitorCleanup::Pending { + return Ok(state); + } + sleep(POLL_INTERVAL).await; + } +} + +/// 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> { + request(paths, subject, Operation::Release).await?; + drop(wait_for_reservation(paths, subject).await?); + + 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.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 { + 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() => 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()); + } + Command::Stop(stop) => { + self.stop.get_or_insert(stop); + self.start_cleanup(stop); + } + Command::Release(reply) => { + let _send_result = reply.send(self.releasable()); + } + // 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; + 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) +} + +/// 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); + 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); + for _poll in 0..PAUSE_POLLS { + if fs::path_exists(&resume) { + return; + } + 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) +} + +/// 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> { + 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) + } + 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..6ee9b535 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,10 @@ 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, 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(()); } @@ -1219,17 +1223,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 +1277,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 +1516,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 +1546,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..66510557 --- /dev/null +++ b/crates/daemon/tests/monitor.rs @@ -0,0 +1,829 @@ +//! 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, kill_process_group, 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?; + // 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!(!state::fs::path_exists(&harness.monitor_dir())); + + 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()?; + + // 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?; + // 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!( + 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_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 { + 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_foreign_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( + 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()); + // 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 + )); + 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 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": version, + "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/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/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/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..2c6569ca 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)] @@ -106,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}"), }) 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) 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=()