From 0333b34771c188901d27816fa6be3f3fccd424fd Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Sun, 3 May 2026 22:22:47 -0700 Subject: [PATCH] fix(tauri-shell): proactively kill stale openhuman RPC on startup (#1166) --- app/src-tauri/src/core_process.rs | 358 ++++++++++++++++++++++-- app/src-tauri/src/core_process_tests.rs | 87 +++++- docs/BUILDING.md | 21 ++ 3 files changed, 443 insertions(+), 23 deletions(-) diff --git a/app/src-tauri/src/core_process.rs b/app/src-tauri/src/core_process.rs index bc7edf779..8690d82a2 100644 --- a/app/src-tauri/src/core_process.rs +++ b/app/src-tauri/src/core_process.rs @@ -2,9 +2,22 @@ //! //! The core's HTTP/JSON-RPC server runs as a tokio task inside the Tauri host //! so its lifetime is tied to the GUI process — there is no sidecar to leak -//! on Cmd+Q. If something is already listening on the configured port (e.g. -//! a manual `openhuman-core run` harness for debugging), `ensure_running` -//! attaches to it instead of spawning a duplicate listener. +//! on Cmd+Q. +//! +//! Stale-listener policy (see issue #1130): if something is already listening +//! on the configured port when `ensure_running` runs, we probe `GET /` to see +//! whether it is an OpenHuman core. If it is, we treat it as a stale process +//! left behind by a previous build/dev session and proactively terminate it +//! (graceful signal, then a force-kill that *revalidates* the pid is still +//! the same listener — guards against PID reuse if the original exits inside +//! the grace window) before spawning a fresh embedded server — otherwise the +//! new UI would silently bind to an older RPC implementation. If the listener +//! is something else (or unreachable), we refuse to attach and surface the +//! conflict so it can be diagnosed instead of producing 401s and version +//! drift downstream. +//! Set `OPENHUMAN_CORE_REUSE_EXISTING=1` to opt back into the legacy +//! attach-to-whatever-is-listening behavior (e.g. a manual `openhuman-core +//! run` harness for debugging). use std::sync::Arc; use std::sync::LazyLock; @@ -82,27 +95,37 @@ impl CoreProcessHandle { } async fn is_rpc_port_open(&self) -> bool { - matches!( - timeout( - Duration::from_millis(150), - TcpStream::connect(("127.0.0.1", self.port)), - ) - .await, - Ok(Ok(_)) - ) + is_port_open(self.port).await } pub async fn ensure_running(&self) -> Result<(), String> { if self.is_rpc_port_open().await { - log::info!( - "[core] found existing core rpc endpoint at {}", - self.rpc_url() - ); - log::warn!( - "[core] reusing port {} — another `openhuman-core` instance is already listening; this Tauri host will not spawn an embedded server. Authenticated Tauri-side calls will 401 unless the listener was started with this process's OPENHUMAN_CORE_TOKEN.", - self.port - ); - return Ok(()); + if reuse_existing_listener_enabled() { + log::warn!( + "[core] OPENHUMAN_CORE_REUSE_EXISTING=1 — attaching to whatever is listening on port {} without identification (legacy behavior)", + self.port + ); + return Ok(()); + } + + match identify_listener(self.port).await { + ListenerKind::OpenHuman => { + log::warn!( + "[core] found stale OpenHuman listener on port {} — taking over (issue #1130)", + self.port + ); + self.takeover_stale_listener().await?; + // Fall through to spawn-and-wait below. + } + ListenerKind::Unknown { reason } => { + let msg = format!( + "Core RPC port {} is in use by something that is not an OpenHuman core ({reason}). Refusing to attach (set OPENHUMAN_CORE_REUSE_EXISTING=1 to override) — quit the other process or set OPENHUMAN_CORE_PORT to a different port and relaunch.", + self.port + ); + log::error!("[core] {msg}"); + return Err(msg); + } + } } { @@ -160,6 +183,92 @@ impl CoreProcessHandle { Err("core process did not become ready".to_string()) } + /// Identify the OS pid currently bound to our port and terminate it, + /// then wait for the port to free. Used when the listener has been + /// fingerprinted as an OpenHuman core (via `GET /`) so killing it is safe. + async fn takeover_stale_listener(&self) -> Result<(), String> { + let pid = match find_pid_on_port(self.port) { + Some(pid) => pid, + None => { + return Err(format!( + "could not determine pid bound to port {} — refusing to take over", + self.port + )); + } + }; + let self_pid = std::process::id(); + if pid == self_pid { + // Defensive — `ensure_running` checks the port before spawning, + // so this branch should be unreachable. If it ever hits, killing + // ourselves would be catastrophic. + return Err(format!( + "stale-listener pid {pid} matches the Tauri host pid; refusing to self-terminate" + )); + } + log::warn!( + "[core] terminating stale OpenHuman process pid={pid} on port {} (issue #1130)", + self.port + ); + if let Err(e) = kill_pid_term(pid) { + return Err(format!("failed to signal stale openhuman pid {pid}: {e}")); + } + + // Wait for the graceful exit, then revalidate ownership before any + // force-kill — between the SIGTERM and a delayed SIGKILL the original + // pid could have exited and been reused by an unrelated process. If + // the port is now bound to a different pid (or to nothing), we do + // NOT escalate to a force-kill against the originally-resolved pid. + // (CodeRabbit feedback on #1166.) + const GRACE_MS: u64 = 750; + tokio::time::sleep(Duration::from_millis(GRACE_MS)).await; + + if is_port_open(self.port).await { + match find_pid_on_port(self.port) { + Some(current) if current == pid => { + log::warn!( + "[core] pid {pid} still bound to port {} after SIGTERM — escalating to SIGKILL", + self.port + ); + if let Err(e) = kill_pid_force(pid) { + return Err(format!( + "failed to force-kill stale openhuman pid {pid}: {e}" + )); + } + } + Some(current) => { + return Err(format!( + "port {} rebounded to pid {current} after terminating pid {pid}; refusing to kill a different process", + self.port + )); + } + None => { + // Port still showed open in `is_port_open` but pid lookup + // returned nothing — likely a transient race with the + // listener tearing down. Fall through to the poll loop. + } + } + } + + const POLL_MS: u64 = 100; + const MAX_WAIT_MS: u64 = 5_000; + let mut waited_ms: u64 = GRACE_MS; + while is_port_open(self.port).await { + if waited_ms >= MAX_WAIT_MS { + return Err(format!( + "signaled pid {pid} but port {} remained bound after {MAX_WAIT_MS}ms", + self.port + )); + } + tokio::time::sleep(Duration::from_millis(POLL_MS)).await; + waited_ms += POLL_MS; + } + log::info!( + "[core] stale listener cleared (pid={pid}, port={}) after {waited_ms}ms", + self.port + ); + Ok(()) + } + /// Restart the embedded core to pick up updated macOS permission grants. /// /// macOS caches permission state per-process; restarting forces a fresh @@ -244,6 +353,215 @@ pub fn default_core_port() -> u16 { .unwrap_or(7788) } +/// Whether `OPENHUMAN_CORE_REUSE_EXISTING` is set to a truthy value. Opts +/// back into the pre-#1130 behavior of attaching to whatever is listening +/// on the port without identification — useful for manual harnesses. +fn reuse_existing_listener_enabled() -> bool { + std::env::var("OPENHUMAN_CORE_REUSE_EXISTING") + .map(|v| matches!(v.trim(), "1" | "true" | "TRUE" | "yes" | "YES")) + .unwrap_or(false) +} + +async fn is_port_open(port: u16) -> bool { + matches!( + timeout( + Duration::from_millis(150), + TcpStream::connect(("127.0.0.1", port)), + ) + .await, + Ok(Ok(_)) + ) +} + +/// What is currently listening on the core RPC port. +#[derive(Debug)] +enum ListenerKind { + /// `GET /` returned a JSON body with `"name": "openhuman"` — i.e. a + /// stale OpenHuman core process from a previous build/session. + OpenHuman, + /// Either the listener didn't speak HTTP, didn't respond, or returned + /// a body that doesn't identify as openhuman. + Unknown { reason: String }, +} + +/// Probe `GET http://127.0.0.1:/` to fingerprint the listener. +/// Unauthenticated — the core's root handler does not require a token. +async fn identify_listener(port: u16) -> ListenerKind { + let url = format!("http://127.0.0.1:{port}/"); + let client = match reqwest::Client::builder() + .timeout(Duration::from_millis(750)) + .build() + { + Ok(c) => c, + Err(e) => { + return ListenerKind::Unknown { + reason: format!("reqwest client build failed: {e}"), + }; + } + }; + let resp = match client.get(&url).send().await { + Ok(r) => r, + Err(e) => { + return ListenerKind::Unknown { + reason: format!("probe GET / failed: {e}"), + }; + } + }; + if !resp.status().is_success() { + return ListenerKind::Unknown { + reason: format!("probe GET / returned status {}", resp.status()), + }; + } + let body = match resp.text().await { + Ok(b) => b, + Err(e) => { + return ListenerKind::Unknown { + reason: format!("probe GET / body read failed: {e}"), + }; + } + }; + if is_openhuman_root_body(&body) { + log::info!("[core] listener on port {port} identified as openhuman core"); + ListenerKind::OpenHuman + } else { + let preview: String = body.chars().take(80).collect(); + ListenerKind::Unknown { + reason: format!("probe GET / body did not identify as openhuman ({preview:?})"), + } + } +} + +/// Pure parse of the root-handler JSON. Public-by-test so the fingerprinting +/// logic stays unit-testable without standing up an HTTP server. +fn is_openhuman_root_body(body: &str) -> bool { + let value: serde_json::Value = match serde_json::from_str(body) { + Ok(v) => v, + Err(_) => return false, + }; + value + .get("name") + .and_then(|v| v.as_str()) + .map(|s| s == "openhuman") + .unwrap_or(false) +} + +#[cfg(unix)] +fn find_pid_on_port(port: u16) -> Option { + let output = std::process::Command::new("lsof") + .args(["-nP", "-iTCP", &format!("-i:{port}"), "-sTCP:LISTEN", "-t"]) + .output() + .ok()?; + if !output.status.success() { + return None; + } + parse_lsof_pid(&String::from_utf8_lossy(&output.stdout)) +} + +#[cfg(windows)] +fn find_pid_on_port(port: u16) -> Option { + let output = std::process::Command::new("netstat") + .args(["-ano", "-p", "TCP"]) + .output() + .ok()?; + if !output.status.success() { + return None; + } + parse_netstat_pid(&String::from_utf8_lossy(&output.stdout), port) +} + +/// Pure parse of `lsof -t` output (one pid per line; first wins). +fn parse_lsof_pid(stdout: &str) -> Option { + stdout + .lines() + .map(str::trim) + .find(|l| !l.is_empty()) + .and_then(|l| l.parse::().ok()) +} + +/// Pure parse of `netstat -ano` output for a LISTENING entry on `port`. +#[allow(dead_code)] // exercised only on windows builds +fn parse_netstat_pid(stdout: &str, port: u16) -> Option { + let needle = format!(":{port}"); + for line in stdout.lines() { + let trimmed = line.trim(); + if !trimmed.contains("LISTENING") { + continue; + } + let parts: Vec<&str> = trimmed.split_whitespace().collect(); + // Expected: ["TCP", "127.0.0.1:7788", "0.0.0.0:0", "LISTENING", "1234"] + if parts.len() >= 5 && parts[1].ends_with(&needle) { + if let Ok(pid) = parts[parts.len() - 1].parse::() { + return Some(pid); + } + } + } + None +} + +/// Send the graceful-shutdown signal to `pid`. Returns `Ok` if the process +/// exited cleanly, was already gone, or accepted the signal. Callers must +/// re-check ownership of the resource (e.g. that the same pid is still bound +/// to the port) before escalating to `kill_pid_force` — see the PID-reuse +/// hazard discussed on #1166. +#[cfg(unix)] +fn kill_pid_term(pid: u32) -> Result<(), String> { + use nix::sys::signal::{kill, Signal}; + use nix::unistd::Pid; + let target = Pid::from_raw(pid as i32); + if let Err(e) = kill(target, Signal::SIGTERM) { + // ESRCH means already gone — treat as success. + if e != nix::errno::Errno::ESRCH { + return Err(format!("SIGTERM pid {pid}: {e}")); + } + } + Ok(()) +} + +/// Force-kill `pid` after `kill_pid_term` failed to free the resource. Caller +/// is responsible for revalidating that `pid` still owns the resource we're +/// trying to free — see `takeover_stale_listener` for the revalidation step. +#[cfg(unix)] +fn kill_pid_force(pid: u32) -> Result<(), String> { + use nix::sys::signal::{kill, Signal}; + use nix::unistd::Pid; + let target = Pid::from_raw(pid as i32); + match kill(Pid::from_raw(pid as i32), Signal::SIGKILL) { + Ok(()) => Ok(()), + // ESRCH means the process exited between our re-validation and the + // SIGKILL — the resource is freeing on its own, treat as success. + Err(e) if e == nix::errno::Errno::ESRCH => { + let _ = target; + Ok(()) + } + Err(e) => Err(format!("SIGKILL pid {pid}: {e}")), + } +} + +/// Windows has no graceful equivalent for a windowless RPC server — `taskkill` +/// without `/F` only delivers `WM_CLOSE` to GUI apps. Send the WM_CLOSE first +/// (best-effort) so console subprocesses can run shutdown handlers; the +/// follow-up `kill_pid_force` does the actual termination. +#[cfg(windows)] +fn kill_pid_term(pid: u32) -> Result<(), String> { + // Best-effort — ignore non-zero exit (e.g. process is windowless). + let _ = std::process::Command::new("taskkill") + .args(["/PID", &pid.to_string()]) + .status(); + Ok(()) +} + +#[cfg(windows)] +fn kill_pid_force(pid: u32) -> Result<(), String> { + let status = std::process::Command::new("taskkill") + .args(["/F", "/T", "/PID", &pid.to_string()]) + .status() + .map_err(|e| format!("taskkill spawn: {e}"))?; + if !status.success() { + return Err(format!("taskkill exited with {status}")); + } + Ok(()) +} + #[cfg(test)] #[path = "core_process_tests.rs"] mod tests; diff --git a/app/src-tauri/src/core_process_tests.rs b/app/src-tauri/src/core_process_tests.rs index 6a7f6b6f1..fe09766e8 100644 --- a/app/src-tauri/src/core_process_tests.rs +++ b/app/src-tauri/src/core_process_tests.rs @@ -1,4 +1,7 @@ -use super::{current_rpc_token, default_core_port, generate_rpc_token, CoreProcessHandle}; +use super::{ + current_rpc_token, default_core_port, generate_rpc_token, is_openhuman_root_body, + parse_lsof_pid, parse_netstat_pid, CoreProcessHandle, +}; use std::sync::{Mutex, MutexGuard, OnceLock}; fn env_lock() -> MutexGuard<'static, ()> { @@ -54,8 +57,36 @@ fn core_process_handle_new_creates_instance() { assert_eq!(handle.rpc_url(), "http://127.0.0.1:9999/rpc"); } +/// Issue #1130: a non-OpenHuman listener on the RPC port must NOT be +/// silently attached to. The test binds a bare `TcpListener` (which never +/// answers HTTP) so the identification probe sees an unknown listener and +/// `ensure_running` must surface the conflict instead of returning Ok. #[test] -fn ensure_running_returns_ok_when_rpc_port_already_open() { +fn ensure_running_refuses_unknown_listener_on_port() { + let _env_lock = env_lock(); + let _unset = EnvGuard::unset("OPENHUMAN_CORE_REUSE_EXISTING"); + let rt = tokio::runtime::Runtime::new().expect("runtime"); + let result = rt.block_on(async { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind test listener"); + let port = listener.local_addr().expect("local addr").port(); + let handle = CoreProcessHandle::new(port); + handle.ensure_running().await + }); + let err = result.expect_err("ensure_running must refuse an unidentified listener"); + assert!( + err.contains("not an OpenHuman core") || err.contains("port"), + "error should explain the conflict, got: {err}" + ); +} + +/// Escape hatch: setting `OPENHUMAN_CORE_REUSE_EXISTING=1` opts back into +/// the legacy attach-to-anything behavior for manual harnesses. +#[test] +fn ensure_running_reuses_unknown_listener_when_override_set() { + let _env_lock = env_lock(); + let _override = EnvGuard::set("OPENHUMAN_CORE_REUSE_EXISTING", "1"); let rt = tokio::runtime::Runtime::new().expect("runtime"); let result = rt.block_on(async { let listener = tokio::net::TcpListener::bind("127.0.0.1:0") @@ -67,10 +98,60 @@ fn ensure_running_returns_ok_when_rpc_port_already_open() { }); assert!( result.is_ok(), - "ensure_running should fast-path: {result:?}" + "override should restore legacy fast-path: {result:?}" ); } +// --------------------------------------------------------------------------- +// Listener fingerprinting (issue #1130) +// --------------------------------------------------------------------------- + +#[test] +fn is_openhuman_root_body_matches_canonical_root_response() { + // Mirrors the JSON shape produced by `core/jsonrpc.rs::root_handler`. + let body = r#"{ + "name": "openhuman", + "ok": true, + "endpoints": {"health": "/health", "rpc": "/rpc"} + }"#; + assert!(is_openhuman_root_body(body)); +} + +#[test] +fn is_openhuman_root_body_rejects_other_services() { + assert!(!is_openhuman_root_body(r#"{"name": "something-else"}"#)); + assert!(!is_openhuman_root_body(r#"{"ok": true}"#)); + assert!(!is_openhuman_root_body("not json at all")); + assert!(!is_openhuman_root_body("")); + // Wrong type for `name`. + assert!(!is_openhuman_root_body(r#"{"name": 42}"#)); +} + +#[test] +fn parse_lsof_pid_picks_first_pid() { + assert_eq!(parse_lsof_pid("12345\n"), Some(12345)); + // Multiple pids — pick the first non-empty line. lsof can emit several + // when multiple sockets share the port (IPv4/IPv6). + assert_eq!(parse_lsof_pid("\n 9876 \n12345\n"), Some(9876)); + assert_eq!(parse_lsof_pid(""), None); + assert_eq!(parse_lsof_pid("not-a-pid\n"), None); +} + +#[test] +fn parse_netstat_pid_finds_listening_entry() { + // Sample shape from `netstat -ano -p TCP` on Windows. + let stdout = "\ +Active Connections + + Proto Local Address Foreign Address State PID + TCP 0.0.0.0:135 0.0.0.0:0 LISTENING 1024 + TCP 127.0.0.1:7788 0.0.0.0:0 LISTENING 4242 + TCP 127.0.0.1:50000 127.0.0.1:7788 ESTABLISHED 5555 +"; + assert_eq!(parse_netstat_pid(stdout, 7788), Some(4242)); + assert_eq!(parse_netstat_pid(stdout, 9999), None); +} + // --------------------------------------------------------------------------- // Token generation tests // --------------------------------------------------------------------------- diff --git a/docs/BUILDING.md b/docs/BUILDING.md index b87266c21..127d0a9a2 100644 --- a/docs/BUILDING.md +++ b/docs/BUILDING.md @@ -182,3 +182,24 @@ If the lock is left behind by a crashed process (PID no longer alive), the prefl **Known limitation** Dev and release builds still share `com.openhuman.app` as the cache identifier. Isolating dev to a separate `com.openhuman.app.dev` cache requires changes to the vendored `tauri-runtime-cef` (cache path is built inside the runtime from the bundle identifier, not exposed to the openhuman shell). Tracked as a follow-up to #864. + +### Stale `openhuman` RPC process on the core port + +**Symptom** + +A previous Tauri build or `openhuman-core run` harness left a process listening on `OPENHUMAN_CORE_PORT` (default `7788`). Until issue #1130 the new Tauri build would silently attach to that listener — leading to version drift and 401s when the new build's `OPENHUMAN_CORE_TOKEN` didn't match. + +**Current behavior (issue #1130)** + +`core_process::ensure_running` now probes the port at startup: + +- If `GET /` identifies the listener as an OpenHuman core (JSON body with `"name": "openhuman"`), it is treated as a stale process from a previous run and proactively terminated (`SIGTERM`, then `SIGKILL` after 750ms on Unix; `taskkill /F /T /PID` on Windows). The Tauri host then spawns its own fresh embedded core. +- If the listener is something else (or doesn't speak HTTP), startup fails loudly with the conflict surfaced in the log instead of silently attaching. +- Set `OPENHUMAN_CORE_REUSE_EXISTING=1` to opt back into the legacy attach-to-anything behavior — useful when running `openhuman-core run` as a manual debugging harness. + +**Manual cleanup (still works)** + +```bash +pkill -f "OpenHuman.app/Contents" +pkill -f "openhuman-core" +```