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..761d2ce 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; @@ -16,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 new file mode 100644 index 0000000..75cf079 --- /dev/null +++ b/crates/bsk-cli/src/daemon/probe.rs @@ -0,0 +1,237 @@ +//! 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, +} + +#[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. + 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 { + return Err(DiscoveryRace::PidMismatch { + expected: info.pid, + actual: actual_pid, + } + .into()); + } + 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") + })? +} + +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 + ) + }) +} + +/// 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()); + 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::*; + + #[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)); + } + } + + #[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 4d58457..c3636b9 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,114 @@ 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. +/// 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, + 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 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!( + "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) + { + // 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")), + } + 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 +1005,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 +1022,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}; @@ -1101,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/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/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))); } 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)