From 673993393e4eb0e42ecc714f789370f45ebdb580 Mon Sep 17 00:00:00 2001 From: drakezhang Date: Thu, 10 Sep 2026 16:45:36 +0800 Subject: [PATCH 1/2] fix(daemon): verify IPC for discovery and process management Share a bounded IPC status probe across discovery, readiness and doctor. Do not treat local PID visibility as daemon availability. Only a missing or refused endpoint permits startup; preserve access, timeout and response errors, and re-read discovery metadata once across replacement races. Require the kernel's IPC peer PID before signaling a process. Hold the daemon lock before stale metadata cleanup, preserve replacement records, and keep stopping without discovery a no-op. Reuse the same stop checks for CLI self-update. Cover unavailable local PIDs, inconsistent replies, permission failures, timeouts, stale locks, concurrent starts and restart. Include an optional Linux PID namespace regression. This addresses the reliability portion of #214; sandbox process persistence and runtime paths remain separate. Validation: workspace tests pass serially (632 tests); final targeted regressions pass (24 tests); cargo fmt and strict workspace Clippy pass. An unrelated parallel skill-install test failure also reproduces on main. --- crates/bsk-cli/Cargo.toml | 1 + crates/bsk-cli/src/cli/doctor.rs | 96 ++--- crates/bsk-cli/src/cli/ensure_daemon.rs | 80 +--- crates/bsk-cli/src/cli/error.rs | 2 +- crates/bsk-cli/src/cli/update.rs | 6 +- crates/bsk-cli/src/daemon/info.rs | 4 +- crates/bsk-cli/src/daemon/lockfile.rs | 16 +- crates/bsk-cli/src/daemon/mod.rs | 1 + crates/bsk-cli/src/daemon/probe.rs | 168 ++++++++ crates/bsk-cli/src/daemon/start.rs | 215 +++++------ crates/bsk-cli/src/ipc_client.rs | 105 +++-- crates/bsk-cli/tests/daemon_discovery.rs | 469 +++++++++++++++++++++++ crates/bsk-cli/tests/status_cmd.rs | 20 +- 13 files changed, 862 insertions(+), 321 deletions(-) create mode 100644 crates/bsk-cli/src/daemon/probe.rs create mode 100644 crates/bsk-cli/tests/daemon_discovery.rs diff --git a/crates/bsk-cli/Cargo.toml b/crates/bsk-cli/Cargo.toml index 291901a..dfca670 100644 --- a/crates/bsk-cli/Cargo.toml +++ b/crates/bsk-cli/Cargo.toml @@ -64,6 +64,7 @@ libc = "0.2" windows-sys = { version = "0.59", features = [ "Win32_Foundation", "Win32_System_Threading", + "Win32_System_Pipes", "Win32_Security", "Win32_Storage_FileSystem", ] } diff --git a/crates/bsk-cli/src/cli/doctor.rs b/crates/bsk-cli/src/cli/doctor.rs index fcd0b1e..9f6b06f 100644 --- a/crates/bsk-cli/src/cli/doctor.rs +++ b/crates/bsk-cli/src/cli/doctor.rs @@ -7,11 +7,14 @@ use bsk_protocol::StatusResult; use console::style; use serde::Serialize; -use crate::cli::browser_wait::doctor_browser_connect_wait; +use crate::cli::browser_wait::{ + browser_query_ipc_timeout, doctor_browser_connect_wait, wait_for_browser_ms, +}; use crate::cli::ensure_daemon::ensure_daemon; -use crate::cli::status::{self, Output}; -use crate::daemon::info::{self, DaemonInfo}; +use crate::cli::status::Output; +use crate::daemon::info::DaemonInfo; use crate::daemon::paths; +use crate::daemon::probe::{self, Probe}; use crate::daemon::state::PROTOCOL_VERSION; /// Chrome Web Store listing for the browser-skill extension. @@ -123,12 +126,10 @@ pub fn has_failures(checks: &[CheckResult]) -> bool { fn resolve_daemon_state(output: Output) -> DaemonState { let mut state = current_state(Duration::ZERO); - if matches!(state, DaemonState::Missing | DaemonState::StaleDead(_)) && ensure_daemon().is_err() - { - return current_state(Duration::ZERO); - } - - if matches!(state, DaemonState::Missing | DaemonState::StaleDead(_)) { + if matches!(state, DaemonState::Missing | DaemonState::NoListener(_)) { + if let Err(err) = ensure_daemon() { + return DaemonState::ProbeError(format!("{err:#}")); + } state = current_state(Duration::ZERO); } @@ -159,32 +160,16 @@ fn needs_browser_wait(state: &DaemonState) -> bool { /// What the disk + IPC says about a possibly-running daemon. Threaded /// through every check so they share one snapshot. enum DaemonState { - /// No `daemon.json` at all. Missing, - /// `daemon.json` exists but reading it failed. - ReadError(String), - /// `daemon.json` exists but its pid is not alive on this host. - StaleDead(DaemonInfo), - /// `daemon.json` exists, pid is alive, but IPC didn't answer. - IpcUnreachable(DaemonInfo, String), - /// `daemon.json` exists, IPC answered, but the pid reported by - /// `system.status` does not match the pid recorded on disk. This - /// usually means a stale `daemon.json` left over from a previous - /// daemon points at a sock that now belongs to a different daemon. - PidMismatch { - info: DaemonInfo, - status: StatusResult, - }, - /// Everything lines up: pid alive + IPC answered + pid matches. + NoListener(DaemonInfo), + ProbeError(String), Verified { status: StatusResult }, } impl DaemonState { fn status(&self) -> Option<&StatusResult> { match self { - DaemonState::Verified { status, .. } | DaemonState::PidMismatch { status, .. } => { - Some(status) - } + DaemonState::Verified { status } => Some(status), _ => None, } } @@ -202,23 +187,17 @@ fn collect_checks(state: DaemonState) -> Vec { } fn current_state(browser_wait: Duration) -> DaemonState { - let info = match info::read() { - Ok(Some(info)) => info, - Ok(None) => return DaemonState::Missing, - Err(err) => return DaemonState::ReadError(format!("{err:#}")), + let params = bsk_protocol::StatusParams { + wait_for_browser_ms: wait_for_browser_ms(browser_wait), }; - if !crate::daemon::lockfile::pid_alive(info.pid) { - return DaemonState::StaleDead(info); - } - match status::query_sock_with_wait(info.sock_path.clone(), browser_wait) { - Ok(status) => { - if status.pid == info.pid { - DaemonState::Verified { status } - } else { - DaemonState::PidMismatch { info, status } - } - } - Err(err) => DaemonState::IpcUnreachable(info, format!("{err}")), + let timeout = browser_query_ipc_timeout(browser_wait, Duration::from_secs(2)); + match probe::probe_with_params(timeout, params) { + Ok(Probe::Ready(daemon)) => DaemonState::Verified { + status: daemon.status, + }, + Ok(Probe::Absent(Some(info))) => DaemonState::NoListener(info), + Ok(Probe::Absent(None)) => DaemonState::Missing, + Err(err) => DaemonState::ProbeError(format!("{err:#}")), } } @@ -320,28 +299,19 @@ fn check_daemon_running(state: &DaemonState) -> CheckResult { "daemon.json not found", "run `bsk daemon start` or any `bsk` command (daemon is auto-spawned)", ), - DaemonState::ReadError(err) => CheckResult::fail( - name, - format!("could not read daemon.json: {err}"), - "check permissions on ~/.bsk", - ), - DaemonState::StaleDead(info) => CheckResult::fail( - name, - format!("daemon.json is stale (pid {} does not exist)", info.pid), - "run `bsk daemon start` (or delete ~/.bsk/daemon.json)", - ), - DaemonState::IpcUnreachable(info, err) => CheckResult::fail( - name, - format!("pid {} is alive but IPC is unreachable: {err}", info.pid), - "run `bsk daemon restart`, then check `bsk logs`", - ), - DaemonState::PidMismatch { info, status } => CheckResult::fail( + DaemonState::NoListener(info) => CheckResult::fail( name, format!( - "daemon.json records pid {} but system.status returned pid {}", - info.pid, status.pid + "no daemon listening at {} (recorded pid {})", + info.sock_path.display(), + info.pid ), - "daemon.json is stale; run `bsk daemon stop && bsk status` to reset", + "run `bsk daemon start`; check `bsk logs` if startup fails", + ), + DaemonState::ProbeError(err) => CheckResult::fail( + name, + err.clone(), + "check daemon IPC permissions and `bsk logs`; keep existing runtime files", ), } } diff --git a/crates/bsk-cli/src/cli/ensure_daemon.rs b/crates/bsk-cli/src/cli/ensure_daemon.rs index 2243395..a1c01b8 100644 --- a/crates/bsk-cli/src/cli/ensure_daemon.rs +++ b/crates/bsk-cli/src/cli/ensure_daemon.rs @@ -2,32 +2,33 @@ //! subcommand. //! //! Flow (per design §3.1): -//! 1. Read `daemon.json`. If the recorded pid is alive, return its info. -//! 2. Otherwise spawn `bsk daemon start` (the same binary), inheriting -//! `BSK_HOME` if set, and poll `daemon.json` for a live pid until +//! 1. Verify the daemon over IPC and return its discovery info. +//! 2. Only if no endpoint is listening, spawn `bsk daemon start` (the same binary), inheriting +//! `BSK_HOME` if set, and poll for verified IPC readiness until //! [`SPAWN_DEADLINE`] elapses. //! 3. If polling times out, return an error with hints. use std::path::PathBuf; use std::process::{Command, Stdio}; -use std::time::{Duration, Instant}; +use std::time::Duration; use anyhow::{Context, Result}; -use bsk_protocol::{Method, StatusParams, StatusResult}; -use crate::daemon::info::{self, DaemonInfo}; +use crate::daemon::info::DaemonInfo; +use crate::daemon::probe::{self, PROBE_TIMEOUT, Probe}; /// Maximum time to wait for an auto-spawned daemon to become ready. pub const SPAWN_DEADLINE: Duration = Duration::from_millis(3_000); -/// Read `daemon.json` if it's valid; spawn the daemon otherwise. Returns -/// the connection handle the caller should use. +/// Return verified discovery info, starting a daemon only when its discovery +/// file or IPC listener is absent. pub fn ensure_daemon() -> Result { - if let Some(running) = read_verified()? { - return Ok(running); + if let Probe::Ready(daemon) = probe::probe(PROBE_TIMEOUT)? { + return Ok(daemon.info); } spawn_daemon()?; - wait_for_ready(SPAWN_DEADLINE) + probe::wait_for_ready(SPAWN_DEADLINE) + .map(|daemon| daemon.info) .with_context(|| "auto-spawned daemon failed to become ready in time") } @@ -38,65 +39,22 @@ fn spawn_daemon() -> Result<()> { .arg("start") .stdin(Stdio::null()) .stdout(Stdio::null()) - .stderr(Stdio::null()); + .stderr(Stdio::piped()); // The child re-uses inherited env (BSK_HOME etc), so tests that set // a temp home work transparently. - let status = cmd - .status() + let output = cmd + .output() .context("spawn `bsk daemon start` for auto-spawn")?; - if !status.success() { + if !output.status.success() { return Err(anyhow::anyhow!( - "`bsk daemon start` exited with status {status:?}" + "`bsk daemon start` exited with status {:?}: {}", + output.status, + String::from_utf8_lossy(&output.stderr).trim() )); } Ok(()) } -fn wait_for_ready(timeout: Duration) -> Result { - let deadline = Instant::now() + timeout; - loop { - if let Some(info) = read_verified()? { - return Ok(info); - } - if Instant::now() >= deadline { - return Err(anyhow::anyhow!( - "no valid daemon.json after {timeout:?}; check `bsk logs`" - )); - } - std::thread::sleep(Duration::from_millis(50)); - } -} - -fn read_verified() -> Result> { - let Some(info) = info::read_valid()? else { - return Ok(None); - }; - match verify_daemon(&info, Duration::from_millis(500)) { - Ok(status) if status.pid == info.pid => Ok(Some(info)), - Ok(_) | Err(_) => Ok(None), - } -} - -fn verify_daemon(info: &DaemonInfo, timeout: Duration) -> Result { - let rt = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .context("build tokio runtime for daemon verification")?; - rt.block_on(async { - let mut client = crate::ipc_client::Client::connect_path(info.sock_path.clone()).await?; - let outcome = client - .call::<_, StatusResult>(Method::SystemStatus, &StatusParams::default(), timeout) - .await?; - outcome.map_err(|err| { - anyhow::anyhow!( - "daemon verification RPC failed: {} ({:?})", - err.message, - err.code - ) - }) - }) -} - fn bsk_executable() -> Result { std::env::current_exe().context("locate current executable for auto-spawn") } diff --git a/crates/bsk-cli/src/cli/error.rs b/crates/bsk-cli/src/cli/error.rs index f8a7b1e..2f4f361 100644 --- a/crates/bsk-cli/src/cli/error.rs +++ b/crates/bsk-cli/src/cli/error.rs @@ -51,7 +51,7 @@ pub enum CliError { /// Local transport / setup failure (e.g. couldn't reach the /// daemon, JSON encode failed). Maps to exit code 2. - #[error(transparent)] + #[error("{0:#}")] Local(#[from] Error), } diff --git a/crates/bsk-cli/src/cli/update.rs b/crates/bsk-cli/src/cli/update.rs index 630c99b..f219aa7 100644 --- a/crates/bsk-cli/src/cli/update.rs +++ b/crates/bsk-cli/src/cli/update.rs @@ -356,10 +356,8 @@ fn install_candidate_with_client( let binary = download_candidate_binary(candidate, client)?; let target = std::env::current_exe().context("locate current bsk executable")?; - let daemon_was_running = restart_daemon && crate::daemon::info::read_valid()?.is_some(); - if daemon_was_running { - crate::daemon::start::run_stop().context("stop bsk daemon before update")?; - } + let daemon_was_running = restart_daemon + && crate::daemon::start::stop_if_running().context("stop bsk daemon before update")?; let restart_args = daemon_was_running.then(StartArgs::default); let action = replace_binary_for_update(&target, &binary, restart_args.as_ref())?; diff --git a/crates/bsk-cli/src/daemon/info.rs b/crates/bsk-cli/src/daemon/info.rs index b18b41c..01bdfd8 100644 --- a/crates/bsk-cli/src/daemon/info.rs +++ b/crates/bsk-cli/src/daemon/info.rs @@ -122,8 +122,8 @@ pub fn read_from_path(path: &Path) -> Result> { } /// Read `daemon.json` and only return it if the recorded pid is alive -/// on the local machine. Stale files (daemon crashed without cleanup) -/// surface as `Ok(None)` so callers can fall through to "auto-spawn". +/// in the caller's PID namespace. This legacy helper does not establish +/// daemon availability or identity; production discovery uses IPC probing. pub fn read_valid() -> Result> { Ok(read()?.filter(|info| lockfile::pid_alive(info.pid))) } diff --git a/crates/bsk-cli/src/daemon/lockfile.rs b/crates/bsk-cli/src/daemon/lockfile.rs index 2a33747..4cea46f 100644 --- a/crates/bsk-cli/src/daemon/lockfile.rs +++ b/crates/bsk-cli/src/daemon/lockfile.rs @@ -64,16 +64,26 @@ pub fn acquire() -> Result { match file.try_lock_exclusive() { Ok(()) => Ok(DaemonLock { file, path }), - Err(_) => Err(anyhow::anyhow!(AlreadyLocked { path })), + Err(err) if err.raw_os_error() == fs2::lock_contended_error().raw_os_error() => { + Err(anyhow::anyhow!(AlreadyLocked { path })) + } + Err(err) => Err(anyhow::Error::new(err).context(format!("lock {}", path.display()))), } } /// Check if a pid is alive on the local machine. /// /// Returns `true` if a process with that pid exists (we don't differentiate -/// our own daemon vs an unrelated process — the pid in `daemon.json` is -/// validated against the held lock by [`info::read_valid`] in M2.4). +/// our own daemon vs an unrelated process). This is local process metadata, +/// not a daemon availability or identity check. pub fn pid_alive(pid: u32) -> bool { + if pid == 0 { + return false; + } + #[cfg(unix)] + if pid > i32::MAX as u32 { + return false; + } #[cfg(unix)] { use nix::errno::Errno; diff --git a/crates/bsk-cli/src/daemon/mod.rs b/crates/bsk-cli/src/daemon/mod.rs index 3196062..f8343e3 100644 --- a/crates/bsk-cli/src/daemon/mod.rs +++ b/crates/bsk-cli/src/daemon/mod.rs @@ -9,6 +9,7 @@ pub mod info; pub mod ipc; pub mod lockfile; pub mod paths; +pub(crate) mod probe; pub mod queue; pub mod session_interrupt; pub mod sessions; diff --git a/crates/bsk-cli/src/daemon/probe.rs b/crates/bsk-cli/src/daemon/probe.rs new file mode 100644 index 0000000..c08c33e --- /dev/null +++ b/crates/bsk-cli/src/daemon/probe.rs @@ -0,0 +1,168 @@ +//! Read-only daemon discovery. IPC proves availability; a PID is meaningful +//! only in the namespace that reported it. Only a missing endpoint permits +//! auto-start; access errors, timeouts and invalid replies must remain errors. + +use std::io::ErrorKind; +use std::time::Duration; + +use anyhow::{Context, Result}; +use bsk_protocol::{Method, StatusParams, StatusResult}; + +use super::info::{self, DaemonInfo}; +use crate::ipc_client::Client; + +pub(crate) const PROBE_TIMEOUT: Duration = Duration::from_millis(500); + +pub(crate) enum Probe { + /// No discovery file, or its endpoint has no listener. This does not + /// authorize removing files: startup/cleanup must still acquire the lock. + Absent(Option), + Ready(Box), +} + +pub(crate) struct VerifiedDaemon { + pub info: DaemonInfo, + pub status: StatusResult, + pub client: Client, +} + +impl VerifiedDaemon { + /// File/RPC PID agreement does not authorize signaling that number in + /// this namespace. Check the kernel's view of the connected peer as well. + pub fn require_local_pid(&self) -> Result { + let peer_pid = self + .client + .peer_pid() + .context("read daemon IPC peer identity")?; + let pid = self.info.pid; + anyhow::ensure!( + pid > 0 && peer_pid == Some(pid), + "cannot verify local daemon process {pid} (IPC peer pid {peer_pid:?}); refusing to signal it; manage the daemon from its host" + ); + Ok(pid) + } +} + +pub(crate) fn probe(timeout: Duration) -> Result { + probe_with_params(timeout, StatusParams::default()) +} + +pub(crate) fn probe_with_params(timeout: Duration, params: StatusParams) -> Result { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .context("build runtime for daemon discovery")? + .block_on(probe_async(timeout, params)) +} + +pub(crate) async fn probe_async(timeout: Duration, params: StatusParams) -> Result { + // Covers connect (including a busy Windows pipe), RPC write/read and the + // single metadata-race retry together, rather than resetting each budget. + tokio::time::timeout(timeout, async { + for attempt in 0..2 { + let Some(info) = info::read().context("read daemon discovery file")? else { + return Ok(Probe::Absent(None)); + }; + let result = query(&info, ¶ms, timeout).await; + let unchanged = super::info::read()?.as_ref() == Some(&info); + let mismatch = result.as_ref().ok().and_then(|reply| reply.as_ref()) + .filter(|(_, status)| status.pid != info.pid) + .map(|(_, status)| status.pid); + if unchanged && mismatch.is_none() { + return match result? { + Some((client, status)) => Ok(Probe::Ready(Box::new(VerifiedDaemon { info, status, client }))), + None => Ok(Probe::Absent(Some(info))), + }; + } + if attempt == 0 { + // A replacement may have bound the endpoint just before + // publishing its metadata. Re-read; never accept a mismatch. + tokio::time::sleep(Duration::from_millis(25)).await; + continue; + } + if let Some(actual_pid) = mismatch { + anyhow::bail!( + "daemon.json pid {} does not match IPC daemon pid {actual_pid}; retry after daemon startup completes", + info.pid + ); + } + anyhow::bail!("daemon discovery changed during verification; retry after daemon startup completes"); + } + unreachable!("the final attempt always returns") + }) + .await + .with_context(|| format!("daemon IPC probe timed out after {timeout:?}; existing daemon may be unresponsive"))? +} + +async fn query( + info: &DaemonInfo, + params: &StatusParams, + timeout: Duration, +) -> Result> { + let mut client = match Client::connect_path(info.sock_path.clone()).await { + Ok(client) => client, + Err(err) if endpoint_absent(&err) => return Ok(None), + Err(err) => return Err(err.context("connect to existing daemon")), + }; + let status = client + .call::<_, StatusResult>(Method::SystemStatus, params, timeout) + .await + .context("query existing daemon status")? + .map_err(|err| { + anyhow::anyhow!( + "daemon verification RPC failed: {} ({:?})", + err.message, + err.code + ) + })?; + Ok(Some((client, status))) +} + +fn endpoint_absent(err: &anyhow::Error) -> bool { + err.downcast_ref::().is_some_and(|err| { + matches!( + err.kind(), + ErrorKind::NotFound | ErrorKind::ConnectionRefused + ) + }) +} + +/// Wait only for a not-yet-published/listening endpoint. A blocked or invalid +/// endpoint is an error, not permission to spawn another daemon. +pub(crate) fn wait_for_ready(timeout: Duration) -> Result> { + let deadline = std::time::Instant::now() + timeout; + loop { + let remaining = deadline.saturating_duration_since(std::time::Instant::now()); + anyhow::ensure!( + !remaining.is_zero(), + "daemon failed to become ready within {timeout:?}; check `bsk logs`" + ); + match probe(remaining.min(PROBE_TIMEOUT))? { + Probe::Ready(daemon) => return Ok(daemon), + Probe::Absent(_) => std::thread::sleep(Duration::from_millis(25).min(remaining)), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn only_missing_or_refused_connections_allow_startup() { + for kind in [ + ErrorKind::PermissionDenied, + ErrorKind::TimedOut, + ErrorKind::WouldBlock, + ErrorKind::ConnectionReset, + ErrorKind::InvalidData, + ] { + let err = anyhow::Error::new(std::io::Error::from(kind)).context("connect IPC"); + assert!(!endpoint_absent(&err), "{kind:?} must not permit startup"); + } + for kind in [ErrorKind::NotFound, ErrorKind::ConnectionRefused] { + let err = anyhow::Error::new(std::io::Error::from(kind)).context("connect IPC"); + assert!(endpoint_absent(&err)); + } + } +} diff --git a/crates/bsk-cli/src/daemon/start.rs b/crates/bsk-cli/src/daemon/start.rs index 4d58457..8f2402a 100644 --- a/crates/bsk-cli/src/daemon/start.rs +++ b/crates/bsk-cli/src/daemon/start.rs @@ -18,13 +18,14 @@ use std::sync::Mutex; use std::time::{Duration, Instant}; use anyhow::{Context, Result}; -use bsk_protocol::{Method, StatusParams, StatusResult}; +use bsk_protocol::StatusResult; use tracing::{debug, info, warn}; use crate::cli::daemon::StartArgs; use crate::daemon::{ browsers::{BROWSER_LIVENESS_TICK, BROWSER_LIVENESS_TIMEOUT, EXTENSION_CONNECT_WAIT}, info as daemon_info, ipc, lockfile, paths, + probe::{self, PROBE_TIMEOUT, Probe}, sessions::{StopSessionError, forget_session, stop_session}, state::{DaemonState, PROTOCOL_VERSION}, ws, @@ -119,7 +120,8 @@ pub fn run_start(args: StartArgs) -> Result<()> { return run_foreground(cfg); } - if let Some(status) = verified_existing_daemon()? { + if let Probe::Ready(daemon) = probe::probe(PROBE_TIMEOUT)? { + let status = daemon.status; validate_existing_start(&args, &status)?; info!( pid = status.pid, @@ -131,74 +133,106 @@ pub fn run_start(args: StartArgs) -> Result<()> { // Parent: spawn ourselves detached and wait for ready. spawn_detached(&args)?; - wait_for_ready(Duration::from_secs(3), args.port.filter(|port| *port != 0))?; + let daemon = probe::wait_for_ready(Duration::from_secs(3))?; + if let Some(port) = args.port.filter(|port| *port != 0) { + anyhow::ensure!( + daemon.status.ws_port == port, + "daemon started on ws port {}, expected {port}", + daemon.status.ws_port + ); + } Ok(()) } /// `bsk daemon stop` entrypoint. pub fn run_stop() -> Result<()> { - let info = match daemon_info::read()? { - Some(info) => info, - None => { - info!("no daemon.json present — nothing to stop"); - return Ok(()); + stop_if_running().map(|_| ()) +} + +/// Stop with the same checks for explicit management and CLI self-update. +/// The return value tells the updater whether it should restart a daemon. +pub(crate) fn stop_if_running() -> Result { + let daemon = match probe::probe(Duration::from_secs(2))? { + Probe::Ready(daemon) => daemon, + Probe::Absent(None) => return Ok(false), + Probe::Absent(Some(expected)) => { + // A missing PID cannot authorize cleanup across namespaces. The + // lock excludes a live daemon, including one still starting up. + let _lock = + lockfile::acquire().context("verify daemon is not running before cleanup")?; + anyhow::ensure!( + daemon_info::read()?.as_ref() == Some(&expected), + "daemon discovery changed during stop; retry" + ); + anyhow::ensure!( + !lockfile::pid_alive(expected.pid), + "could not verify daemon identity for pid {}; refusing to stop", + expected.pid + ); + match probe::probe(PROBE_TIMEOUT)? { + Probe::Absent(Some(current)) if current == expected => daemon_info::remove()?, + _ => anyhow::bail!("daemon became reachable during cleanup; retry stop"), + } + return Ok(false); } }; - - if !lockfile::pid_alive(info.pid) { - info!( - pid = info.pid, - "daemon.json points to a dead pid — cleaning up" - ); - let _ = daemon_info::remove(); - return Ok(()); + let pid = daemon.require_local_pid()?; + send_term(pid)?; + if wait_for_stopped(&daemon.info, Duration::from_secs(5))? { + info!(pid, "daemon stopped"); + return Ok(true); } - match confirm_daemon(&info, Duration::from_secs(2)) { - Ok(status) if status.pid == info.pid => {} - Ok(status) => { - warn!( - expected_pid = info.pid, - actual_pid = status.pid, - "daemon.json does not match IPC daemon; refusing to stop" - ); - return Err(anyhow::anyhow!( - "daemon.json pid {} does not match IPC daemon pid {}; refusing to stop", - info.pid, - status.pid - )); - } - Err(err) => { - warn!( - pid = info.pid, - ?err, - "daemon.json pid is alive but IPC validation failed; refusing to stop" - ); - return Err(anyhow::anyhow!( - "could not verify daemon identity for pid {}; refusing to stop: {err:#}", - info.pid - )); + // Revalidate before escalation: a PID may have been reused, or a + // replacement daemon may already own the endpoint. + match probe::probe(Duration::from_secs(2))? { + Probe::Ready(current) if current.info == daemon.info => { + let pid = current.require_local_pid()?; + warn!(pid, "daemon did not exit within 5s; sending KILL"); + send_kill(pid)?; } + _ => anyhow::bail!("daemon identity changed while stopping; refusing to send KILL"), } + anyhow::ensure!( + wait_for_stopped(&daemon.info, Duration::from_secs(5))?, + "daemon did not release its lock after KILL" + ); + Ok(true) +} - send_term(info.pid)?; - let deadline = Instant::now() + Duration::from_secs(5); - while Instant::now() < deadline { - if !lockfile::pid_alive(info.pid) { - let _ = daemon_info::remove(); - info!(pid = info.pid, "daemon stopped"); - return Ok(()); +fn wait_for_stopped(expected: &daemon_info::DaemonInfo, timeout: Duration) -> Result { + let deadline = Instant::now() + timeout; + loop { + match lockfile::acquire() { + Ok(_lock) => { + // Normal shutdown removes its own metadata. A forced exit + // leaves it behind; only remove that exact instance's record. + if daemon_info::read()?.as_ref() == Some(expected) { + match probe::probe(PROBE_TIMEOUT)? { + Probe::Absent(Some(info)) if &info == expected => daemon_info::remove()?, + _ => anyhow::bail!( + "daemon endpoint still active after lock release; refusing cleanup" + ), + } + } + return Ok(true); + } + Err(err) if err.is::() => { + if daemon_info::read()? + .as_ref() + .is_some_and(|info| info != expected) + { + // A replacement acquired the lock; never remove its files. + return Ok(true); + } + } + Err(err) => return Err(err.context("check daemon shutdown lock")), + } + if Instant::now() >= deadline { + return Ok(false); } std::thread::sleep(Duration::from_millis(50)); } - - warn!( - pid = info.pid, - "daemon did not exit within 5s; sending KILL" - ); - let _ = send_kill(info.pid); - let _ = daemon_info::remove(); - Ok(()) } /// Run the daemon in the foreground of the current process: acquire @@ -963,16 +997,6 @@ fn format_duration(d: Duration) -> String { } } -fn verified_existing_daemon() -> Result> { - let Some(info) = daemon_info::read_valid()? else { - return Ok(None); - }; - match confirm_daemon(&info, Duration::from_millis(500)) { - Ok(status) if status.pid == info.pid => Ok(Some(status)), - Ok(_) | Err(_) => Ok(None), - } -} - fn validate_existing_start(args: &StartArgs, status: &StatusResult) -> Result<()> { if let Some(port) = args.port && status.ws_port != port @@ -990,67 +1014,6 @@ fn validate_existing_start(args: &StartArgs, status: &StatusResult) -> Result<() Ok(()) } -fn wait_for_ready(timeout: Duration, expected_port: Option) -> Result<()> { - let info_path = paths::info_path()?; - let deadline = Instant::now() + timeout; - let _ = info_path; - loop { - if let Some(info) = daemon_info::read_valid()? { - match confirm_daemon(&info, Duration::from_millis(500)) { - Ok(status) if status.pid == info.pid => { - if let Some(port) = expected_port - && status.ws_port != port - { - return Err(anyhow::anyhow!( - "daemon started on ws port {}, expected {port}", - status.ws_port - )); - } - tracing::debug!(?info, "daemon ready"); - return Ok(()); - } - Ok(status) => { - tracing::debug!( - expected_pid = info.pid, - actual_pid = status.pid, - "daemon.json did not match IPC status yet" - ); - } - Err(err) => { - tracing::debug!(?err, "daemon IPC not ready yet"); - } - } - } - if Instant::now() >= deadline { - return Err(anyhow::anyhow!( - "daemon failed to start within {:?} (no valid daemon.json)", - timeout - )); - } - std::thread::sleep(Duration::from_millis(50)); - } -} - -fn confirm_daemon(info: &daemon_info::DaemonInfo, timeout: Duration) -> Result { - let rt = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .context("build tokio runtime for daemon verification")?; - rt.block_on(async { - let mut client = crate::ipc_client::Client::connect_path(info.sock_path.clone()).await?; - let outcome = client - .call::<_, StatusResult>(Method::SystemStatus, &StatusParams::default(), timeout) - .await?; - outcome.map_err(|err| { - anyhow::anyhow!( - "daemon verification RPC failed: {} ({:?})", - err.message, - err.code - ) - }) - }) -} - #[cfg(unix)] fn send_term(pid: u32) -> Result<()> { use nix::sys::signal::{Signal, kill}; diff --git a/crates/bsk-cli/src/ipc_client.rs b/crates/bsk-cli/src/ipc_client.rs index 0f15ae9..04e0d4e 100644 --- a/crates/bsk-cli/src/ipc_client.rs +++ b/crates/bsk-cli/src/ipc_client.rs @@ -15,13 +15,12 @@ mod platform { use std::time::Duration; use anyhow::{Context, Result}; - use bsk_protocol::{Frame, Method, RequestFrame, ResponseBody, StatusParams, StatusResult}; + use bsk_protocol::{Frame, Method, RequestFrame, ResponseBody}; use serde::{Serialize, de::DeserializeOwned}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; use tokio::time::timeout; - use crate::daemon::info as daemon_info; use crate::ipc_client::{RpcOutcome, random_id}; /// Connected IPC client. Holds an open UDS connection plus a buffered @@ -33,35 +32,13 @@ mod platform { } impl Client { - /// Connect to the daemon described in `daemon.json`. Fails if no - /// info file exists, the pid is dead, or the UDS is unreachable. - pub async fn connect() -> Result { - let info = daemon_info::read_valid() - .context("read daemon.json")? - .ok_or_else(|| anyhow::anyhow!("no live daemon (daemon.json missing or stale)"))?; - let mut client = Self::connect_path(info.sock_path).await?; - let status = client - .call::<_, StatusResult>( - Method::SystemStatus, - &StatusParams::default(), - Duration::from_secs(2), - ) - .await? - .map_err(|err| { - anyhow::anyhow!( - "daemon verification RPC failed: {} ({:?})", - err.message, - err.code - ) - })?; - if status.pid != info.pid { - return Err(anyhow::anyhow!( - "daemon.json pid {} does not match IPC daemon pid {}", - info.pid, - status.pid - )); - } - Ok(client) + /// PID in the caller's namespace, supplied by the kernel, not JSON. + pub(crate) fn peer_pid(&self) -> std::io::Result> { + let credentials = self.stream.get_ref().as_ref().peer_cred()?; + Ok(credentials + .pid() + .and_then(|pid| u32::try_from(pid).ok()) + .filter(|pid| *pid > 0)) } /// Connect directly to a UDS path (used by tests + auto-spawn @@ -161,50 +138,28 @@ mod platform { use std::time::Duration; use anyhow::{Context, Result}; - use bsk_protocol::{Frame, Method, RequestFrame, ResponseBody, StatusParams, StatusResult}; + use bsk_protocol::{Frame, Method, RequestFrame, ResponseBody}; use serde::{Serialize, de::DeserializeOwned}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::windows::named_pipe::{ClientOptions, NamedPipeClient}; use tokio::time::{sleep, timeout}; use windows_sys::Win32::Foundation::ERROR_PIPE_BUSY; - use crate::daemon::info as daemon_info; use crate::ipc_client::{RpcOutcome, random_id}; pub struct Client { stream: BufReader>, write: tokio::io::WriteHalf, pipe_name: PathBuf, + peer_pid: std::io::Result>, } impl Client { - pub async fn connect() -> Result { - let info = daemon_info::read_valid() - .context("read daemon.json")? - .ok_or_else(|| anyhow::anyhow!("no live daemon (daemon.json missing or stale)"))?; - let mut client = Self::connect_path(info.sock_path).await?; - let status = client - .call::<_, StatusResult>( - Method::SystemStatus, - &StatusParams::default(), - Duration::from_secs(2), - ) - .await? - .map_err(|err| { - anyhow::anyhow!( - "daemon verification RPC failed: {} ({:?})", - err.message, - err.code - ) - })?; - if status.pid != info.pid { - return Err(anyhow::anyhow!( - "daemon.json pid {} does not match IPC daemon pid {}", - info.pid, - status.pid - )); - } - Ok(client) + pub(crate) fn peer_pid(&self) -> std::io::Result> { + self.peer_pid + .as_ref() + .copied() + .map_err(|err| std::io::Error::new(err.kind(), err.to_string())) } /// Total budget for retrying `ERROR_PIPE_BUSY` while connecting @@ -238,11 +193,23 @@ mod platform { Self::CONNECT_BUSY_TIMEOUT ) })??; + // Fetch while the pipe handle is directly available; an identity + // lookup failure affects management only, not normal IPC use. + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::System::Pipes::GetNamedPipeServerProcessId; + let mut pid = 0; + let peer_pid = + if unsafe { GetNamedPipeServerProcessId(client.as_raw_handle(), &mut pid) } != 0 { + Ok((pid > 0).then_some(pid)) + } else { + Err(std::io::Error::last_os_error()) + }; let (read, write) = tokio::io::split(client); Ok(Self { stream: BufReader::new(read), write, pipe_name, + peer_pid, }) } @@ -317,6 +284,24 @@ mod platform { pub use platform::Client; +impl Client { + /// Discover and verify the existing daemon without starting one. + pub async fn connect() -> anyhow::Result { + use crate::daemon::probe::{Probe, probe_async}; + match probe_async( + std::time::Duration::from_secs(2), + bsk_protocol::StatusParams::default(), + ) + .await? + { + Probe::Ready(daemon) => Ok(daemon.client), + Probe::Absent(_) => { + anyhow::bail!("no listening daemon (daemon.json or IPC endpoint missing)") + } + } + } +} + /// Result of a typed RPC: either the deserialised happy-path result or /// the structured `RpcError` returned by the daemon. pub type RpcOutcome = std::result::Result; diff --git a/crates/bsk-cli/tests/daemon_discovery.rs b/crates/bsk-cli/tests/daemon_discovery.rs new file mode 100644 index 0000000..72bae99 --- /dev/null +++ b/crates/bsk-cli/tests/daemon_discovery.rs @@ -0,0 +1,469 @@ +//! Regression coverage for discovery without local PID visibility and for +//! conservative management of missing, blocked and inconsistent endpoints. +#![cfg(unix)] + +use std::io::{BufRead, BufReader, Write}; +use std::os::unix::fs::MetadataExt; +use std::os::unix::net::UnixListener; +use std::path::Path; +use std::process::{Child, Command, Output, Stdio}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::time::{Duration, Instant}; + +use bsk::daemon::info::{DaemonInfo, write_to_path}; +use bsk::daemon::lockfile::pid_alive; +use bsk_protocol::{Frame, ResponseBody, ResponseFrame}; +use fs2::FileExt; +use tempfile::TempDir; + +const FOREIGN_PID: u32 = 0x3fff_fffe; + +fn command(home: &Path, args: &[&str]) -> Command { + let mut cmd = Command::new(env!("CARGO_BIN_EXE_bsk")); + cmd.args(args) + .env("BSK_HOME", home) + .env("BSK_AUTO_UPDATE", "0") + .env("BSK_BROWSER_WAIT_MS", "0") + .env("BSK_DOCTOR_BROWSER_WAIT_MS", "0") + .env("RUST_LOG", "warn"); + cmd +} + +fn run(home: &Path, args: &[&str]) -> Output { + command(home, args).output().unwrap() +} + +fn success(out: &Output) { + assert!( + out.status.success(), + "stdout={} stderr={}", + String::from_utf8_lossy(&out.stdout), + String::from_utf8_lossy(&out.stderr) + ); +} + +struct MockDaemon { + temp: TempDir, + stop: Arc, + thread: Option>, +} + +impl MockDaemon { + fn new( + pid: u32, + reply: impl Fn(usize, &DaemonInfo) -> Option + Send + Sync + 'static, + ) -> Self { + let temp = TempDir::new().unwrap(); + let sock = temp.path().join("daemon.sock"); + let listener = UnixListener::bind(&sock).unwrap(); + listener.set_nonblocking(true).unwrap(); + let info = DaemonInfo::now(pid, sock, 12345, env!("CARGO_PKG_VERSION")); + write_to_path(&info, &temp.path().join("daemon.json")).unwrap(); + let stop = Arc::new(AtomicBool::new(false)); + let done = stop.clone(); + let served_info = info.clone(); + let reply = Arc::new(reply); + let requests = Arc::new(AtomicUsize::new(0)); + let thread = std::thread::spawn(move || { + let mut clients = Vec::new(); + while !done.load(Ordering::Relaxed) { + match listener.accept() { + Ok((stream, _)) => { + let reply = reply.clone(); + let info = served_info.clone(); + let done = done.clone(); + let requests = requests.clone(); + clients.push(std::thread::spawn(move || { + stream + .set_read_timeout(Some(Duration::from_millis(50))) + .unwrap(); + let mut reader = BufReader::new(stream); + while !done.load(Ordering::Relaxed) { + let mut line = String::new(); + match reader.read_line(&mut line) { + Ok(0) => break, + Ok(_) => {} + Err(err) + if matches!( + err.kind(), + std::io::ErrorKind::WouldBlock + | std::io::ErrorKind::TimedOut + ) => + { + continue; + } + Err(_) => break, + } + let Frame::Request(request) = serde_json::from_str(&line).unwrap() + else { + panic!("expected request") + }; + let n = requests.fetch_add(1, Ordering::SeqCst); + if let Some(body) = reply(n, &info) { + let frame = Frame::Response(ResponseFrame { + id: request.id, + body, + }); + if writeln!( + reader.get_mut(), + "{}", + serde_json::to_string(&frame).unwrap() + ) + .is_err() + { + break; + } + } + } + })); + } + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => { + std::thread::sleep(Duration::from_millis(5)) + } + Err(err) => panic!("accept: {err}"), + } + } + for client in clients { + client.join().unwrap(); + } + }); + Self { + temp, + stop, + thread: Some(thread), + } + } + + fn home(&self) -> &Path { + self.temp.path() + } + fn metadata(&self) -> Vec { + std::fs::read(self.home().join("daemon.json")).unwrap() + } +} + +impl Drop for MockDaemon { + fn drop(&mut self) { + self.stop.store(true, Ordering::Relaxed); + self.thread.take().unwrap().join().unwrap(); + } +} + +fn status(info: &DaemonInfo) -> ResponseBody { + ResponseBody::Ok(serde_json::json!({ + "pid": info.pid, "daemon_version": info.version, "protocol_version": "1.1", + "uptime_secs": 1, "ws_port": info.ws_port, "sock_path": info.sock_path, + "browsers": [], "sessions": [] + })) +} + +#[test] +fn discovery_accepts_ipc_without_local_pid_but_management_refuses_it() { + assert!(!pid_alive(FOREIGN_PID)); + let daemon = MockDaemon::new(FOREIGN_PID, |_, info| Some(status(info))); + let original = daemon.metadata(); + success(&run(daemon.home(), &["--json", "status"])); + success(&run(daemon.home(), &["daemon", "start"])); + let doctor = run(daemon.home(), &["--json", "doctor"]); + let checks: Vec = serde_json::from_slice(&doctor.stdout).unwrap(); + assert_eq!( + checks + .iter() + .find(|c| c["name"] == "daemon running") + .unwrap()["ok"], + true + ); + for args in [["daemon", "stop"], ["daemon", "restart"]] { + let out = run(daemon.home(), &args); + assert!(!out.status.success()); + assert!( + String::from_utf8_lossy(&out.stderr).contains("cannot verify local daemon process") + ); + assert_eq!(daemon.metadata(), original); + } + assert!( + !daemon.home().join("daemon.lock").exists(), + "discovery must not attempt startup or cleanup" + ); +} + +#[test] +fn file_and_rpc_agreement_cannot_authorize_signaling_an_unrelated_pid() { + let mut decoy = ManagedChild(Command::new("sleep").arg("30").spawn().unwrap()); + let daemon = MockDaemon::new(decoy.0.id(), |_, info| Some(status(info))); + let original = daemon.metadata(); + let out = run(daemon.home(), &["daemon", "stop"]); + assert!(!out.status.success()); + assert!(decoy.0.try_wait().unwrap().is_none()); + assert_eq!(daemon.metadata(), original); +} + +#[test] +fn unresponsive_endpoint_is_bounded_and_does_not_spawn_or_clean_up() { + let daemon = MockDaemon::new(FOREIGN_PID, |_, _| None); + let original = daemon.metadata(); + let start = Instant::now(); + let out = run(daemon.home(), &["--json", "status"]); + assert!(!out.status.success()); + assert!(start.elapsed() < Duration::from_secs(3)); + assert!(String::from_utf8_lossy(&out.stdout).contains("timed out")); + assert_eq!(daemon.metadata(), original); + assert!(!daemon.home().join("daemon.lock").exists()); +} + +#[test] +fn invalid_and_mismatched_responses_do_not_trigger_startup() { + for mismatch in [false, true] { + let daemon = MockDaemon::new(FOREIGN_PID, move |_, info| { + let mut reply = if mismatch { + status(info) + } else { + ResponseBody::Ok(serde_json::json!({"unexpected": true})) + }; + if mismatch && let ResponseBody::Ok(ref mut value) = reply { + value["pid"] = serde_json::json!(FOREIGN_PID - 1); + } + Some(reply) + }); + let original = daemon.metadata(); + let out = run(daemon.home(), &["daemon", "start"]); + assert!(!out.status.success()); + assert_eq!(daemon.metadata(), original); + assert!(!daemon.home().join("daemon.lock").exists()); + } +} + +#[test] +fn discovery_rereads_metadata_when_an_instance_changes_during_probe() { + let daemon = MockDaemon::new(FOREIGN_PID, |_, info| { + let mut replacement = info.clone(); + replacement.pid -= 1; + write_to_path( + &replacement, + &info.sock_path.parent().unwrap().join("daemon.json"), + ) + .unwrap(); + Some(status(&replacement)) + }); + let out = run(daemon.home(), &["--json", "status"]); + success(&out); + let value: serde_json::Value = serde_json::from_slice(&out.stdout).unwrap(); + assert_eq!(value["pid"], FOREIGN_PID - 1); + assert!(!daemon.home().join("daemon.lock").exists()); +} + +#[test] +fn stopping_without_discovery_does_not_create_runtime_files() { + let temp = TempDir::new().unwrap(); + let home = temp.path().join("absent"); + success(&run(&home, &["daemon", "stop"])); + assert!(!home.exists()); +} + +#[test] +fn held_lock_protects_metadata_even_if_pid_and_endpoint_are_absent() { + let temp = TempDir::new().unwrap(); + let info = DaemonInfo::now(FOREIGN_PID, temp.path().join("missing.sock"), 0, "0.2.1"); + let path = temp.path().join("daemon.json"); + write_to_path(&info, &path).unwrap(); + let original = std::fs::read(&path).unwrap(); + let lock = std::fs::File::create(temp.path().join("daemon.lock")).unwrap(); + lock.try_lock_exclusive().unwrap(); + let out = run(temp.path(), &["daemon", "stop"]); + assert!(!out.status.success()); + assert_eq!(std::fs::read(&path).unwrap(), original); + drop(lock); + success(&run(temp.path(), &["daemon", "stop"])); + assert!(!path.exists()); + assert!(temp.path().join("daemon.lock").exists()); +} + +#[test] +fn ipc_permission_denial_preserves_metadata_and_does_not_start_a_daemon() { + if unsafe { libc::geteuid() } == 0 { + eprintln!("skipping permission denial check for root"); + return; + } + use std::os::unix::fs::PermissionsExt; + let temp = TempDir::new().unwrap(); + let blocked = temp.path().join("blocked"); + std::fs::create_dir(&blocked).unwrap(); + let sock = blocked.join("daemon.sock"); + let _listener = UnixListener::bind(&sock).unwrap(); + let info = DaemonInfo::now(FOREIGN_PID, sock, 0, "0.2.1"); + let path = temp.path().join("daemon.json"); + write_to_path(&info, &path).unwrap(); + let original = std::fs::read(&path).unwrap(); + std::fs::set_permissions(&blocked, std::fs::Permissions::from_mode(0o0)).unwrap(); + let out = run(temp.path(), &["--json", "status"]); + std::fs::set_permissions(&blocked, std::fs::Permissions::from_mode(0o700)).unwrap(); + assert!(!out.status.success()); + assert!(String::from_utf8_lossy(&out.stdout).contains("Permission denied")); + assert_eq!(std::fs::read(path).unwrap(), original); + assert!(!temp.path().join("daemon.lock").exists()); +} + +#[test] +fn direct_client_uses_the_same_discovery_rules() { + // Run the client in a child so BSK_HOME is fixed before any test threads + // start, without mutating the parallel test runner's environment. + if std::env::var_os("BSK_DISCOVERY_TEST_CHILD").is_none() { + let daemon = MockDaemon::new(FOREIGN_PID, |_, info| Some(status(info))); + let out = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "direct_client_uses_the_same_discovery_rules"]) + .env("BSK_HOME", daemon.home()) + .env("BSK_DISCOVERY_TEST_CHILD", "1") + .output() + .unwrap(); + success(&out); + return; + } + tokio::runtime::Runtime::new().unwrap().block_on(async { + let mut client = bsk::ipc_client::Client::connect().await.unwrap(); + let result = client + .call::<_, bsk_protocol::StatusResult>( + bsk_protocol::Method::SystemStatus, + &bsk_protocol::StatusParams::default(), + Duration::from_secs(1), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(result.pid, FOREIGN_PID); + }); +} + +#[cfg(target_os = "linux")] +#[test] +fn host_daemon_is_usable_but_not_signalable_from_a_child_pid_namespace() { + // Some CI hosts disable unprivileged user namespaces. A skip is explicit; + // deterministic IPC/PID regressions above still run on every Unix host. + let available = Command::new("unshare") + .args(["--user", "--map-root-user", "--pid", "--fork", "true"]) + .output(); + if !available.as_ref().is_ok_and(|out| out.status.success()) { + eprintln!( + "skipping real PID namespace test: unprivileged unshare unavailable: {available:?}" + ); + return; + } + let temp = TempDir::new().unwrap(); + let mut daemon = ManagedChild( + command( + temp.path(), + &["daemon", "start", "--foreground", "--port", "0"], + ) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(), + ); + let original = wait_ready(temp.path(), &mut daemon); + let inside = |args: &[&str]| { + Command::new("unshare") + .args(["--user", "--map-root-user", "--pid", "--fork"]) + .arg(env!("CARGO_BIN_EXE_bsk")) + .args(args) + .env("BSK_HOME", temp.path()) + .env("BSK_AUTO_UPDATE", "0") + .env("BSK_BROWSER_WAIT_MS", "0") + .output() + .unwrap() + }; + success(&inside(&["--json", "status"])); + success(&inside(&["daemon", "start"])); + let out = inside(&["daemon", "stop"]); + assert!(!out.status.success()); + assert!(String::from_utf8_lossy(&out.stderr).contains("cannot verify local daemon process")); + assert!(daemon.0.try_wait().unwrap().is_none()); + let after: DaemonInfo = + serde_json::from_slice(&std::fs::read(temp.path().join("daemon.json")).unwrap()).unwrap(); + assert_eq!(after, original); + success(&run(temp.path(), &["daemon", "stop"])); +} + +struct ManagedChild(Child); +impl Drop for ManagedChild { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn wait_ready(home: &Path, child: &mut ManagedChild) -> DaemonInfo { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + assert!( + child.0.try_wait().unwrap().is_none(), + "daemon exited before ready" + ); + if let Ok(bytes) = std::fs::read(home.join("daemon.json")) + && let Ok(info) = serde_json::from_slice(&bytes) + { + return info; + } + assert!(Instant::now() < deadline, "daemon not ready"); + std::thread::sleep(Duration::from_millis(25)); + } +} + +#[test] +fn crash_recovery_reuses_lock_file_and_concurrent_starts_share_one_daemon() { + let temp = TempDir::new().unwrap(); + let mut first = ManagedChild( + command( + temp.path(), + &["daemon", "start", "--foreground", "--port", "0"], + ) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(), + ); + let old = wait_ready(temp.path(), &mut first); + let lock_inode = std::fs::metadata(temp.path().join("daemon.lock")) + .unwrap() + .ino(); + first.0.kill().unwrap(); + first.0.wait().unwrap(); + assert!( + old.sock_path.exists(), + "forced exit should leave its socket" + ); + let port = old.ws_port.to_string(); + let mut starts: Vec<_> = (0..4) + .map(|_| { + command(temp.path(), &["daemon", "start", "--port", &port]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap() + }) + .collect(); + for start in starts.drain(..) { + success(&start.wait_with_output().unwrap()); + } + let current: DaemonInfo = + serde_json::from_slice(&std::fs::read(temp.path().join("daemon.json")).unwrap()).unwrap(); + assert_ne!(current.pid, old.pid); + assert_eq!( + std::fs::metadata(temp.path().join("daemon.lock")) + .unwrap() + .ino(), + lock_inode + ); + success(&run(temp.path(), &["daemon", "restart", "--port", &port])); + let restarted: DaemonInfo = + serde_json::from_slice(&std::fs::read(temp.path().join("daemon.json")).unwrap()).unwrap(); + assert_ne!(restarted.pid, current.pid); + assert_eq!(restarted.ws_port, old.ws_port); + assert_eq!( + std::fs::metadata(temp.path().join("daemon.lock")) + .unwrap() + .ino(), + lock_inode + ); + success(&run(temp.path(), &["daemon", "stop"])); + assert!(!temp.path().join("daemon.json").exists()); +} diff --git a/crates/bsk-cli/tests/status_cmd.rs b/crates/bsk-cli/tests/status_cmd.rs index 18513e2..b28ca5d 100644 --- a/crates/bsk-cli/tests/status_cmd.rs +++ b/crates/bsk-cli/tests/status_cmd.rs @@ -40,7 +40,19 @@ fn bsk_status_json_returns_structured_payload() { let home = tmp.path().join("bsk"); std::fs::create_dir_all(&home).unwrap(); - // Auto-spawn via `bsk status` — should bring up the daemon. + // Isolate the status rendering check from any user daemon on port 52800. + let start = Command::new(bsk_bin()) + .args(["daemon", "start", "--port", "0", "--daemon-idle", "60s"]) + .env("BSK_HOME", &home) + .env("BSK_AUTO_UPDATE", "0") + .output() + .unwrap(); + assert!( + start.status.success(), + "{}", + String::from_utf8_lossy(&start.stderr) + ); + let out = Command::new(bsk_bin()) .args(["--json", "status"]) .env("BSK_HOME", &home) @@ -197,6 +209,12 @@ fn bsk_doctor_does_not_treat_live_non_daemon_pid_as_running() { ) .unwrap(); + // Keep the recorded instance unavailable rather than allowing doctor to + // recover it by starting a fresh daemon. This must not depend on whether + // the user's default WS port happens to be occupied. + let lock = std::fs::File::create(home.join("daemon.lock")).unwrap(); + fs2::FileExt::try_lock_exclusive(&lock).unwrap(); + let out = Command::new(bsk_bin()) .args(["--json", "doctor"]) .env("BSK_HOME", &home) From 55a06d7592ba413d3306bd3d858d3bd6f722d363 Mon Sep 17 00:00:00 2001 From: drakezhang Date: Thu, 10 Sep 2026 18:47:53 +0800 Subject: [PATCH 2/2] fix(daemon): preserve readiness retries and reject stop handoffs Retry typed timeout and discovery-race errors only in the post-spawn readiness wait, bounded by the caller deadline. Keep initial discovery fail-closed and retain the last transient error when readiness times out. Reject changed instance metadata while stopping, including when another process holds the lock. Propagate that error through the existing update and restart paths instead of treating a replacement as a successful stop. Add regression coverage for delayed IPC readiness, deadline expiry, metadata races, terminal responses and replacement preservation. Restore empty-home auto-spawn coverage through status and exercise actual instance reuse. Require a free default port in CI; only local runs may skip when that port is occupied by a developer's daemon. Validation: serial workspace tests, cargo fmt, strict workspace Clippy, and cargo build passed. The local default-port auto-spawn case was skipped because port 52800 was occupied; both auto-spawn tests also passed in a temporary source copy with only the CLI default port changed to a free port. Updated Windows probe modules passed isolated cross-compilation. --- crates/bsk-cli/src/daemon/mod.rs | 3 + crates/bsk-cli/src/daemon/probe.rs | 103 +++++++++-- crates/bsk-cli/src/daemon/probe_tests.rs | 206 ++++++++++++++++++++++ crates/bsk-cli/src/daemon/start.rs | 46 ++++- crates/bsk-cli/src/daemon/test_support.rs | 24 +++ crates/bsk-cli/tests/auto_spawn.rs | 108 +++++++++--- 6 files changed, 442 insertions(+), 48 deletions(-) create mode 100644 crates/bsk-cli/src/daemon/probe_tests.rs create mode 100644 crates/bsk-cli/src/daemon/test_support.rs diff --git a/crates/bsk-cli/src/daemon/mod.rs b/crates/bsk-cli/src/daemon/mod.rs index f8343e3..761d2ce 100644 --- a/crates/bsk-cli/src/daemon/mod.rs +++ b/crates/bsk-cli/src/daemon/mod.rs @@ -17,6 +17,9 @@ pub mod start; pub mod state; pub mod ws; +#[cfg(test)] +mod test_support; + pub use start::{DaemonConfig, run_foreground}; pub use state::{DaemonHandle, DaemonState}; diff --git a/crates/bsk-cli/src/daemon/probe.rs b/crates/bsk-cli/src/daemon/probe.rs index c08c33e..75cf079 100644 --- a/crates/bsk-cli/src/daemon/probe.rs +++ b/crates/bsk-cli/src/daemon/probe.rs @@ -26,6 +26,16 @@ pub(crate) struct VerifiedDaemon { pub client: Client, } +#[derive(Debug, thiserror::Error)] +enum DiscoveryRace { + #[error( + "daemon.json pid {expected} does not match IPC daemon pid {actual}; retry after daemon startup completes" + )] + PidMismatch { expected: u32, actual: u32 }, + #[error("daemon discovery changed during verification; retry after daemon startup completes")] + Changed, +} + impl VerifiedDaemon { /// File/RPC PID agreement does not authorize signaling that number in /// this namespace. Check the kernel's view of the connected peer as well. @@ -65,12 +75,19 @@ pub(crate) async fn probe_async(timeout: Duration, params: StatusParams) -> Resu }; let result = query(&info, ¶ms, timeout).await; let unchanged = super::info::read()?.as_ref() == Some(&info); - let mismatch = result.as_ref().ok().and_then(|reply| reply.as_ref()) + let mismatch = result + .as_ref() + .ok() + .and_then(|reply| reply.as_ref()) .filter(|(_, status)| status.pid != info.pid) .map(|(_, status)| status.pid); if unchanged && mismatch.is_none() { return match result? { - Some((client, status)) => Ok(Probe::Ready(Box::new(VerifiedDaemon { info, status, client }))), + Some((client, status)) => Ok(Probe::Ready(Box::new(VerifiedDaemon { + info, + status, + client, + }))), None => Ok(Probe::Absent(Some(info))), }; } @@ -81,17 +98,20 @@ pub(crate) async fn probe_async(timeout: Duration, params: StatusParams) -> Resu continue; } if let Some(actual_pid) = mismatch { - anyhow::bail!( - "daemon.json pid {} does not match IPC daemon pid {actual_pid}; retry after daemon startup completes", - info.pid - ); + return Err(DiscoveryRace::PidMismatch { + expected: info.pid, + actual: actual_pid, + } + .into()); } - anyhow::bail!("daemon discovery changed during verification; retry after daemon startup completes"); + return Err(DiscoveryRace::Changed.into()); } unreachable!("the final attempt always returns") }) .await - .with_context(|| format!("daemon IPC probe timed out after {timeout:?}; existing daemon may be unresponsive"))? + .with_context(|| { + format!("daemon IPC probe timed out after {timeout:?}; existing daemon may be unresponsive") + })? } async fn query( @@ -127,23 +147,43 @@ fn endpoint_absent(err: &anyhow::Error) -> bool { }) } -/// Wait only for a not-yet-published/listening endpoint. A blocked or invalid -/// endpoint is an error, not permission to spawn another daemon. +/// After spawning, tolerate timeouts and discovery races until the caller's +/// deadline. This only probes: it never authorizes another spawn. Permission +/// and protocol errors remain terminal, just as they are during discovery. pub(crate) fn wait_for_ready(timeout: Duration) -> Result> { let deadline = std::time::Instant::now() + timeout; + let mut last_error = None; loop { let remaining = deadline.saturating_duration_since(std::time::Instant::now()); - anyhow::ensure!( - !remaining.is_zero(), - "daemon failed to become ready within {timeout:?}; check `bsk logs`" - ); - match probe(remaining.min(PROBE_TIMEOUT))? { - Probe::Ready(daemon) => return Ok(daemon), - Probe::Absent(_) => std::thread::sleep(Duration::from_millis(25).min(remaining)), + if remaining.is_zero() { + let message = + format!("daemon failed to become ready within {timeout:?}; check `bsk logs`"); + return Err(match last_error { + Some(err) => anyhow::Error::context(err, message), + None => anyhow::anyhow!(message), + }); } + match probe(remaining.min(PROBE_TIMEOUT)) { + Ok(Probe::Ready(daemon)) => return Ok(daemon), + Ok(Probe::Absent(_)) => {} + Err(err) if retryable_during_startup(&err) => last_error = Some(err), + Err(err) => return Err(err), + } + std::thread::sleep( + Duration::from_millis(25) + .min(deadline.saturating_duration_since(std::time::Instant::now())), + ); } } +fn retryable_during_startup(err: &anyhow::Error) -> bool { + err.is::() + || err.is::() + || err + .downcast_ref::() + .is_some_and(|err| err.kind() == ErrorKind::TimedOut) +} + #[cfg(test)] mod tests { use super::*; @@ -165,4 +205,33 @@ mod tests { assert!(endpoint_absent(&err)); } } + + #[test] + fn startup_retries_only_timeouts_and_discovery_races() { + for err in [ + anyhow::Error::new(DiscoveryRace::Changed), + DiscoveryRace::PidMismatch { + expected: 1, + actual: 2, + } + .into(), + std::io::Error::from(ErrorKind::TimedOut).into(), + ] { + assert!(retryable_during_startup(&err.context("probe"))); + } + for kind in [ + ErrorKind::PermissionDenied, + ErrorKind::InvalidData, + ErrorKind::ConnectionReset, + ] { + assert!(!retryable_during_startup( + &std::io::Error::from(kind).into() + )); + } + assert!(!retryable_during_startup(&anyhow::anyhow!("timed out"))); + } } + +#[cfg(all(test, unix))] +#[path = "probe_tests.rs"] +mod readiness_tests; diff --git a/crates/bsk-cli/src/daemon/probe_tests.rs b/crates/bsk-cli/src/daemon/probe_tests.rs new file mode 100644 index 0000000..acc9205 --- /dev/null +++ b/crates/bsk-cli/src/daemon/probe_tests.rs @@ -0,0 +1,206 @@ +use super::*; +use crate::daemon::{paths, test_support::isolated}; +use bsk_protocol::{Frame, ResponseBody, ResponseFrame}; +use std::io::{BufRead, BufReader, Write}; +use std::os::unix::net::UnixListener; +use std::sync::{ + Arc, + atomic::{AtomicBool, AtomicUsize, Ordering}, +}; +use std::time::Instant; + +fn fixture_info() -> DaemonInfo { + DaemonInfo::now( + std::process::id(), + paths::bsk_home().unwrap().join("probe.sock"), + 12345, + env!("CARGO_PKG_VERSION"), + ) +} + +fn status(info: &DaemonInfo) -> ResponseBody { + ResponseBody::Ok(serde_json::json!({ + "pid": info.pid, "daemon_version": info.version, "protocol_version": "1.1", + "uptime_secs": 1, "ws_port": info.ws_port, "sock_path": info.sock_path, + "browsers": [], "sessions": [] + })) +} + +struct Server { + stop: Arc, + requests: Arc, + thread: Option>, +} + +impl Server { + fn new( + listener: UnixListener, + mut reply: impl FnMut(usize) -> ResponseBody + Send + 'static, + ) -> Self { + listener.set_nonblocking(true).unwrap(); + let stop = Arc::new(AtomicBool::new(false)); + let done = stop.clone(); + let requests = Arc::new(AtomicUsize::new(0)); + let count = requests.clone(); + let thread = std::thread::spawn(move || { + while !done.load(Ordering::Relaxed) { + let stream = match listener.accept() { + Ok((stream, _)) => stream, + Err(err) if err.kind() == ErrorKind::WouldBlock => { + std::thread::sleep(Duration::from_millis(5)); + continue; + } + Err(err) => panic!("accept: {err}"), + }; + stream + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + let mut reader = BufReader::new(stream); + let mut line = String::new(); + if !matches!(reader.read_line(&mut line), Ok(n) if n > 0) { + continue; + } + let Frame::Request(request) = serde_json::from_str(&line).unwrap() else { + panic!("expected request"); + }; + let body = reply(count.fetch_add(1, Ordering::SeqCst)); + let response = Frame::Response(ResponseFrame { + id: request.id, + body, + }); + // A timed-out client may already have closed its connection. + let _ = writeln!( + reader.get_mut(), + "{}", + serde_json::to_string(&response).unwrap() + ); + } + }); + Self { + stop, + requests, + thread: Some(thread), + } + } +} + +impl Drop for Server { + fn drop(&mut self) { + self.stop.store(true, Ordering::Relaxed); + self.thread.take().unwrap().join().unwrap(); + } +} + +#[test] +fn readiness_waits_for_bound_endpoint_to_serve_and_publish_new_metadata() { + isolated( + concat!( + module_path!(), + "::readiness_waits_for_bound_endpoint_to_serve_and_publish_new_metadata" + ), + || { + let current = fixture_info(); + let mut stale = current.clone(); + stale.pid += 1; + info::write(&stale).unwrap(); + let listener = UnixListener::bind(¤t.sock_path).unwrap(); + let server = Server::new(listener, move |n| { + if n == 0 { + // Start the delay only after receiving the first probe. The + // endpoint is bound but cannot serve within one probe budget. + std::thread::sleep(PROBE_TIMEOUT + Duration::from_millis(150)); + info::write(¤t).unwrap(); + } + status(¤t) + }); + let started = Instant::now(); + let daemon = wait_for_ready(Duration::from_secs(3)).unwrap(); + assert_eq!(daemon.info.pid, std::process::id()); + assert!(started.elapsed() >= PROBE_TIMEOUT); + assert!(server.requests.load(Ordering::SeqCst) >= 2); + assert!(!paths::lock_path().unwrap().exists()); + }, + ); +} + +#[test] +fn readiness_timeout_uses_the_caller_deadline_and_preserves_the_cause() { + isolated( + concat!( + module_path!(), + "::readiness_timeout_uses_the_caller_deadline_and_preserves_the_cause" + ), + || { + let stale = fixture_info(); + info::write(&stale).unwrap(); + let _listener = UnixListener::bind(&stale.sock_path).unwrap(); + let deadline = Duration::from_millis(1200); + let started = Instant::now(); + let error = wait_for_ready(deadline) + .err() + .expect("unserved endpoint must time out"); + assert!(started.elapsed() >= deadline); + assert!(started.elapsed() < Duration::from_secs(3)); + assert!(error.is::()); + assert!(format!("{error:#}").contains("failed to become ready within")); + assert_eq!(info::read().unwrap(), Some(stale)); + assert!(!paths::lock_path().unwrap().exists()); + }, + ); +} + +#[test] +fn readiness_retries_pid_mismatch_and_discovery_changes() { + isolated( + concat!( + module_path!(), + "::readiness_retries_pid_mismatch_and_discovery_changes" + ), + || { + for changing_metadata in [false, true] { + let current = fixture_info(); + let mut initial = current.clone(); + initial.pid += 1; + info::write(&initial).unwrap(); + let listener = UnixListener::bind(¤t.sock_path).unwrap(); + let path = current.sock_path.clone(); + let server = Server::new(listener, move |n| { + let mut published = current.clone(); + if changing_metadata { + published.started_at_epoch_secs += n.min(4) as u64; + } + if changing_metadata || n >= 4 { + info::write(&published).unwrap(); + } + status(&published) + }); + let daemon = wait_for_ready(Duration::from_secs(3)).unwrap(); + assert_eq!(daemon.info.pid, std::process::id()); + assert!(server.requests.load(Ordering::SeqCst) >= 5); + drop(server); + std::fs::remove_file(path).unwrap(); + } + }, + ); +} + +#[test] +fn readiness_does_not_retry_invalid_responses() { + isolated( + concat!( + module_path!(), + "::readiness_does_not_retry_invalid_responses" + ), + || { + let current = fixture_info(); + info::write(¤t).unwrap(); + let listener = UnixListener::bind(¤t.sock_path).unwrap(); + let server = Server::new(listener, |_| { + ResponseBody::Ok(serde_json::json!({"unexpected": true})) + }); + assert!(wait_for_ready(Duration::from_secs(3)).is_err()); + assert_eq!(server.requests.load(Ordering::SeqCst), 1); + assert_eq!(info::read().unwrap(), Some(current)); + }, + ); +} diff --git a/crates/bsk-cli/src/daemon/start.rs b/crates/bsk-cli/src/daemon/start.rs index 8f2402a..c3636b9 100644 --- a/crates/bsk-cli/src/daemon/start.rs +++ b/crates/bsk-cli/src/daemon/start.rs @@ -151,6 +151,7 @@ pub fn run_stop() -> Result<()> { /// Stop with the same checks for explicit management and CLI self-update. /// The return value tells the updater whether it should restart a daemon. +/// Observed replacement instances are errors, so update/restart must abort. pub(crate) fn stop_if_running() -> Result { let daemon = match probe::probe(Duration::from_secs(2))? { Probe::Ready(daemon) => daemon, @@ -207,7 +208,11 @@ fn wait_for_stopped(expected: &daemon_info::DaemonInfo, timeout: Duration) -> Re Ok(_lock) => { // Normal shutdown removes its own metadata. A forced exit // leaves it behind; only remove that exact instance's record. - if daemon_info::read()?.as_ref() == Some(expected) { + if let Some(current) = daemon_info::read()? { + anyhow::ensure!( + ¤t == expected, + "daemon instance changed while stopping; aborting stop to preserve the replacement" + ); match probe::probe(PROBE_TIMEOUT)? { Probe::Absent(Some(info)) if &info == expected => daemon_info::remove()?, _ => anyhow::bail!( @@ -222,8 +227,11 @@ fn wait_for_stopped(expected: &daemon_info::DaemonInfo, timeout: Duration) -> Re .as_ref() .is_some_and(|info| info != expected) { - // A replacement acquired the lock; never remove its files. - return Ok(true); + // The old instance exited, but update/restart must not + // treat a replacement still running as a successful stop. + anyhow::bail!( + "daemon instance changed while stopping; aborting stop to preserve the replacement" + ); } } Err(err) => return Err(err.context("check daemon shutdown lock")), @@ -1064,6 +1072,38 @@ fn send_kill(_pid: u32) -> Result<()> { mod tests { use super::*; + #[test] + fn stopping_rejects_replacement_metadata_with_or_without_a_held_lock() { + crate::daemon::test_support::isolated( + concat!( + module_path!(), + "::stopping_rejects_replacement_metadata_with_or_without_a_held_lock" + ), + || { + let expected = daemon_info::DaemonInfo::now( + 123, + paths::bsk_home().unwrap().join("unused.sock"), + 12345, + env!("CARGO_PKG_VERSION"), + ); + let mut replacement = expected.clone(); + replacement.pid += 1; + daemon_info::write(&replacement).unwrap(); + for held in [true, false] { + let _lock = held.then(|| lockfile::acquire().unwrap()); + let error = wait_for_stopped(&expected, Duration::from_secs(1)).unwrap_err(); + assert!( + error + .to_string() + .contains("daemon instance changed while stopping") + ); + assert_eq!(daemon_info::read().unwrap(), Some(replacement.clone())); + assert!(paths::lock_path().unwrap().exists()); + } + }, + ); + } + #[test] fn format_duration_round_trips_seconds_and_millis() { assert_eq!(format_duration(Duration::from_secs(5)), "5s"); diff --git a/crates/bsk-cli/src/daemon/test_support.rs b/crates/bsk-cli/src/daemon/test_support.rs new file mode 100644 index 0000000..1795584 --- /dev/null +++ b/crates/bsk-cli/src/daemon/test_support.rs @@ -0,0 +1,24 @@ +//! Run lifecycle tests with a fixed BSK_HOME, without mutating the parallel +//! test runner's process environment. + +pub(super) fn isolated(test_name: &str, test: impl FnOnce()) { + let test_name = test_name.split_once("::").unwrap().1; + if std::env::var("BSK_LIFECYCLE_TEST").as_deref() == Ok(test_name) { + test(); + return; + } + let home = tempfile::tempdir().unwrap(); + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", test_name, "--nocapture"]) + .env("BSK_LIFECYCLE_TEST", test_name) + .env("BSK_HOME", home.path()) + .env("BSK_AUTO_UPDATE", "off") + .output() + .unwrap(); + assert!( + output.status.success(), + "stdout={} stderr={}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +} diff --git a/crates/bsk-cli/tests/auto_spawn.rs b/crates/bsk-cli/tests/auto_spawn.rs index ad22357..f6f6a88 100644 --- a/crates/bsk-cli/tests/auto_spawn.rs +++ b/crates/bsk-cli/tests/auto_spawn.rs @@ -1,15 +1,9 @@ -//! Verify `ensure_daemon` spawns the daemon when none is running. -//! -//! We can't call `ensure_daemon` directly from a test process because -//! `current_exe()` would point to the test binary, not `bsk`. Instead we -//! drive the same effect end-to-end via `bsk status` (which itself calls -//! into `ensure_daemon` in M3.3) — but for M3.2 we test the helper by -//! pointing `current_exe` indirection at the actual `bsk` binary through -//! a small shim test. +//! Exercise automatic daemon startup and reuse through the real `bsk status` +//! command, so ensure_daemon's current_exe() points to the CLI binary. #![cfg(unix)] -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::process::Command; use std::time::{Duration, Instant}; @@ -19,6 +13,66 @@ fn bsk_bin() -> PathBuf { PathBuf::from(env!("CARGO_BIN_EXE_bsk")) } +fn command(home: &Path, args: &[&str]) -> Command { + let mut cmd = Command::new(bsk_bin()); + cmd.args(args) + .env("BSK_HOME", home) + .env("BSK_AUTO_UPDATE", "off") + .env("BSK_BROWSER_WAIT_MS", "0") + .env("RUST_LOG", "warn"); + cmd +} + +struct StopOnDrop(PathBuf); + +impl Drop for StopOnDrop { + fn drop(&mut self) { + let _ = command(&self.0, &["daemon", "stop"]).output(); + } +} + +#[test] +fn status_auto_spawns_from_an_empty_home() { + // BSK_HOME isolates files, not the production default WS port. Exercise + // the real default on clean CI hosts; never stop a developer's daemon. + match std::net::TcpListener::bind(("127.0.0.1", bsk::cli::daemon::DEFAULT_WS_PORT)) { + Ok(port) => drop(port), + Err(err) if err.kind() == std::io::ErrorKind::AddrInUse => { + assert!( + std::env::var_os("CI").is_none(), + "CI must provide a free default WS port for the auto-spawn regression: {err}" + ); + eprintln!("skipping default-port auto-spawn test: port 52800 is already in use"); + return; + } + Err(err) => panic!("check default WS port: {err}"), + } + let tmp = TempDir::new().unwrap(); + let home = tmp.path().join("bsk"); + let _cleanup = StopOnDrop(home.clone()); + assert!(!home.exists()); + + let out = command(&home, &["--json", "status"]).output().unwrap(); + assert!( + out.status.success(), + "{}", + String::from_utf8_lossy(&out.stderr) + ); + let reported: bsk_protocol::StatusResult = serde_json::from_slice(&out.stdout).unwrap(); + let info: bsk::daemon::info::DaemonInfo = + serde_json::from_slice(&std::fs::read(home.join("daemon.json")).unwrap()).unwrap(); + assert_eq!(reported.pid, info.pid); + assert_eq!(reported.sock_path, info.sock_path); + assert!(info.pid > 0); + assert!(home.join("daemon.lock").exists()); + + // A second status must use the same successfully started instance. + let again = command(&home, &["--json", "status"]).output().unwrap(); + assert!(again.status.success()); + let reused: bsk_protocol::StatusResult = serde_json::from_slice(&again.stdout).unwrap(); + assert_eq!(reused.pid, info.pid); +} + fn wait_for_pid_exit(pid: i32, timeout: Duration) -> bool { let deadline = Instant::now() + timeout; while Instant::now() < deadline { @@ -39,12 +93,13 @@ fn ensure_daemon_idempotent_when_already_running() { std::fs::create_dir_all(&home).unwrap(); // Start a daemon manually first. - let out = Command::new(bsk_bin()) - .args(["daemon", "start", "--port", "0", "--daemon-idle", "60s"]) - .env("BSK_HOME", &home) - .env("RUST_LOG", "warn") - .output() - .unwrap(); + let _cleanup = StopOnDrop(home.clone()); + let out = command( + &home, + &["daemon", "start", "--port", "0", "--daemon-idle", "60s"], + ) + .output() + .unwrap(); assert!(out.status.success()); let info_path = home.join("daemon.json"); @@ -52,15 +107,15 @@ fn ensure_daemon_idempotent_when_already_running() { serde_json::from_slice(&std::fs::read(&info_path).unwrap()).unwrap(); let pid_before = info["pid"].as_u64().unwrap() as i32; - // Now another `bsk daemon start` is invoked. Because the lock is - // held, the spawned child should exit quickly and the existing - // daemon should keep its pid. - let _ = Command::new(bsk_bin()) - .args(["daemon", "start", "--port", "0", "--daemon-idle", "60s"]) - .env("BSK_HOME", &home) - .env("RUST_LOG", "warn") - .output() - .unwrap(); + // Exercise ensure_daemon through status, not a second explicit start. + let status = command(&home, &["--json", "status"]).output().unwrap(); + assert!( + status.status.success(), + "{}", + String::from_utf8_lossy(&status.stderr) + ); + let reported: bsk_protocol::StatusResult = serde_json::from_slice(&status.stdout).unwrap(); + assert_eq!(reported.pid, pid_before as u32); let info_after: serde_json::Value = serde_json::from_slice(&std::fs::read(&info_path).unwrap()).unwrap(); @@ -68,9 +123,6 @@ fn ensure_daemon_idempotent_when_already_running() { assert_eq!(pid_before, pid_after, "daemon pid should not change"); // Clean up. - let _ = Command::new(bsk_bin()) - .args(["daemon", "stop"]) - .env("BSK_HOME", &home) - .output(); + let _ = command(&home, &["daemon", "stop"]).output(); assert!(wait_for_pid_exit(pid_before, Duration::from_secs(5))); }