diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6c0a879..a74512f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -37,6 +37,21 @@ jobs: - name: Run Rust tests run: cargo test --workspace --locked + windows-daemon: + name: Windows daemon regression tests + runs-on: windows-latest + + steps: + - uses: actions/checkout@v6 + + - uses: dtolnay/rust-toolchain@stable + + - name: Run Windows named-pipe tests + run: cargo test -p bsk --lib --locked daemon::ipc::windows::tests + + - name: Run Windows process liveness tests + run: cargo test -p bsk --lib --locked daemon::lockfile::tests + frontend: name: Frontend lint, typecheck, tests, build runs-on: ubuntu-latest diff --git a/crates/bsk-cli/src/daemon/ipc.rs b/crates/bsk-cli/src/daemon/ipc.rs index 186cade..45b64ff 100644 --- a/crates/bsk-cli/src/daemon/ipc.rs +++ b/crates/bsk-cli/src/daemon/ipc.rs @@ -1399,6 +1399,28 @@ mod windows { connected = pipe.connect() => { match connected { Ok(()) => { + // Keep the connected instance alive until its replacement + // exists. Otherwise a fast handler can close the last + // instance and make new clients fail with NotFound. + loop { + match ServerOptions::new() + .access_inbound(true) + .access_outbound(true) + .create(&listener.pipe_name) + { + Ok(next) => { + listener.first = Some(next); + break; + } + Err(err) => { + warn!(?err, "create next named-pipe instance failed"); + tokio::select! { + _ = &mut shutdown => return, + _ = tokio::time::sleep(std::time::Duration::from_millis(50)) => {} + } + } + } + } on_open(); let handler = handler.clone(); let on_act = on_activity.clone(); @@ -1467,6 +1489,118 @@ mod windows { } Ok(()) } + #[cfg(test)] + mod tests { + use super::*; + use std::sync::Mutex; + use std::time::Duration; + use tokio::net::windows::named_pipe::ClientOptions; + use tokio::sync::oneshot; + + fn isolated_listener() -> NamedPipeListener { + let pipe_name = format!(r"\\.\pipe\bsk-test-{}", uuid::Uuid::new_v4()); + let first = ServerOptions::new() + .first_pipe_instance(true) + .create(&pipe_name) + .unwrap(); + NamedPipeListener { + pipe_name, + first: Some(first), + } + } + + #[tokio::test] + async fn next_instance_exists_before_connection_is_handed_off() { + let listener = isolated_listener(); + let name = listener.pipe_name.clone(); + let probe_name = name.clone(); + let (probe_tx, probe_rx) = oneshot::channel(); + let probe_tx = Mutex::new(Some(probe_tx)); + let (stop_tx, stop_rx) = oneshot::channel(); + let server = tokio::spawn(serve( + listener, + crate::daemon::ipc::default_ping_handler(), + move || { + if let Some(tx) = probe_tx.lock().unwrap().take() { + // This callback runs before the connection task can finish. + // A second client must already have an instance to open. + let result = ClientOptions::new().open(&probe_name).map(drop); + let _ = tx.send(result); + } + }, + || {}, + || {}, + async { + let _ = stop_rx.await; + }, + )); + let client = ClientOptions::new().open(&name).unwrap(); + let result = tokio::time::timeout(Duration::from_secs(5), probe_rx).await; + drop(client); + let _ = stop_tx.send(()); + server.await.unwrap(); + result + .expect("accept callback ran") + .unwrap() + .expect("next pipe instance is ready"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn rapid_disconnects_and_concurrent_rpc_connections() { + let listener = isolated_listener(); + let name = listener.pipe_name.clone(); + let (stop_tx, stop_rx) = oneshot::channel(); + let server = tokio::spawn(serve( + listener, + crate::daemon::ipc::default_ping_handler(), + || {}, + || {}, + || {}, + async { + let _ = stop_rx.await; + }, + )); + let exercise = async { + // Include clients that close without sending any request. + for _ in 0..64 { + drop( + crate::ipc_client::Client::connect_path(name.clone().into()) + .await + .unwrap(), + ); + } + let mut clients = tokio::task::JoinSet::new(); + for _ in 0..8 { + let name = name.clone(); + clients.spawn(async move { + for _ in 0..32 { + let mut client = + crate::ipc_client::Client::connect_path(name.clone().into()) + .await + .unwrap(); + let reply: bsk_protocol::PingResult = client + .call( + bsk_protocol::Method::SystemPing, + &serde_json::json!({}), + Duration::from_secs(5), + ) + .await + .unwrap() + .unwrap(); + assert!(reply.pong); + } + }); + } + while let Some(result) = clients.join_next().await { + result.unwrap(); + } + }; + let result = tokio::time::timeout(Duration::from_secs(30), exercise).await; + let _ = stop_tx.send(()); + server.await.unwrap(); + result.expect("connection exercise completed"); + } + } } #[cfg(windows)] diff --git a/crates/bsk-cli/src/daemon/lockfile.rs b/crates/bsk-cli/src/daemon/lockfile.rs index e0bf402..2a33747 100644 --- a/crates/bsk-cli/src/daemon/lockfile.rs +++ b/crates/bsk-cli/src/daemon/lockfile.rs @@ -90,17 +90,21 @@ pub fn pid_alive(pid: u32) -> bool { #[cfg(windows)] { - use windows_sys::Win32::Foundation::CloseHandle; + use windows_sys::Win32::Foundation::{CloseHandle, WAIT_TIMEOUT}; use windows_sys::Win32::System::Threading::{ - OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION, + OpenProcess, PROCESS_SYNCHRONIZE, WaitForSingleObject, }; unsafe { - let handle = OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid); + let handle = OpenProcess(PROCESS_SYNCHRONIZE, 0, pid); if handle.is_null() { return false; } + // An exited process can still be opened while another handle + // keeps its kernel object alive. Poll its termination signal; + // unlike checking STILL_ACTIVE, this also handles exit code 259. + let alive = WaitForSingleObject(handle, 0) == WAIT_TIMEOUT; CloseHandle(handle); - true + alive } } @@ -110,3 +114,39 @@ pub fn pid_alive(pid: u32) -> bool { false } } + +#[cfg(all(test, windows))] +mod tests { + use super::pid_alive; + use std::os::windows::process::CommandExt; + use std::process::Command; + + #[test] + fn current_process_is_alive() { + assert!(pid_alive(std::process::id())); + } + + #[test] + fn invalid_process_is_not_alive() { + assert!(!pid_alive(0)); + assert!(!pid_alive(u32::MAX)); + } + + #[test] + fn exited_process_with_retained_handle_is_not_alive() { + for exit_code in [0, 259] { + let mut child = Command::new("cmd.exe") + .args(["/D", "/C", &format!("exit {exit_code}")]) + .creation_flags(0x0800_0000) // CREATE_NO_WINDOW + .spawn() + .expect("spawn short-lived process"); + assert_eq!(child.wait().unwrap().code(), Some(exit_code)); + // Keep Child (and its process handle) alive during the probe. + assert!( + !pid_alive(child.id()), + "exited process with code {exit_code}" + ); + drop(child); + } + } +}