diff --git a/crates/cli/src/bootstrap/mod.rs b/crates/cli/src/bootstrap/mod.rs index 6190b7ff4..f8dd94c7d 100644 --- a/crates/cli/src/bootstrap/mod.rs +++ b/crates/cli/src/bootstrap/mod.rs @@ -90,7 +90,8 @@ impl GatewaySpec { log::error!( target: "nemo_relay.bootstrap", event = "gateway_acquisition_failed", - bind = self.bind.to_string().as_str(); + bind = self.bind.to_string().as_str(), + failure_kind = bootstrap_failure_kind(&error); "Gateway acquisition failed" ); Err(error) @@ -120,7 +121,8 @@ impl GatewaySpec { log::error!( target: "nemo_relay.bootstrap", event = "gateway_recovery_failed", - instance_id = expected_instance; + instance_id = expected_instance, + failure_kind = bootstrap_failure_kind(&error); "Gateway recovery failed" ); Err(error) @@ -200,8 +202,17 @@ fn acquire_gateway(spec: &GatewaySpec) -> Result { Err(incompatible_relay_error(&url)) } } - (RelayHealth::Foreign, _) => Err(foreign_listener_error(&url)), - (RelayHealth::Unavailable, _) => start_gateway(spec, &state), + (RelayHealth::Foreign, _) => { + if state::stop_unhealthy_owned_gateway_locked(&state, &url)? { + start_gateway(spec, &state) + } else { + Err(foreign_listener_error(&url)) + } + } + (RelayHealth::Unavailable, _) => { + state::stop_unhealthy_owned_gateway_locked(&state, &url)?; + start_gateway(spec, &state) + } } } @@ -220,8 +231,14 @@ fn recover_gateway(spec: &GatewaySpec, expected_instance: &str) -> Result return Err(incompatible_relay_error(&requested_url)), - (RelayHealth::Foreign, _) => return Err(foreign_listener_error(&requested_url)), - (RelayHealth::Unavailable, _) => {} + (RelayHealth::Foreign, _) => { + if !state::stop_unhealthy_owned_gateway_locked(&state, &requested_url)? { + return Err(foreign_listener_error(&requested_url)); + } + } + (RelayHealth::Unavailable, _) => { + state::stop_unhealthy_owned_gateway_locked(&state, &requested_url)?; + } } } @@ -285,17 +302,47 @@ fn compatible_endpoint( } fn foreign_listener_error(url: &str) -> String { + log::error!( + target: "nemo_relay.bootstrap", + event = "gateway_port_conflict", + endpoint = url, + observed_health = "unverified_listener", + remediation = "stop_listener_or_configure_another_port"; + "Gateway endpoint is occupied by an unverified listener" + ); format!( "{url} is occupied by a service that is not a compatible NeMo Relay gateway; stop that service or configure another port" ) } fn incompatible_relay_error(url: &str) -> String { + log::error!( + target: "nemo_relay.bootstrap", + event = "gateway_port_conflict", + endpoint = url, + observed_health = "incompatible_relay", + remediation = "stop_gateway_wait_for_idle_shutdown_or_reinstall_with_force"; + "Gateway endpoint is occupied by an incompatible NeMo Relay gateway" + ); format!( "{url} is occupied by NeMo Relay with a different version or persistent configuration; stop it, wait for idle shutdown, or reinstall the integration with --force" ) } +fn bootstrap_failure_kind(error: &str) -> &'static str { + if error.contains("not a compatible NeMo Relay gateway") { + "foreign_listener" + } else if error.contains("different version or persistent configuration") { + "incompatible_gateway" + } else if error.contains("became unhealthy") { + "unhealthy_gateway" + } else if error.contains("did not become ready") { + "gateway_readiness_timeout" + } else { + "bootstrap_failure" + } +} + fn start_gateway(spec: &GatewaySpec, state: &Path) -> Result { log::info!( target: "nemo_relay.bootstrap", diff --git a/crates/cli/src/bootstrap/state.rs b/crates/cli/src/bootstrap/state.rs index 3eacede47..865b32ca9 100644 --- a/crates/cli/src/bootstrap/state.rs +++ b/crates/cli/src/bootstrap/state.rs @@ -21,6 +21,7 @@ use crate::gateway::client::{RelayHealth, probe, request_shutdown}; pub(crate) const BOOTSTRAP_STATE_DIR_ENV: &str = "NEMO_RELAY_BOOTSTRAP_STATE_DIR"; pub(crate) const BOOTSTRAP_SHUTDOWN_TOKEN_ENV: &str = "NEMO_RELAY_BOOTSTRAP_SHUTDOWN_TOKEN"; const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5); +const UNHEALTHY_GATEWAY_TERMINATION_TIMEOUT: Duration = Duration::from_secs(1); #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)] pub(super) struct OwnerRecord { @@ -28,6 +29,8 @@ pub(super) struct OwnerRecord { version: String, bootstrap_protocol: u64, pid: u32, + #[serde(default)] + process_identity: Option, url: String, shutdown_token: String, bootstrap_fingerprint: Option, @@ -47,6 +50,7 @@ impl OwnerRecord { version: env!("CARGO_PKG_VERSION").into(), bootstrap_protocol: BOOTSTRAP_PROTOCOL_VERSION, pid, + process_identity: process_identity(pid), url: url.into(), shutdown_token: shutdown_token.into(), bootstrap_fingerprint: fingerprint.map(str::to_owned), @@ -248,6 +252,32 @@ pub(crate) fn stop_version_mismatched_owned_gateway_locked( stop_owned_gateway_locked(&path, &owner, url) } +/// Terminates a managed gateway whose health endpoint is unavailable so its +/// listener cannot block a replacement process from binding the endpoint. +/// +/// The caller holds the endpoint startup lock. A valid ownership record is the +/// authority for signalling the process; listeners without one remain untouched. +pub(crate) fn stop_unhealthy_owned_gateway_locked(state: &Path, url: &str) -> Result { + let path = owner_path(state, url); + let Some(owner) = read_owner_record(&path)? else { + return Ok(false); + }; + if !owner.valid_for(url) { + return Ok(false); + } + if probe(url, owner.bootstrap_fingerprint.as_deref()) == RelayHealth::Compatible { + return Ok(false); + } + if !owner_process_identity_matches(&owner) { + remove_if_matches(&path, &owner)?; + return Ok(false); + } + + let was_running = terminate_owned_gateway_process(owner.pid)?; + remove_if_matches(&path, &owner)?; + Ok(was_running) +} + fn stop_owned_and_reset_locked(state: &Path, url: &str) -> Result { let path = owner_path(state, url); let Some(owner) = read_owner_record(&path)? else { @@ -304,6 +334,247 @@ fn stop_owned_gateway_locked(path: &Path, owner: &OwnerRecord, url: &str) -> Res Ok(true) } +#[cfg(unix)] +fn terminate_owned_gateway_process(pid: u32) -> Result { + let Ok(pid) = i32::try_from(pid) else { + return Ok(false); + }; + if !process_is_running(pid) { + return Ok(false); + } + + log_unhealthy_gateway_termination_started(pid); + let target = signal_gateway_process_group(pid, libc::SIGTERM)?; + if wait_for_termination(target, UNHEALTHY_GATEWAY_TERMINATION_TIMEOUT) { + return Ok(true); + } + + log_unhealthy_gateway_termination_escalated(pid); + signal_gateway_termination_target(target, libc::SIGKILL)?; + if wait_for_termination(target, UNHEALTHY_GATEWAY_TERMINATION_TIMEOUT) { + Ok(true) + } else { + Err(format!( + "managed Relay gateway process {pid} did not terminate" + )) + } +} + +#[cfg(target_os = "linux")] +fn process_identity(pid: u32) -> Option { + let stat = fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; + // `comm` is parenthesized and may contain spaces, so process fields only + // after the final `)` character. + stat.rsplit_once(')')? + .1 + .split_whitespace() + // Field 22 is process start time in clock ticks. The first field here + // is field 3 (`state`). + .nth(19)? + .parse() + .ok() +} + +#[cfg(target_os = "macos")] +fn process_identity(pid: u32) -> Option { + let pid = i32::try_from(pid).ok()?; + // SAFETY: `info` is initialized and passed with its exact size for the + // documented PROC_PIDTBSDINFO query. + let mut info: libc::proc_bsdinfo = unsafe { std::mem::zeroed() }; + let result = unsafe { + libc::proc_pidinfo( + pid, + libc::PROC_PIDTBSDINFO, + 0, + (&raw mut info).cast(), + i32::try_from(std::mem::size_of::()).ok()?, + ) + }; + (result == std::mem::size_of::() as i32) + .then(|| { + info.pbi_start_tvsec + .checked_mul(1_000_000)? + .checked_add(info.pbi_start_tvusec) + }) + .flatten() +} + +#[cfg(windows)] +fn process_identity(pid: u32) -> Option { + use windows_sys::Win32::Foundation::{CloseHandle, FILETIME}; + use windows_sys::Win32::System::Threading::{ + GetProcessTimes, OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION, + }; + + // A handle keeps the queried process instance stable while its creation + // identity is read, even if this PID is recycled concurrently. + let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) }; + if handle.is_null() { + return None; + } + let mut creation: FILETIME = unsafe { std::mem::zeroed() }; + let mut exit: FILETIME = unsafe { std::mem::zeroed() }; + let mut kernel: FILETIME = unsafe { std::mem::zeroed() }; + let mut user: FILETIME = unsafe { std::mem::zeroed() }; + let result = + unsafe { GetProcessTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user) }; + unsafe { CloseHandle(handle) }; + (result != 0) + .then(|| (u64::from(creation.dwHighDateTime) << 32) | u64::from(creation.dwLowDateTime)) +} + +#[cfg(not(any(target_os = "linux", target_os = "macos", windows)))] +fn process_identity(_pid: u32) -> Option { + None +} + +fn owner_process_identity_matches(owner: &OwnerRecord) -> bool { + owner.process_identity.is_some_and(|identity| { + process_identity(owner.pid).is_some_and(|current| current == identity) + }) +} + +#[cfg(unix)] +#[derive(Clone, Copy)] +enum GatewayTerminationTarget { + ProcessGroup(i32), + Process(i32), +} + +#[cfg(unix)] +fn signal_gateway_process_group(pid: i32, signal: i32) -> Result { + // Detached gateways call setsid, making their PID the process-group ID. + // Fall back to the direct PID for an older or otherwise non-detached sidecar. + let group_result = unsafe { libc::kill(-pid, signal) }; + if group_result == 0 { + return Ok(GatewayTerminationTarget::ProcessGroup(pid)); + } + let group_error = std::io::Error::last_os_error(); + if group_error.raw_os_error() != Some(libc::ESRCH) { + return Err(format!( + "failed to signal managed Relay gateway process {pid}: {group_error}" + )); + } + if unsafe { libc::kill(pid, signal) } == 0 { + Ok(GatewayTerminationTarget::Process(pid)) + } else { + let error = std::io::Error::last_os_error(); + (error.raw_os_error() == Some(libc::ESRCH)) + .then_some(GatewayTerminationTarget::Process(pid)) + .ok_or_else(|| format!("failed to signal managed Relay gateway process {pid}: {error}")) + } +} + +#[cfg(unix)] +fn signal_gateway_termination_target( + target: GatewayTerminationTarget, + signal: i32, +) -> Result<(), String> { + let (pid, target_pid) = match target { + GatewayTerminationTarget::ProcessGroup(pid) => (pid, -pid), + GatewayTerminationTarget::Process(pid) => (pid, pid), + }; + if unsafe { libc::kill(target_pid, signal) } == 0 { + return Ok(()); + } + let error = std::io::Error::last_os_error(); + (error.raw_os_error() == Some(libc::ESRCH)) + .then_some(()) + .ok_or_else(|| format!("failed to signal managed Relay gateway process {pid}: {error}")) +} + +#[cfg(unix)] +fn process_is_running(pid: i32) -> bool { + (unsafe { libc::kill(pid, 0) }) == 0 + || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) +} + +#[cfg(unix)] +fn wait_for_termination(target: GatewayTerminationTarget, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + while termination_target_is_running(target) && Instant::now() < deadline { + thread::sleep(Duration::from_millis(50)); + } + !termination_target_is_running(target) +} + +#[cfg(unix)] +fn termination_target_is_running(target: GatewayTerminationTarget) -> bool { + match target { + GatewayTerminationTarget::ProcessGroup(pid) => process_is_running(-pid), + GatewayTerminationTarget::Process(pid) => process_is_running(pid), + } +} + +#[cfg(windows)] +fn terminate_owned_gateway_process(pid: u32) -> Result { + if !windows_process_is_running(pid) { + return Ok(false); + } + + log_unhealthy_gateway_termination_started(pid as i32); + let _graceful = std::process::Command::new("taskkill") + .args(["/PID", &pid.to_string(), "/T"]) + .status(); + if wait_for_windows_process_exit(pid, UNHEALTHY_GATEWAY_TERMINATION_TIMEOUT) { + return Ok(true); + } + + log_unhealthy_gateway_termination_escalated(pid as i32); + let _forced = std::process::Command::new("taskkill") + .args(["/PID", &pid.to_string(), "/T", "/F"]) + .status() + .map_err(|error| { + format!("failed to force-kill managed Relay gateway process {pid}: {error}") + })?; + if wait_for_windows_process_exit(pid, UNHEALTHY_GATEWAY_TERMINATION_TIMEOUT) { + Ok(true) + } else { + Err(format!( + "managed Relay gateway process {pid} did not terminate" + )) + } +} + +#[cfg(windows)] +fn windows_process_is_running(pid: u32) -> bool { + process_identity(pid).is_some() +} + +#[cfg(windows)] +fn wait_for_windows_process_exit(pid: u32, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + while windows_process_is_running(pid) && Instant::now() < deadline { + thread::sleep(Duration::from_millis(50)); + } + !windows_process_is_running(pid) +} + +fn log_unhealthy_gateway_termination_started(pid: i32) { + log::warn!( + target: "nemo_relay.bootstrap", + event = "unhealthy_gateway_termination_started", + process_id = pid; + "Terminating unhealthy managed gateway" + ); +} + +fn log_unhealthy_gateway_termination_escalated(pid: i32) { + log::warn!( + target: "nemo_relay.bootstrap", + event = "unhealthy_gateway_termination_escalated", + process_id = pid; + "Force-killing unhealthy managed gateway after graceful termination timeout" + ); +} + +#[cfg(not(any(unix, windows)))] +fn terminate_owned_gateway_process(pid: u32) -> Result { + Err(format!( + "cannot terminate unhealthy managed Relay gateway process {pid} on this platform" + )) +} + fn write_owner_record(path: &Path, record: &OwnerRecord) -> Result<(), String> { let bytes = serde_json::to_vec(record) .map_err(|error| format!("failed to encode gateway ownership: {error}"))?; diff --git a/crates/cli/tests/coverage/shared/bootstrap_state_tests.rs b/crates/cli/tests/coverage/shared/bootstrap_state_tests.rs index 627c7d8a5..abdf9ea5b 100644 --- a/crates/cli/tests/coverage/shared/bootstrap_state_tests.rs +++ b/crates/cli/tests/coverage/shared/bootstrap_state_tests.rs @@ -23,6 +23,20 @@ fn owner_records_are_versioned_endpoint_scoped_and_round_trip() { assert_eq!(lock_name("not a url/with spaces"), "not_a_url_with_spaces"); } +#[cfg(any(target_os = "linux", target_os = "macos", windows))] +#[test] +fn live_owner_record_uses_a_process_instance_identity() { + let owner = OwnerRecord::new( + std::process::id(), + "http://127.0.0.1:47632", + "shutdown", + Some("fingerprint"), + ); + + assert!(owner.process_identity.is_some()); + assert!(owner_process_identity_matches(&owner)); +} + #[test] fn recovery_records_preserve_pending_and_ready_attempts() { let dir = tempfile::tempdir().unwrap(); @@ -234,7 +248,7 @@ fn same_version_or_invalid_owned_gateway_is_not_stopped_for_replacement() { let state = state_dir().unwrap(); create_private_dir(&state).unwrap(); let path = owner_path(&state, url); - let owner = OwnerRecord::new(42, url, "shutdown-token", Some("fingerprint")); + let owner = OwnerRecord::new(i32::MAX as u32, url, "shutdown-token", Some("fingerprint")); write_owner_record(&path, &owner).unwrap(); let _lock = lock_endpoint(&state, url).unwrap(); @@ -249,3 +263,62 @@ fn same_version_or_invalid_owned_gateway_is_not_stopped_for_replacement() { assert!(!stop_version_mismatched_owned_gateway_locked(&state, url).unwrap()); assert!(path.exists()); } + +#[test] +fn stale_unhealthy_gateway_owner_is_removed() { + let dir = tempfile::tempdir().unwrap(); + let url = "http://127.0.0.1:9"; + let path = owner_path(dir.path(), url); + let owner = OwnerRecord::new(u32::MAX, url, "shutdown-token", Some("fingerprint")); + write_owner_record(&path, &owner).unwrap(); + + assert!(!stop_unhealthy_owned_gateway_locked(dir.path(), url).unwrap()); + assert!(!path.exists()); +} + +#[cfg(unix)] +#[test] +fn unhealthy_owned_gateway_is_force_killed_after_the_grace_period() { + use std::os::unix::process::CommandExt; + + let dir = tempfile::tempdir().unwrap(); + let url = "http://127.0.0.1:9"; + let path = owner_path(dir.path(), url); + let child_pid_path = dir.path().join("child.pid"); + let mut command = std::process::Command::new("sh"); + command.args([ + "-c", + "sh -c 'trap \"\" TERM; while :; do sleep 60; done' & echo $! > \"$1\"; wait", + "sh", + child_pid_path.to_str().unwrap(), + ]); + // SAFETY: The child calls only async-signal-safe `setsid` before exec. + unsafe { + command.pre_exec(|| { + if libc::setsid() == -1 { + Err(std::io::Error::last_os_error()) + } else { + Ok(()) + } + }); + } + let mut child = command.spawn().unwrap(); + let child_pid = loop { + if let Ok(value) = std::fs::read_to_string(&child_pid_path) { + break value.trim().parse::().unwrap(); + } + std::thread::sleep(Duration::from_millis(10)); + }; + let owner = OwnerRecord::new(child.id(), url, "shutdown-token", Some("fingerprint")); + write_owner_record(&path, &owner).unwrap(); + let waiter = std::thread::spawn(move || child.wait()); + + assert!(stop_unhealthy_owned_gateway_locked(dir.path(), url).unwrap()); + assert!(!waiter.join().unwrap().unwrap().success()); + let deadline = Instant::now() + Duration::from_secs(1); + while process_is_running(child_pid) && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + assert!(!process_is_running(child_pid)); + assert!(!path.exists()); +}