mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-29 22:23:01 +00:00
1114 lines
46 KiB
Rust
1114 lines
46 KiB
Rust
//! In-process core lifecycle.
|
|
//!
|
|
//! 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.
|
|
//!
|
|
//! 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;
|
|
|
|
use parking_lot::RwLock;
|
|
use tokio::net::TcpStream;
|
|
use tokio::sync::oneshot::error::TryRecvError;
|
|
use tokio::sync::Mutex;
|
|
use tokio::task::JoinHandle;
|
|
use tokio::time::{timeout, Duration};
|
|
use tokio_util::sync::CancellationToken;
|
|
|
|
use crate::process_kill::{kill_pid_force, kill_pid_term};
|
|
|
|
const CORE_READY_POLL_MS: u64 = 100;
|
|
// 60s ceiling. The embedded core boots in ~11s standalone, but inside the
|
|
// desktop app it contends with CEF process startup (GPU/renderer/utility
|
|
// re-execs) and can intermittently cross a 20s cap on a cold/loaded machine.
|
|
// Polling returns the instant the core is ready, so a higher ceiling costs
|
|
// nothing in the common fast case — it only avoids spurious startup-timeout
|
|
// aborts under contention.
|
|
const CORE_READY_ATTEMPTS: usize = 600;
|
|
const CORE_READY_TIMEOUT_MS: u64 = CORE_READY_POLL_MS * CORE_READY_ATTEMPTS as u64;
|
|
|
|
/// Generate a 256-bit cryptographically-random bearer token as a hex string.
|
|
///
|
|
/// Uses the same encoding as `openhuman_core::core::auth::generate_token`
|
|
/// (`hex::encode`) so the token format never silently diverges between the
|
|
/// Tauri-side generator and the core-side validator.
|
|
pub fn generate_rpc_token() -> String {
|
|
use rand::RngCore as _;
|
|
let mut bytes = [0u8; 32];
|
|
rand::rng().fill_bytes(&mut bytes);
|
|
hex::encode(bytes)
|
|
}
|
|
|
|
static CURRENT_RPC_TOKEN: LazyLock<RwLock<Option<String>>> = LazyLock::new(|| RwLock::new(None));
|
|
|
|
pub fn current_rpc_token() -> Option<String> {
|
|
CURRENT_RPC_TOKEN.read().clone()
|
|
}
|
|
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct PortFallbackNotice {
|
|
pub preferred_port: u16,
|
|
pub chosen_port: u16,
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
pub struct CoreProcessHandle {
|
|
task: Arc<Mutex<Option<JoinHandle<anyhow::Result<()>>>>>,
|
|
shutdown_token: Arc<Mutex<CancellationToken>>,
|
|
restart_lock: Arc<Mutex<()>>,
|
|
preferred_port: u16,
|
|
active_port: Arc<RwLock<u16>>,
|
|
last_port_fallback: Arc<RwLock<Option<PortFallbackNotice>>>,
|
|
/// Bearer token the embedded server validates on every inbound request.
|
|
///
|
|
/// Handed to the embedded server **in-memory** (via the `rpc_token`
|
|
/// argument of [`openhuman_core::core::jsonrpc::run_server_embedded_with_ready`])
|
|
/// rather than through `OPENHUMAN_CORE_TOKEN` on the process environment.
|
|
/// Avoiding the env crossing keeps the bearer off `/proc/<pid>/environ`
|
|
/// (Linux) and out of `sysctl KERN_PROCARGS2` / `ps eww -p <pid>` (macOS)
|
|
/// where any same-UID process could otherwise read it without entitlement.
|
|
/// The same value is exposed to the renderer via the `core_rpc_token`
|
|
/// Tauri command so every RPC call can attach `Authorization: Bearer`.
|
|
rpc_token: Arc<String>,
|
|
}
|
|
|
|
impl CoreProcessHandle {
|
|
pub fn new(port: u16) -> Self {
|
|
// CURRENT_RPC_TOKEN is intentionally NOT set here. It is published by
|
|
// ensure_running() only after the embedded server has been spawned
|
|
// with this token handed over via the in-memory `rpc_token` arg of
|
|
// `run_server_embedded_with_ready`. Setting it here would advertise
|
|
// a token that an existing process listening on the port (the
|
|
// harness-attach fast-path) has never seen, causing 401s on every
|
|
// authenticated call.
|
|
let rpc_token = generate_rpc_token();
|
|
Self {
|
|
task: Arc::new(Mutex::new(None)),
|
|
shutdown_token: Arc::new(Mutex::new(CancellationToken::new())),
|
|
restart_lock: Arc::new(Mutex::new(())),
|
|
preferred_port: port,
|
|
active_port: Arc::new(RwLock::new(port)),
|
|
last_port_fallback: Arc::new(RwLock::new(None)),
|
|
rpc_token: Arc::new(rpc_token),
|
|
}
|
|
}
|
|
|
|
/// The bearer token the embedded core validates on inbound RPC requests.
|
|
pub fn rpc_token(&self) -> &str {
|
|
&self.rpc_token
|
|
}
|
|
|
|
pub fn rpc_url(&self) -> String {
|
|
format!("http://127.0.0.1:{}/rpc", self.port())
|
|
}
|
|
|
|
pub fn port(&self) -> u16 {
|
|
*self.active_port.read()
|
|
}
|
|
|
|
pub fn take_last_port_fallback_notice(&self) -> Option<PortFallbackNotice> {
|
|
self.last_port_fallback.write().take()
|
|
}
|
|
|
|
/// Acquire the restart lock to serialize overlapping restart requests.
|
|
pub async fn restart_lock(&self) -> tokio::sync::MutexGuard<'_, ()> {
|
|
self.restart_lock.lock().await
|
|
}
|
|
|
|
async fn is_rpc_port_open(&self) -> bool {
|
|
is_port_open(self.port()).await
|
|
}
|
|
|
|
pub async fn ensure_running(&self) -> Result<(), String> {
|
|
// Idempotent fast path: if we already spawned the embedded server in
|
|
// *this* process and it's still alive on the port, the listener is
|
|
// us — return Ok without identifying or taking over. Without this,
|
|
// a second `start_core_process` call (e.g. HMR re-mounting the boot
|
|
// gate) sees its own port as bound, classifies the listener as
|
|
// "stale OpenHuman", and walks into the SIGTERM/SIGKILL takeover
|
|
// path against itself. (#1130 takeover is meant to recover from
|
|
// *external* leftover binaries, not our own in-process spawn.)
|
|
{
|
|
let guard = self.task.lock().await;
|
|
if let Some(task) = guard.as_ref() {
|
|
if !task.is_finished() && self.is_rpc_port_open().await {
|
|
log::debug!(
|
|
"[core] ensure_running: embedded task already running on port {} — no-op",
|
|
self.port()
|
|
);
|
|
return Ok(());
|
|
}
|
|
}
|
|
}
|
|
|
|
if is_port_open(self.preferred_port).await {
|
|
// Idempotent fast-path: if we already own a running embedded
|
|
// task, the listener on this port is us — not a stale external
|
|
// process. Without this short-circuit, a second `ensure_running`
|
|
// call (from BootCheckGate re-render, React StrictMode mount, or
|
|
// any double-invoke of `start_core_process`) hits the
|
|
// `identify_listener` path, identifies the listener as
|
|
// OpenHuman, calls `takeover_stale_listener`, and aborts with
|
|
// "stale-listener pid <self> matches the Tauri host pid;
|
|
// refusing to self-terminate". (#1316 introduced the
|
|
// frontend-driven `start_core_process` invoke without
|
|
// hardening `ensure_running` against double-invoke.)
|
|
{
|
|
let guard = self.task.lock().await;
|
|
if let Some(task) = guard.as_ref() {
|
|
if !task.is_finished() {
|
|
log::debug!(
|
|
"[core] ensure_running: embedded task already running on port {}, returning Ok (idempotent)",
|
|
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.preferred_port
|
|
);
|
|
return Ok(());
|
|
}
|
|
|
|
match identify_listener(self.preferred_port).await {
|
|
ListenerKind::OpenHuman => {
|
|
log::warn!(
|
|
"[core] found stale OpenHuman listener on port {} — taking over (issue #1130)",
|
|
self.preferred_port
|
|
);
|
|
self.takeover_stale_listener().await?;
|
|
// Fall through to spawn-and-wait below.
|
|
}
|
|
ListenerKind::Unknown { reason } => {
|
|
if is_expected_port_clash(&reason) {
|
|
log::warn!(
|
|
"[core] preferred RPC port {} is occupied by non-OpenHuman listener ({reason}); attempting fallback bind range",
|
|
self.preferred_port
|
|
);
|
|
} else {
|
|
log::error!(
|
|
"[core] preferred RPC port {} occupied by unexpected listener ({reason}); attempting fallback bind range",
|
|
self.preferred_port
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
for startup_attempt in 0..=1u8 {
|
|
let mut retry_after_takeover = false;
|
|
let shutdown_token = self.fresh_shutdown_token().await;
|
|
let (ready_tx, mut ready_rx) = tokio::sync::oneshot::channel::<
|
|
openhuman_core::core::jsonrpc::EmbeddedReadySignal,
|
|
>();
|
|
let mut received_ready = false;
|
|
|
|
{
|
|
let mut guard = self.task.lock().await;
|
|
if guard.is_none() {
|
|
let port = self.preferred_port;
|
|
// RPC bearer is handed to the embedded server in-memory
|
|
// via the `rpc_token` argument of
|
|
// run_server_embedded_with_ready (see below) — never
|
|
// through OPENHUMAN_CORE_TOKEN on the process env.
|
|
// Sidecar-era env-var transport was a leftover from the
|
|
// PR #1061 cleanup; with the core in-process there is no
|
|
// child process that needs the env crossing, and
|
|
// sharing the bearer via env put it within reach of any
|
|
// same-UID process that could read /proc/<pid>/environ
|
|
// (Linux) or sysctl KERN_PROCARGS2 / ps eww -p <pid>
|
|
// (macOS).
|
|
let token_for_core = self.rpc_token.clone();
|
|
// Surface the Tauri shell version to the in-process core so
|
|
// backend-bound HTTP requests can attach `x-tauri-version`
|
|
// analytics headers alongside `x-core-version`.
|
|
std::env::set_var("OPENHUMAN_TAURI_VERSION", env!("CARGO_PKG_VERSION"));
|
|
*self.active_port.write() = port;
|
|
*self.last_port_fallback.write() = None;
|
|
|
|
// Debug-build only: surface the RPC bearer token at a known
|
|
// tmpdir path so the e2e test runner (a separate Node process)
|
|
// can authenticate against the in-process core. Release builds
|
|
// never write this file. The test harness reads it from
|
|
// ${tmpdir}/openhuman-e2e-rpc-token.
|
|
//
|
|
// Token file is owner-read-write only (mode 0600) on Unix so a
|
|
// shared dev box doesn't leak the bearer to other local users.
|
|
#[cfg(debug_assertions)]
|
|
{
|
|
use std::io::Write as _;
|
|
let token_path = std::env::temp_dir().join("openhuman-e2e-rpc-token");
|
|
let write_result = (|| -> std::io::Result<()> {
|
|
let mut options = std::fs::OpenOptions::new();
|
|
options.create(true).write(true).truncate(true);
|
|
#[cfg(unix)]
|
|
{
|
|
use std::os::unix::fs::OpenOptionsExt as _;
|
|
options.mode(0o600);
|
|
}
|
|
let mut file = options.open(&token_path)?;
|
|
file.write_all(self.rpc_token.as_bytes())?;
|
|
file.sync_all()?;
|
|
Ok(())
|
|
})();
|
|
if let Err(err) = write_result {
|
|
log::warn!(
|
|
"[core] failed to write e2e token file at {}: {err}",
|
|
token_path.display()
|
|
);
|
|
} else {
|
|
log::debug!(
|
|
"[core] wrote e2e token file at {} (debug build only)",
|
|
token_path.display()
|
|
);
|
|
}
|
|
}
|
|
log::info!(
|
|
"[core] spawning embedded in-process core server on preferred port {port}"
|
|
);
|
|
let task = tokio::spawn(async move {
|
|
openhuman_core::core::jsonrpc::run_server_embedded_with_ready(
|
|
None,
|
|
Some(port),
|
|
true,
|
|
shutdown_token,
|
|
ready_tx,
|
|
// In-memory bearer handoff: the embedded server
|
|
// seeds its auth subsystem from this value via
|
|
// `auth::init_rpc_token_with_value`, so the token
|
|
// never crosses OPENHUMAN_CORE_TOKEN on the
|
|
// process env.
|
|
Some(token_for_core),
|
|
)
|
|
.await
|
|
});
|
|
*guard = Some(task);
|
|
// Publish only after the embedded server has been spawned
|
|
// with the in-memory bearer in scope. Setting this earlier
|
|
// would advertise a token to the frontend that the server
|
|
// hadn't loaded yet.
|
|
*CURRENT_RPC_TOKEN.write() = Some(self.rpc_token.to_string());
|
|
log::debug!("[auth] CURRENT_RPC_TOKEN set after embedded spawn");
|
|
}
|
|
}
|
|
|
|
// Readiness budget: 600 iterations x 100ms = 60s (CORE_READY_ATTEMPTS).
|
|
// The embedded core's JSON-RPC controller registry has grown over
|
|
// time and earlier 4s/10s/20s budgets started flaking under CI
|
|
// worker load (issue: core_process tests intermittently failing with
|
|
// "core process did not become ready"), especially under
|
|
// cargo-llvm-cov instrumentation where the binary runs ~2x
|
|
// slower, and under desktop CEF startup contention. 60s is still
|
|
// well under any user-visible startup
|
|
// expectation: in normal runs the ready signal arrives in well
|
|
// under 1s and the loop exits immediately; the headroom only
|
|
// matters on heavily loaded instrumented CI workers.
|
|
for _ in 0..CORE_READY_ATTEMPTS {
|
|
if !received_ready {
|
|
match ready_rx.try_recv() {
|
|
Ok(ready_signal) => {
|
|
self.apply_embedded_ready_signal(ready_signal);
|
|
received_ready = true;
|
|
}
|
|
Err(TryRecvError::Empty) => {}
|
|
Err(TryRecvError::Closed) => {}
|
|
}
|
|
}
|
|
|
|
if received_ready && self.is_rpc_port_open().await {
|
|
log::info!("[core] core rpc became ready at {}", self.rpc_url());
|
|
return Ok(());
|
|
}
|
|
|
|
let mut guard = self.task.lock().await;
|
|
if let Some(task) = guard.as_ref() {
|
|
if task.is_finished() {
|
|
let task = guard.take().expect("checked is_some");
|
|
drop(guard);
|
|
return match task.await {
|
|
Ok(Ok(())) => {
|
|
Err("in-process core server exited before becoming ready"
|
|
.to_string())
|
|
}
|
|
Ok(Err(err)) => {
|
|
if let Some(openhuman_core::openhuman::connectivity::rpc::PickListenPortError::WouldTakeOver { preferred, .. }) = err
|
|
.downcast_ref::<openhuman_core::openhuman::connectivity::rpc::PickListenPortError>()
|
|
{
|
|
if startup_attempt == 0 {
|
|
log::warn!(
|
|
"[core] preferred port {} requested stale-listener takeover from embedded bind path; retrying once",
|
|
preferred
|
|
);
|
|
self.takeover_stale_listener().await?;
|
|
retry_after_takeover = true;
|
|
break;
|
|
}
|
|
}
|
|
Err(format!(
|
|
"in-process core server exited before becoming ready: {err}"
|
|
))
|
|
}
|
|
Err(err) => Err(format!(
|
|
"in-process core server task failed before ready: {err}"
|
|
)),
|
|
};
|
|
}
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(CORE_READY_POLL_MS)).await;
|
|
}
|
|
|
|
if retry_after_takeover {
|
|
continue;
|
|
}
|
|
|
|
// One last non-sleeping check avoids declaring a timeout when the
|
|
// ready signal arrived during the final poll sleep.
|
|
if !received_ready {
|
|
if let Ok(ready_signal) = ready_rx.try_recv() {
|
|
self.apply_embedded_ready_signal(ready_signal);
|
|
received_ready = true;
|
|
}
|
|
}
|
|
if received_ready && self.is_rpc_port_open().await {
|
|
log::info!("[core] core rpc became ready at {}", self.rpc_url());
|
|
return Ok(());
|
|
}
|
|
|
|
let port_open = self.is_rpc_port_open().await;
|
|
return Err(self
|
|
.cleanup_startup_timeout(received_ready, port_open, startup_attempt + 1)
|
|
.await);
|
|
}
|
|
|
|
let port_open = self.is_rpc_port_open().await;
|
|
Err(self.cleanup_startup_timeout(false, port_open, 2).await)
|
|
}
|
|
|
|
async fn cleanup_startup_timeout(
|
|
&self,
|
|
received_ready: bool,
|
|
port_open: bool,
|
|
attempt: u8,
|
|
) -> String {
|
|
let port = self.port();
|
|
let task_state = {
|
|
let guard = self.task.lock().await;
|
|
match guard.as_ref() {
|
|
None => "missing",
|
|
Some(task) if task.is_finished() => "finished",
|
|
Some(_) => "running",
|
|
}
|
|
};
|
|
log::error!(
|
|
"[core] startup timed out after {CORE_READY_TIMEOUT_MS}ms \
|
|
(port={port}, ready_signal={received_ready}, port_open={port_open}, \
|
|
task_state={task_state}, attempt={attempt}); \
|
|
aborting embedded startup task before retry"
|
|
);
|
|
self.cancel_shutdown_token(" after startup timeout").await;
|
|
self.abort_task(" after startup timeout").await;
|
|
format!(
|
|
"core process did not become ready within {CORE_READY_TIMEOUT_MS}ms \
|
|
(port={port}, ready_signal={received_ready}, port_open={port_open}, \
|
|
task_state={task_state}, attempt={attempt})"
|
|
)
|
|
}
|
|
|
|
pub(crate) fn apply_embedded_ready_signal(
|
|
&self,
|
|
ready: openhuman_core::core::jsonrpc::EmbeddedReadySignal,
|
|
) {
|
|
*self.active_port.write() = ready.port;
|
|
std::env::set_var("OPENHUMAN_CORE_RPC_URL", self.rpc_url());
|
|
if let Some(preferred) = ready.fallback_from {
|
|
let message = format!("port_fallback_engaged: {preferred} -> {}", ready.port);
|
|
log::warn!("[core] {message}");
|
|
sentry::add_breadcrumb(sentry::Breadcrumb {
|
|
category: Some("core.port".to_string()),
|
|
level: sentry::Level::Warning,
|
|
message: Some(message),
|
|
..Default::default()
|
|
});
|
|
*self.last_port_fallback.write() = Some(PortFallbackNotice {
|
|
preferred_port: preferred,
|
|
chosen_port: ready.port,
|
|
});
|
|
} else {
|
|
*self.last_port_fallback.write() = None;
|
|
}
|
|
}
|
|
|
|
/// 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 port = self.preferred_port;
|
|
let pid = match find_pid_on_port(port) {
|
|
Some(pid) => pid,
|
|
None => {
|
|
return Err(format!(
|
|
"could not determine pid bound to port {} — refusing to take over",
|
|
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)",
|
|
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(port).await {
|
|
match find_pid_on_port(port) {
|
|
Some(current) if current == pid => {
|
|
log::warn!(
|
|
"[core] pid {pid} still bound to port {} after SIGTERM — escalating to SIGKILL",
|
|
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",
|
|
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(port).await {
|
|
if waited_ms >= MAX_WAIT_MS {
|
|
return Err(format!(
|
|
"signaled pid {pid} but port {} remained bound after {MAX_WAIT_MS}ms",
|
|
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",
|
|
port
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
/// Restart the embedded core to pick up updated macOS permission grants.
|
|
///
|
|
/// macOS caches permission state per-process; restarting forces a fresh
|
|
/// read. If something else is bound to the port (e.g. a manual
|
|
/// `openhuman-core run` harness) we surface that instead of looping.
|
|
///
|
|
/// Issue: <https://github.com/tinyhumansai/openhuman/issues/133>
|
|
pub async fn restart(&self) -> Result<(), String> {
|
|
log::info!("[core] restarting embedded core server for permission refresh");
|
|
|
|
let had_managed_task = {
|
|
let guard = self.task.lock().await;
|
|
guard.is_some()
|
|
};
|
|
|
|
self.shutdown().await;
|
|
if !had_managed_task {
|
|
log::debug!(
|
|
"[core] restart: no managed embedded task was running; ensure_running will resolve ownership/fallback"
|
|
);
|
|
}
|
|
|
|
const POLL_MS: u64 = 50;
|
|
const MAX_WAIT_MS: u64 = 10_000;
|
|
let mut waited_ms: u64 = 0;
|
|
while self.is_rpc_port_open().await {
|
|
if waited_ms >= MAX_WAIT_MS {
|
|
return Err(format!(
|
|
"Core RPC port {} did not become free after stopping the embedded server.",
|
|
self.port()
|
|
));
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(POLL_MS)).await;
|
|
waited_ms += POLL_MS;
|
|
}
|
|
|
|
let result = self.ensure_running().await;
|
|
match &result {
|
|
Ok(()) => log::info!("[core] restart: embedded core ready after restart"),
|
|
Err(e) => log::error!("[core] restart: failed to restart embedded core: {e}"),
|
|
}
|
|
result
|
|
}
|
|
|
|
/// Lock the task slot, take its handle if any, and abort it. Shared by
|
|
/// `shutdown` (cleanup-on-drop semantics) and `send_terminate_signal`
|
|
/// (cooperative early teardown from `RunEvent::ExitRequested`).
|
|
async fn abort_task(&self, log_context: &str) {
|
|
let mut task_guard = self.task.lock().await;
|
|
if let Some(task) = task_guard.take() {
|
|
log::info!("[core] aborting embedded core server task{log_context}");
|
|
task.abort();
|
|
}
|
|
}
|
|
|
|
async fn fresh_shutdown_token(&self) -> CancellationToken {
|
|
let mut guard = self.shutdown_token.lock().await;
|
|
if guard.is_cancelled() {
|
|
log::debug!("[core] resetting embedded core shutdown token for new spawn");
|
|
*guard = CancellationToken::new();
|
|
}
|
|
guard.clone()
|
|
}
|
|
|
|
async fn cancel_shutdown_token(&self, log_context: &str) {
|
|
let token = self.shutdown_token.lock().await.clone();
|
|
if token.is_cancelled() {
|
|
log::debug!("[core] embedded core shutdown token already cancelled{log_context}");
|
|
} else {
|
|
log::info!("[core] cancelling embedded core shutdown token{log_context}");
|
|
token.cancel();
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
async fn shutdown_token_is_cancelled(&self) -> bool {
|
|
self.shutdown_token.lock().await.is_cancelled()
|
|
}
|
|
|
|
/// Stop the embedded server task. Safe to call when nothing is running.
|
|
pub async fn shutdown(&self) {
|
|
self.cancel_shutdown_token("").await;
|
|
let task = {
|
|
let mut task_guard = self.task.lock().await;
|
|
task_guard.take()
|
|
};
|
|
let Some(mut task) = task else {
|
|
return;
|
|
};
|
|
|
|
match timeout(Duration::from_secs(5), &mut task).await {
|
|
Ok(Ok(Ok(()))) => {
|
|
log::info!("[core] embedded core server task stopped gracefully");
|
|
}
|
|
Ok(Ok(Err(err))) => {
|
|
log::warn!("[core] embedded core server task ended during shutdown: {err}");
|
|
}
|
|
Ok(Err(err)) => {
|
|
log::warn!("[core] embedded core server task join failed during shutdown: {err}");
|
|
}
|
|
Err(_) => {
|
|
log::warn!(
|
|
"[core] graceful embedded core shutdown timed out; aborting server task"
|
|
);
|
|
task.abort();
|
|
let _ = task.await;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Synchronous-friendly shutdown for `RunEvent::ExitRequested`.
|
|
///
|
|
/// Aborts the embedded server task so any background tokio tasks the
|
|
/// server spawned stop driving I/O before CEF's teardown runs. Cheap
|
|
/// and non-blocking on the UI thread — `JoinHandle::abort` returns
|
|
/// immediately.
|
|
pub async fn send_terminate_signal(&self) {
|
|
self.cancel_shutdown_token(" on app shutdown").await;
|
|
self.abort_task(" on app shutdown").await;
|
|
}
|
|
}
|
|
|
|
/// A non-OpenHuman process holding the core RPC port. Surfaced to the frontend
|
|
/// so the user can see what to free, and so the consent-gated force-quit knows
|
|
/// which pid the user agreed to terminate.
|
|
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
|
pub struct PortOwner {
|
|
pub pid: u32,
|
|
pub name: String,
|
|
}
|
|
|
|
/// Result returned to the frontend after a port-conflict auto-recovery attempt.
|
|
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
|
pub struct RecoveryOutcome {
|
|
pub success: bool,
|
|
pub message: String,
|
|
pub new_port: Option<u16>,
|
|
/// Set when recovery failed because a foreign process still holds the port.
|
|
/// `None` on success or when the owner can't be identified (e.g. a kernel
|
|
/// reservation), in which case the frontend offers guidance, not a kill.
|
|
pub foreign_owner: Option<PortOwner>,
|
|
}
|
|
|
|
impl CoreProcessHandle {
|
|
/// Attempt to recover from a port conflict: reap stale OpenHuman processes,
|
|
/// wait briefly for the port to free, then start the embedded core.
|
|
///
|
|
/// Called from the `recover_port_conflict` Tauri command when the frontend's
|
|
/// boot-check detects the core is unreachable due to a port conflict.
|
|
pub async fn recover_port_conflict(&self) -> RecoveryOutcome {
|
|
log::debug!(
|
|
"[core_process] recover_port_conflict: starting recovery for port {}",
|
|
self.preferred_port
|
|
);
|
|
|
|
tokio::task::spawn_blocking(crate::process_recovery::reap_stale_openhuman_processes)
|
|
.await
|
|
.unwrap_or_else(|e| {
|
|
log::warn!("[core_process] recover_port_conflict: reap task panicked: {e}")
|
|
});
|
|
log::debug!("[core_process] recover_port_conflict: stale process reap complete");
|
|
|
|
// Give the OS time to release the port after process termination.
|
|
tokio::time::sleep(Duration::from_millis(500)).await;
|
|
log::debug!("[core_process] recover_port_conflict: post-reap wait complete");
|
|
|
|
match self.ensure_running().await {
|
|
Ok(()) => {
|
|
let new_port = self.port();
|
|
log::info!(
|
|
"[core_process] recover_port_conflict: recovery succeeded, core on port {new_port}"
|
|
);
|
|
RecoveryOutcome {
|
|
success: true,
|
|
message: format!("Core recovered on port {new_port}"),
|
|
new_port: Some(new_port),
|
|
foreign_owner: None,
|
|
}
|
|
}
|
|
Err(err) => {
|
|
// Reaping stale OpenHuman processes didn't free the port, so a
|
|
// foreign process is holding it. Identify it (name + pid) so the
|
|
// frontend can surface it and offer a consent-gated force-quit.
|
|
let port = self.preferred_port;
|
|
let foreign_owner = tokio::task::spawn_blocking(move || identify_port_owner(port))
|
|
.await
|
|
.unwrap_or(None);
|
|
log::warn!(
|
|
"[core_process] recover_port_conflict: recovery failed: {err}; foreign_owner={foreign_owner:?}"
|
|
);
|
|
RecoveryOutcome {
|
|
success: false,
|
|
message: format!("Recovery failed: {err}"),
|
|
new_port: None,
|
|
foreign_owner,
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Terminate the foreign process the user consented to (the pid surfaced in
|
|
/// the prior [`RecoveryOutcome::foreign_owner`]), then restart the core.
|
|
///
|
|
/// Re-validates that `expected_pid` still owns the port immediately before
|
|
/// killing — closing the PID-reuse window and refusing if the owner changed,
|
|
/// the port is already free, or the owner is OpenHuman's own process.
|
|
pub async fn force_quit_port_owner(&self, expected_pid: u32) -> RecoveryOutcome {
|
|
let port = self.preferred_port;
|
|
let self_pid = std::process::id();
|
|
log::info!("[core_process] force_quit_port_owner: port {port} expected_pid {expected_pid}");
|
|
|
|
let current = tokio::task::spawn_blocking(move || find_pid_on_port(port))
|
|
.await
|
|
.unwrap_or(None);
|
|
let pid = match validate_kill_target(current, expected_pid, self_pid) {
|
|
Ok(pid) => pid,
|
|
Err(reason) => {
|
|
log::warn!("[core_process] force_quit_port_owner: refused: {reason}");
|
|
return RecoveryOutcome {
|
|
success: false,
|
|
message: reason,
|
|
new_port: None,
|
|
foreign_owner: None,
|
|
};
|
|
}
|
|
};
|
|
|
|
// Best-effort: a TERM failure is non-fatal because the escalation below
|
|
// re-validates ownership and force-kills if the process is still there.
|
|
if let Err(err) =
|
|
tokio::task::spawn_blocking(move || crate::process_kill::kill_pid_term(pid))
|
|
.await
|
|
.unwrap_or_else(|e| Err(format!("SIGTERM task panicked: {e}")))
|
|
{
|
|
log::debug!(
|
|
"[core_process] force_quit_port_owner: SIGTERM pid {pid} (best-effort): {err}"
|
|
);
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(500)).await;
|
|
|
|
// Re-validate before escalating: a bare `== Some(pid)` would SIGKILL a
|
|
// different process that reused the pid and rebound the port within the
|
|
// grace window, and would skip the self/protected-pid guards.
|
|
let still = tokio::task::spawn_blocking(move || find_pid_on_port(port))
|
|
.await
|
|
.unwrap_or(None);
|
|
if validate_kill_target(still, pid, self_pid).is_ok() {
|
|
if let Err(err) =
|
|
tokio::task::spawn_blocking(move || crate::process_kill::kill_pid_force(pid))
|
|
.await
|
|
.unwrap_or_else(|e| Err(format!("force-kill task panicked: {e}")))
|
|
{
|
|
log::warn!("[core_process] force_quit_port_owner: force-kill pid {pid}: {err}");
|
|
return RecoveryOutcome {
|
|
success: false,
|
|
message: format!("Could not terminate pid {pid}: {err}"),
|
|
new_port: None,
|
|
foreign_owner: None,
|
|
};
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(500)).await;
|
|
}
|
|
|
|
match self.ensure_running().await {
|
|
Ok(()) => {
|
|
let new_port = self.port();
|
|
log::info!(
|
|
"[core_process] force_quit_port_owner: core recovered on port {new_port}"
|
|
);
|
|
RecoveryOutcome {
|
|
success: true,
|
|
message: format!("Core recovered on port {new_port}"),
|
|
new_port: Some(new_port),
|
|
foreign_owner: None,
|
|
}
|
|
}
|
|
Err(err) => RecoveryOutcome {
|
|
success: false,
|
|
message: format!("Recovery failed after terminating pid {pid}: {err}"),
|
|
new_port: None,
|
|
foreign_owner: None,
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
pub fn default_core_port() -> u16 {
|
|
std::env::var("OPENHUMAN_CORE_PORT")
|
|
.ok()
|
|
.and_then(|v| v.parse::<u16>().ok())
|
|
.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.
|
|
pub(crate) 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:<port>/` 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)
|
|
}
|
|
|
|
/// Returns true when a port conflict is deterministic environment state, not
|
|
/// a high-signal unknown squatter worth sending to Sentry at error level.
|
|
fn is_expected_port_clash(reason: &str) -> bool {
|
|
let reason = reason.to_ascii_lowercase();
|
|
reason.contains("error sending request for url")
|
|
|| reason.contains("connection refused")
|
|
|| reason.contains("returned status 404")
|
|
|| reason.contains("returned status 200")
|
|
|| reason.contains("body did not identify as openhuman")
|
|
|| reason.contains("already in use by another process")
|
|
|| reason.contains("os error 10013")
|
|
|| reason.contains("wsaeacces")
|
|
}
|
|
|
|
#[cfg(unix)]
|
|
fn find_pid_on_port(port: u16) -> Option<u32> {
|
|
// Single port-qualified selector: `-iTCP:<port>` ANDs protocol+port. Two
|
|
// separate `-i` flags (`-iTCP -i:<port>`) are ORed by lsof and would match
|
|
// any TCP listener — risking returning an unrelated process as the owner.
|
|
let output = std::process::Command::new("lsof")
|
|
.args(["-nP", &format!("-iTCP:{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<u32> {
|
|
use std::os::windows::process::CommandExt;
|
|
const CREATE_NO_WINDOW: u32 = 0x0800_0000;
|
|
let output = std::process::Command::new("netstat")
|
|
.args(["-ano", "-p", "TCP"])
|
|
.creation_flags(CREATE_NO_WINDOW)
|
|
.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<u32> {
|
|
stdout
|
|
.lines()
|
|
.map(str::trim)
|
|
.find(|l| !l.is_empty())
|
|
.and_then(|l| l.parse::<u32>().ok())
|
|
}
|
|
|
|
/// Pure parse of `netstat -ano` output for a LISTENING entry on `port`.
|
|
///
|
|
/// Skips kernel-protected PIDs 0 (System Idle Process) and 4 (NT Kernel) —
|
|
/// `HTTP.sys` and kernel-mode socket reservations occasionally surface as
|
|
/// LISTENING under PID 4 even though no user-mode owner exists. Killing
|
|
/// those is impossible and would otherwise abort startup recovery; if the
|
|
/// "owner" is the kernel, callers should fall back to a port reroute
|
|
/// instead of trying to take over.
|
|
#[allow(dead_code)] // exercised only on windows builds
|
|
fn parse_netstat_pid(stdout: &str, port: u16) -> Option<u32> {
|
|
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::<u32>() {
|
|
if pid == 0 || pid == 4 {
|
|
log::warn!(
|
|
"[core] netstat reports port {port} owned by protected windows pid {pid}; treating as no-owner"
|
|
);
|
|
continue;
|
|
}
|
|
return Some(pid);
|
|
}
|
|
}
|
|
}
|
|
None
|
|
}
|
|
|
|
/// Identify the process currently holding `port`, resolving its pid (via
|
|
/// [`find_pid_on_port`]) to a human-readable name. Returns `None` when the
|
|
/// port is free or owned by an unidentifiable (e.g. kernel-reserved) holder.
|
|
fn identify_port_owner(port: u16) -> Option<PortOwner> {
|
|
let pid = find_pid_on_port(port)?;
|
|
// Empty when the name can't be resolved — the frontend already shows the
|
|
// pid, so a `"PID <n>"` fallback here would render as "PID n (PID n)".
|
|
let name = process_name(pid).unwrap_or_default();
|
|
Some(PortOwner { pid, name })
|
|
}
|
|
|
|
#[cfg(windows)]
|
|
fn process_name(pid: u32) -> Option<String> {
|
|
use std::os::windows::process::CommandExt;
|
|
const CREATE_NO_WINDOW: u32 = 0x0800_0000;
|
|
let output = std::process::Command::new("tasklist")
|
|
.args(["/FI", &format!("PID eq {pid}"), "/FO", "CSV", "/NH"])
|
|
.creation_flags(CREATE_NO_WINDOW)
|
|
.output()
|
|
.ok()?;
|
|
if !output.status.success() {
|
|
return None;
|
|
}
|
|
parse_tasklist_name(&String::from_utf8_lossy(&output.stdout))
|
|
}
|
|
|
|
#[cfg(unix)]
|
|
fn process_name(pid: u32) -> Option<String> {
|
|
let output = std::process::Command::new("ps")
|
|
.args(["-p", &pid.to_string(), "-o", "comm="])
|
|
.output()
|
|
.ok()?;
|
|
if !output.status.success() {
|
|
return None;
|
|
}
|
|
parse_ps_comm(&String::from_utf8_lossy(&output.stdout))
|
|
}
|
|
|
|
/// Pure parse of `tasklist /FO CSV /NH`: the image name is the first quoted
|
|
/// CSV field (`"chrome.exe","1234",...`). The "no tasks" sentinel and any
|
|
/// non-CSV line yield `None`.
|
|
#[allow(dead_code)] // exercised only on windows builds
|
|
fn parse_tasklist_name(stdout: &str) -> Option<String> {
|
|
let line = stdout.lines().map(str::trim).find(|l| !l.is_empty())?;
|
|
let name = line.strip_prefix('"')?.split('"').next()?.trim();
|
|
if name.is_empty() {
|
|
None
|
|
} else {
|
|
Some(name.to_string())
|
|
}
|
|
}
|
|
|
|
/// Pure parse of `ps -p <pid> -o comm=`: a single line, reduced to its
|
|
/// trailing path component so a full executable path collapses to a file name.
|
|
#[allow(dead_code)] // exercised only on unix builds
|
|
fn parse_ps_comm(stdout: &str) -> Option<String> {
|
|
let line = stdout.lines().map(str::trim).find(|l| !l.is_empty())?;
|
|
let name = line.rsplit(['/', '\\']).next().unwrap_or(line).trim();
|
|
if name.is_empty() {
|
|
None
|
|
} else {
|
|
Some(name.to_string())
|
|
}
|
|
}
|
|
|
|
/// Decide whether the consent-gated force-quit may proceed. `current_owner` is
|
|
/// the pid `find_pid_on_port` reports right before the kill; `expected_pid` is
|
|
/// the owner the user was shown and agreed to terminate. Returns the pid to
|
|
/// kill, or the reason the kill must be refused.
|
|
fn validate_kill_target(
|
|
current_owner: Option<u32>,
|
|
expected_pid: u32,
|
|
self_pid: u32,
|
|
) -> Result<u32, String> {
|
|
match current_owner {
|
|
None => Err("the port is no longer held by any process".to_string()),
|
|
Some(pid) if pid == self_pid => Err(format!(
|
|
"refusing to terminate OpenHuman's own process (pid {pid})"
|
|
)),
|
|
// Defense-in-depth: low/kernel pids are never legitimate port owners and
|
|
// must never be signalled, even if one slipped past `find_pid_on_port`.
|
|
// 0 — `kill(0, sig)` broadcasts to the caller's whole process group on
|
|
// Unix; 1 — init/launchd; 4 — Windows NT kernel.
|
|
Some(pid) if pid <= 1 || pid == 4 => {
|
|
Err(format!("refusing to terminate protected pid {pid}"))
|
|
}
|
|
Some(pid) if pid != expected_pid => Err(format!(
|
|
"port owner changed (was pid {expected_pid}, now pid {pid}); not terminating"
|
|
)),
|
|
Some(pid) => Ok(pid),
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "core_process_tests.rs"]
|
|
mod tests;
|