From caea98cd52b65ac5bb0c05c2eed658a383c9a8a0 Mon Sep 17 00:00:00 2001 From: Zhang GH <5351707+iuyo5678@users.noreply.github.com> Date: Tue, 8 Sep 2026 19:21:54 +0800 Subject: [PATCH 1/2] fix(windows): harden installation and parent cancellation --- .github/workflows/ci.yml | 14 +- crates/bsk-cli/src/cli/business_rpc.rs | 69 +++++++- crates/bsk-cli/tests/windows_parent_cancel.rs | 163 ++++++++++++++++++ install.ps1 | 56 ++++-- .../dsh-plugin-browserskill/src/runner.ts | 60 +++++-- .../tests/runner.test.ts | 72 +++++++- scripts/install-windows.test.ps1 | 135 +++++++++++++++ 7 files changed, 534 insertions(+), 35 deletions(-) create mode 100644 crates/bsk-cli/tests/windows_parent_cancel.rs create mode 100644 scripts/install-windows.test.ps1 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index be5c2eb..be78236 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -38,7 +38,7 @@ jobs: run: cargo test --workspace --locked windows-daemon: - name: Windows daemon and update regression tests + name: Windows daemon, installer and cancellation regression tests runs-on: windows-latest steps: @@ -55,8 +55,16 @@ jobs: - name: Run Windows update tests run: cargo test -p bsk --lib --locked cli::update::tests - - name: Run Windows update process tests - run: cargo test -p bsk --test windows_update --locked + - name: Run Windows update and cancellation process tests + run: cargo test -p bsk --test windows_update --test windows_parent_cancel --locked + + - name: Run installer tests (Windows PowerShell 5.1) + shell: powershell + run: ./scripts/install-windows.test.ps1 + + - name: Run installer tests (PowerShell 7) + shell: pwsh + run: ./scripts/install-windows.test.ps1 frontend: name: Frontend lint, typecheck, tests, build diff --git a/crates/bsk-cli/src/cli/business_rpc.rs b/crates/bsk-cli/src/cli/business_rpc.rs index 77687b3..46c7efe 100644 --- a/crates/bsk-cli/src/cli/business_rpc.rs +++ b/crates/bsk-cli/src/cli/business_rpc.rs @@ -73,6 +73,14 @@ where P: Serialize + Send + 'static, R: DeserializeOwned + Send + 'static, { + let parent_cancel = stdin_cancellation(); + if parent_cancel.as_ref().is_some_and(|rx| *rx.borrow()) { + return Err(CliError::from_rpc(RpcError { + code: ErrorCode::Cancelled, + message: "parent closed the cancellation pipe".into(), + data: None, + })); + } let rpc_id: RpcId = format!("{}-{}", rpc_id_prefix, random_short_id()); let mut client = IpcClient::connect(sock.clone()).await?; let rpc_id_for_call = rpc_id.clone(); @@ -88,16 +96,16 @@ where let outcome = tokio::select! { biased; res = &mut main_fut => res?, - sig = wait_for_sigint() => { - sig.context("install SIGINT handler").map_err(CliError::Local)?; - debug!(rpc_id = %rpc_id_for_cancel, "SIGINT: forwarding cancel to daemon"); + sig = wait_for_cancel(parent_cancel) => { + sig.context("wait for cancellation").map_err(CliError::Local)?; + debug!(rpc_id = %rpc_id_for_cancel, "forwarding cancel to daemon"); let _ = send_cancel(&sock, &rpc_id_for_cancel).await; match tokio::time::timeout(CANCEL_RESPONSE_GRACE, &mut main_fut).await { Ok(res) => res?, Err(_) => { debug!( rpc_id = %rpc_id_for_cancel, - "SIGINT: cancel grace elapsed; synthesising cancelled error" + "cancel grace elapsed; synthesising cancelled error" ); return Err(CliError::from_rpc(RpcError { code: ErrorCode::Cancelled, @@ -146,8 +154,57 @@ fn random_short_id() -> String { bytes.iter().map(|b| format!("{b:02x}")).collect() } -/// Wait for either Ctrl-C or process SIGTERM. Returning here means -/// the user has asked us to stop. +/// Opt-in control pipe for parent processes that cannot send Ctrl-C (Node on +/// Windows). One reader survives the short-lived runtimes used by multi-RPC +/// commands such as upload. Ordinary CLI stdin and terminal Ctrl-C are unchanged. +fn stdin_cancellation() -> Option> { + if std::env::var("BSK_CANCEL_ON_STDIN_CLOSE").as_deref() != Ok("1") { + return None; + } + static CANCEL: std::sync::OnceLock> = + std::sync::OnceLock::new(); + Some( + CANCEL + .get_or_init(|| { + let (tx, rx) = tokio::sync::watch::channel(false); + std::thread::spawn(move || { + use std::io::Read; + let mut stdin = std::io::stdin().lock(); + let mut buf = [0; 64]; + loop { + match stdin.read(&mut buf) { + Ok(0) => break, + Ok(_) => {} + Err(err) if err.kind() == std::io::ErrorKind::Interrupted => continue, + Err(_) => break, + } + } + let _ = tx.send(true); + }); + rx + }) + .clone(), + ) +} + +async fn wait_for_cancel( + parent_cancel: Option>, +) -> anyhow::Result<()> { + let parent_closed = async move { + match parent_cancel { + Some(mut rx) => { + let _ = rx.wait_for(|closed| *closed).await; + } + None => std::future::pending::<()>().await, + } + }; + tokio::select! { + result = wait_for_sigint() => result, + _ = parent_closed => Ok(()), + } +} + +/// Wait for terminal Ctrl-C. async fn wait_for_sigint() -> anyhow::Result<()> { tokio::signal::ctrl_c().await.context("listen for SIGINT")?; Ok(()) diff --git a/crates/bsk-cli/tests/windows_parent_cancel.rs b/crates/bsk-cli/tests/windows_parent_cancel.rs new file mode 100644 index 0000000..44b4452 --- /dev/null +++ b/crates/bsk-cli/tests/windows_parent_cancel.rs @@ -0,0 +1,163 @@ +//! Real Windows CLI processes cancel over IPC when their parent closes stdin. +#![cfg(windows)] + +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use bsk::daemon::info::DaemonInfo; +use bsk_protocol::{ErrorCode, Frame, Method, ResponseBody, ResponseFrame, RpcError}; +use serde_json::json; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::windows::named_pipe::ServerOptions; +use tokio::process::Command; +use tokio::sync::{Notify, watch}; + +#[tokio::test] +async fn stdin_close_cancels_single_and_multi_rpc_commands_but_is_opt_in() { + for mode in ["wait", "upload", "ordinary"] { + let home = tempfile::tempdir().unwrap(); + let pipe_name = format!(r"\\.\pipe\bsk-parent-cancel-{}", uuid::Uuid::new_v4()); + let info = DaemonInfo::now(std::process::id(), (&pipe_name).into(), 0, "0.2.0"); + std::fs::write( + home.path().join("daemon.json"), + serde_json::to_vec(&info).unwrap(), + ) + .unwrap(); + let mut pipe = ServerOptions::new() + .first_pipe_instance(true) + .create(&pipe_name) + .unwrap(); + let registered = Arc::new(Notify::new()); + let pending = Arc::new(Mutex::new(None::)); + let (cancel, cancelled) = watch::channel(false); + let ready = Arc::clone(®istered); + let name = pipe_name.clone(); + let server = tokio::spawn(async move { + let mut connections = tokio::task::JoinSet::new(); + loop { + pipe.connect().await.unwrap(); + let next = ServerOptions::new().create(&name).unwrap(); + let connection = std::mem::replace(&mut pipe, next); + let ready = Arc::clone(&ready); + let pending = Arc::clone(&pending); + let cancel = cancel.clone(); + let mut cancelled = cancelled.clone(); + let name = name.clone(); + connections.spawn(async move { + let (read, mut write) = tokio::io::split(connection); + let mut reader = BufReader::new(read); + let mut line = String::new(); + while reader.read_line(&mut line).await.unwrap_or(0) > 0 { + let Frame::Request(req) = serde_json::from_str::(&line).unwrap() + else { + panic!("expected request"); + }; + line.clear(); + let body = match req.method { + Method::SystemStatus => ResponseBody::Ok(json!({ + "daemon_version": "0.2.0", "protocol_version": "1.1", + "pid": std::process::id(), "uptime_secs": 0, + "ws_port": 0, "sock_path": name, + "browsers": [], "sessions": [], "version_skew_browsers": [] + })), + Method::TransferBegin => ResponseBody::Ok(json!({ + "transfer_id": "fixture", "chunk_size": 512 + })), + Method::ToolWaitMs if mode == "ordinary" => { + ResponseBody::Ok(json!({"waited_ms": 1})) + } + Method::ToolWaitMs | Method::TransferChunk => { + *pending.lock().unwrap() = Some(req.id.clone()); + ready.notify_one(); + cancelled.wait_for(|value| *value).await.unwrap(); + ResponseBody::Err(RpcError { + code: ErrorCode::Cancelled, + message: "parent cancellation reached daemon".into(), + data: None, + }) + } + Method::Cancel => { + let target = + req.params.as_ref().unwrap()["rpc_id"].as_str().unwrap(); + assert_eq!(pending.lock().unwrap().as_deref(), Some(target)); + cancel.send(true).unwrap(); + ResponseBody::Ok(json!({"cancelled": true})) + } + other => panic!("unexpected request after cancellation: {other:?}"), + }; + let response = Frame::Response(ResponseFrame { id: req.id, body }); + let mut bytes = serde_json::to_vec(&response).unwrap(); + bytes.push(b'\n'); + if write.write_all(&bytes).await.is_err() { + break; + } + } + }); + } + }); + + let mut cmd = Command::new(env!("CARGO_BIN_EXE_bsk")); + cmd.env("BSK_HOME", home.path()) + .env("BSK_AUTO_UPDATE", "off") + .env_remove("BSK_CANCEL_ON_STDIN_CLOSE") + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .creation_flags(0x0800_0000) + .kill_on_drop(true); + if mode == "upload" { + let source = home.path().join("upload.txt"); + std::fs::write(&source, b"upload fixture").unwrap(); + cmd.args([ + "upload", + "--session", + "fixture", + "--selector", + "input", + "--file", + ]) + .arg(source); + } else { + cmd.args(["wait-ms", if mode == "ordinary" { "1ms" } else { "60s" }]); + } + cmd.arg("--json"); + if mode == "ordinary" { + cmd.stdin(std::process::Stdio::null()); + } else { + cmd.env("BSK_CANCEL_ON_STDIN_CLOSE", "1") + .stdin(std::process::Stdio::piped()); + } + let mut child = cmd.spawn().unwrap(); + if mode != "ordinary" { + tokio::time::timeout(Duration::from_secs(10), registered.notified()) + .await + .expect("CLI registered business RPC"); + // Normal stdin bytes are not a cancellation; only closing the pipe is. + child + .stdin + .as_mut() + .unwrap() + .write_all(b"still connected") + .await + .unwrap(); + drop(child.stdin.take()); + } + let output = tokio::time::timeout(Duration::from_secs(5), child.wait_with_output()) + .await + .expect("CLI must settle promptly") + .unwrap(); + server.abort(); + let _ = server.await; + assert_eq!( + output.status.code(), + Some(if mode == "ordinary" { 0 } else { 2 }), + "{mode}: stdout={} stderr={}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + if mode != "ordinary" { + let body: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(body["code"], "cancelled"); + assert_eq!(body["message"], "parent cancellation reached daemon"); + } + } +} diff --git a/install.ps1 b/install.ps1 index ccfacec..5dd7889 100644 --- a/install.ps1 +++ b/install.ps1 @@ -21,6 +21,7 @@ $ErrorActionPreference = "Stop" $Repo = if ($env:BSK_REPO) { $env:BSK_REPO } else { "Tencent/BrowserSkill" } $InstallDir = if ($env:BSK_INSTALL_DIR) { $env:BSK_INSTALL_DIR } else { Join-Path $HOME ".local\bin" } +$InstallDir = $ExecutionContext.SessionState.Path.GetUnresolvedProviderPathFromPSPath($InstallDir) $GitHub = "https://github.com/${Repo}" function Write-Log { @@ -64,7 +65,7 @@ function Get-PlatformTriple { function Add-ToUserPath { param([string]$Dir) - $currentUserPath = [Environment]::GetEnvironmentVariable("PATH", "User") -split ";" | Where-Object { $_ } + $currentUserPath = @([Environment]::GetEnvironmentVariable("PATH", "User") -split ";" | Where-Object { $_ }) if ($currentUserPath -contains $Dir) { Write-Log "$Dir is already in your user PATH" @@ -90,30 +91,59 @@ function Add-ToSessionPath { # ── Git Bash (bash environment) PATH helper ────────────────────────────────── function Add-ToBashProfile { - param([string]$Dir) + param([string]$Dir, [string]$BashRc = (Join-Path $HOME ".bashrc")) # Convert Windows path (e.g. C:\Users\foo\.local\bin) to Git-Bash Unix-style (/c/Users/foo/.local/bin) $unixPath = $Dir -replace '\\', '/' if ($unixPath -match '^([A-Z]):(.*)$') { $unixPath = '/' + $matches[1].ToLower() + $matches[2] } - $bashRc = Join-Path $HOME ".bashrc" - $exportLine = "export PATH=""${unixPath}:`$PATH"" # bsk CLI" + # Single-quote the literal directory; only the existing PATH is expanded. + $shellQuote = "'" + [char]34 + "'" + [char]34 + "'" + $quotedPath = "'" + $unixPath.Replace("'", $shellQuote) + "'" + $exportLine = "export PATH=${quotedPath}:`"`$PATH`" # bsk CLI" - if (Test-Path $bashRc) { - $content = Get-Content $bashRc -Raw -ErrorAction SilentlyContinue - if ($content -match [regex]::Escape($unixPath)) { + if (Test-Path -LiteralPath $BashRc) { + $content = [System.IO.File]::ReadAllText($BashRc) + if ($content.Contains($exportLine)) { Write-Log "$unixPath is already in ~/.bashrc" return } } - Add-Content $bashRc "`n$exportLine" -Encoding ASCII + # Explicit BOM-less UTF-8 also works in Windows PowerShell 5.1. + [System.IO.File]::AppendAllText($BashRc, "`n$exportLine`n", (New-Object System.Text.UTF8Encoding($false))) Write-Log "added ${unixPath} to ~/.bashrc" } # ── Main ────────────────────────────────────────────────────────────────────── +# Stage on the destination volume before stopping the daemon. Never truncate +# the installed executable: failed replacement must leave it usable. +function Install-Binary { + param([string]$Source, [string]$Target) + + $staged = "$Target.install-$([Guid]::NewGuid().ToString('N'))" + try { + [System.IO.File]::Copy($Source, $staged) + if ([System.IO.File]::Exists($Target)) { + # Use the downloaded CLI: older versions have broken Windows + # liveness checks. Stop verifies daemon identity before terminating it. + & $Source daemon stop + if ($LASTEXITCODE -ne 0) { throw "could not stop bsk daemon; existing installation was not replaced" } + # PowerShell 5.1 converts $null to an empty path for string parameters. + [System.IO.File]::Replace($staged, $Target, [NullString]::Value) + Write-Log "daemon will restart automatically on the next browser command" + } + else { + [System.IO.File]::Move($staged, $Target) + } + } + finally { + if ([System.IO.File]::Exists($staged)) { [System.IO.File]::Delete($staged) } + } +} + function Main { $platform = Get-PlatformTriple @@ -184,7 +214,7 @@ function Main { New-Item -ItemType Directory -Path $InstallDir -Force | Out-Null } - Copy-Item -Path (Join-Path $tempDir "bsk.exe") -Destination (Join-Path $InstallDir "bsk.exe") -Force + Install-Binary -Source (Join-Path $tempDir "bsk.exe") -Target (Join-Path $InstallDir "bsk.exe") Write-Log "installed bsk to $InstallDir\bsk.exe" @@ -199,12 +229,8 @@ function Main { # Verify $bskPath = Join-Path $InstallDir "bsk.exe" - if (Get-Command bsk -ErrorAction SilentlyContinue) { - & bsk --version - } - else { - Write-Log "verify install: & ""$bskPath"" --version" - } + & $bskPath --version + if ($LASTEXITCODE -ne 0) { throw "installed bsk failed verification" } Write-Log "done" Write-Host "" diff --git a/packages/dsh-plugin-browserskill/src/runner.ts b/packages/dsh-plugin-browserskill/src/runner.ts index 13a5752..c80080e 100644 --- a/packages/dsh-plugin-browserskill/src/runner.ts +++ b/packages/dsh-plugin-browserskill/src/runner.ts @@ -5,7 +5,7 @@ * map the CLI's JSON error envelope onto a thrown `BskError`. */ -import { type ChildProcess, spawn } from "node:child_process"; +import { type ChildProcess, type SpawnOptionsWithoutStdio, spawn } from "node:child_process"; /** Shape of the JSON error envelope `bsk --json` prints on failure. */ export interface BskErrorBody { @@ -53,7 +53,11 @@ export interface BskRunOptions { } /** Minimal spawn signature so tests can substitute a fake child process. */ -export type SpawnImpl = (command: string, args: string[]) => ChildProcess; +export type SpawnImpl = ( + command: string, + args: string[], + options?: SpawnOptionsWithoutStdio, +) => ChildProcess; export interface BskRunner { /** Run `bsk --json` and collect its output. */ @@ -64,29 +68,65 @@ export interface BskRunner { killFor(tag: string): number; } -// Business RPCs translate SIGINT into the daemon's cancel(rpc_id) protocol. -// Give that bounded reconciliation path time to settle before the hard kill. +// Business RPCs translate Ctrl-C / opt-in stdin EOF into cancel(rpc_id). +// Allow reconciliation before hard-killing an old or unresponsive CLI. const KILL_GRACE_MS = 3000; +// Windows IPC may spend 5s connecting, then 2s cancelling and 2s settling. +const WINDOWS_KILL_GRACE_MS = 10_000; const SESSION_BUSY_RETRY_DELAY_MS = 100; export function createBskRunner(bskPath: string, spawnImpl: SpawnImpl = spawn): BskRunner { const live = new Map(); + const windows = process.platform === "win32"; + const cancelling = new Set(); function killChild(child: ChildProcess): void { - if (child.exitCode !== null || child.signalCode !== null) return; - child.kill("SIGINT"); - const force = setTimeout(() => { - if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); - }, KILL_GRACE_MS); + if (child.exitCode !== null || child.signalCode !== null || cancelling.has(child)) return; + cancelling.add(child); + // Node kills Windows children outright for SIGINT. EOF asks the CLI to + // send its existing cancel RPC and wait for browser reconciliation. + if (windows && child.stdin) child.stdin.end(); + else child.kill("SIGINT"); + const force = setTimeout( + () => { + if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); + }, + windows ? WINDOWS_KILL_GRACE_MS : KILL_GRACE_MS, + ); force.unref(); + child.once("close", () => { + clearTimeout(force); + cancelling.delete(child); + }); } return { run(args, options = {}) { + if (options.signal?.aborted) { + return Promise.resolve({ + code: null, + stdout: "", + stderr: "", + timedOut: false, + aborted: true, + }); + } return new Promise((resolve, reject) => { let child: ChildProcess; try { - child = spawnImpl(bskPath, [...args, "--json"]); + child = spawnImpl( + bskPath, + [...args, "--json"], + windows + ? { + windowsHide: true, + env: { ...process.env, BSK_CANCEL_ON_STDIN_CLOSE: "1" }, + } + : undefined, + ); + // A child exiting while cancellation closes stdin may report EPIPE. + // Its close/error event remains the authority for the run result. + child.stdin?.on("error", () => {}); } catch (error) { reject(error); return; diff --git a/packages/dsh-plugin-browserskill/tests/runner.test.ts b/packages/dsh-plugin-browserskill/tests/runner.test.ts index ad6d81a..8ea7af4 100644 --- a/packages/dsh-plugin-browserskill/tests/runner.test.ts +++ b/packages/dsh-plugin-browserskill/tests/runner.test.ts @@ -1,6 +1,7 @@ import type { ChildProcess } from "node:child_process"; import { EventEmitter } from "node:events"; -import { describe, expect, it } from "vitest"; +import { PassThrough } from "node:stream"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { type BskRunResult, createBskRunner, @@ -92,6 +93,75 @@ describe("createBskRunner", () => { }); }); +describe("Windows parent cancellation", () => { + const originalPlatform = Object.getOwnPropertyDescriptor(process, "platform")!; + beforeEach(() => Object.defineProperty(process, "platform", { value: "win32" })); + afterEach(() => { + Object.defineProperty(process, "platform", originalPlatform); + vi.useRealTimers(); + }); + + it("opts in to stdin cancellation and waits for CLI reconciliation", async () => { + const child = Object.assign(new FakeChild(), { stdin: new PassThrough() }); + const spawn = vi.fn(() => child as unknown as ChildProcess); + const runner = createBskRunner("bsk", spawn); + const abort = new AbortController(); + const result = runner.run(["snapshot"], { signal: abort.signal }); + expect(spawn.mock.calls[0]).toEqual([ + "bsk", + ["snapshot", "--json"], + { + windowsHide: true, + env: { ...process.env, BSK_CANCEL_ON_STDIN_CLOSE: "1" }, + }, + ]); + abort.abort(); + expect(child.stdin.writableEnded).toBe(true); + expect(child.killedWith).toEqual([]); + // Closing an already-exited child's pipe must not crash the host. + child.stdin.emit("error", Object.assign(new Error("closed"), { code: "EPIPE" })); + child.finish(2, '{"code":"cancelled"}'); + expect(await result).toMatchObject({ code: 2, aborted: true }); + }); + + it("bounds cancellation of an old or unresponsive CLI", async () => { + vi.useFakeTimers(); + const child = Object.assign(new FakeChild(), { stdin: new PassThrough() }); + const runner = createBskRunner("bsk", fakeSpawn([child])); + const result = runner.run(["snapshot"], { timeoutMs: 5 }); + await vi.advanceTimersByTimeAsync(5); + expect(child.stdin.writableEnded).toBe(true); + expect(child.killedWith).toEqual([]); + await vi.advanceTimersByTimeAsync(10_000); + expect(child.killedWith).toEqual(["SIGKILL"]); + expect(await result).toMatchObject({ timedOut: true }); + }); + + it("cancels only the requested tag and clears the fallback after exit", async () => { + vi.useFakeTimers(); + const a = Object.assign(new FakeChild(), { stdin: new PassThrough() }); + const b = Object.assign(new FakeChild(), { stdin: new PassThrough() }); + const runner = createBskRunner("bsk", fakeSpawn([a, b])); + const first = runner.run(["snapshot"], { tag: "a" }); + const second = runner.run(["snapshot"], { tag: "b" }); + expect(runner.killFor("a")).toBe(1); + expect(a.stdin.writableEnded).toBe(true); + expect(b.stdin.writableEnded).toBe(false); + a.finish(2); + b.finish(0); + await Promise.all([first, second]); + expect(vi.getTimerCount()).toBe(0); + }); + + it("does not start work for an already-aborted call", async () => { + const spawn = vi.fn(); + const runner = createBskRunner("bsk", spawn); + const signal = AbortSignal.abort(); + expect(await runner.run(["snapshot"], { signal })).toMatchObject({ aborted: true }); + expect(spawn).not.toHaveBeenCalled(); + }); +}); + describe("parseBskJson", () => { const base: BskRunResult = { code: 0, stdout: "", stderr: "", timedOut: false, aborted: false }; diff --git a/scripts/install-windows.test.ps1 b/scripts/install-windows.test.ps1 new file mode 100644 index 0000000..4dbe589 --- /dev/null +++ b/scripts/install-windows.test.ps1 @@ -0,0 +1,135 @@ +#Requires -Version 5.1 +param([string]$BskPath = (Join-Path $PSScriptRoot "../target/debug/bsk.exe")) +$ErrorActionPreference = "Stop" + +# Load definitions, never Main: no downloads or changes to the real user PATH. +$installer = Join-Path $PSScriptRoot "../install.ps1" +$tokens = $null +$errors = $null +$ast = [System.Management.Automation.Language.Parser]::ParseFile($installer, [ref]$tokens, [ref]$errors) +if ($errors.Count) { throw ($errors | Out-String) } +foreach ($statement in $ast.EndBlock.Statements) { + if ($statement -is [System.Management.Automation.Language.FunctionDefinitionAst]) { + $definition = $statement.Extent.Text + if ($statement.Name -eq "Add-ToUserPath") { + $definition = $definition.Replace('[Environment]::GetEnvironmentVariable("PATH", "User")', '$script:UserPath') + $definition = $definition.Replace('[Environment]::SetEnvironmentVariable("PATH", $newUserPath, "User")', '$script:UserPath = $newUserPath') + } + Invoke-Expression $definition + } +} +function Assert-Equal($Actual, $Expected) { + if ($Actual -cne $Expected) { throw "expected [$Expected], got [$Actual]" } +} +function Assert-Fails([scriptblock]$Action) { + $failed = $false + try { & $Action } catch { $failed = $true } + if (-not $failed) { throw "expected failure" } +} + +$dir = 'C:\Users\Alice\.local\bin' +foreach ($existing in @('', 'C:\WindowsApps', 'C:\A;C:\B')) { + $script:UserPath = $existing + Add-ToUserPath $dir + $expected = if ($existing) { "$existing;$dir" } else { $dir } + Assert-Equal $script:UserPath $expected + Add-ToUserPath $dir + Assert-Equal $script:UserPath $expected +} + +$root = Join-Path ([IO.Path]::GetTempPath()) ("bsk-install-test-" + [Guid]::NewGuid().ToString('N')) +[IO.Directory]::CreateDirectory($root) | Out-Null +$oldBskHome = $env:BSK_HOME +$oldAutoUpdate = $env:BSK_AUTO_UPDATE +$daemon = $null +try { + $utf8 = New-Object System.Text.UTF8Encoding($false) + $bashRc = Join-Path $root ".bashrc" + $existing = "# existing 中文 configuration" + [char]10 + [IO.File]::WriteAllText($bashRc, $existing, $utf8) + $special = 'C:\Users\张三 space $HOME $(echo unsafe) ! & [x] O''Brien\.local\bin' + Add-ToBashProfile $special $bashRc + $first = [IO.File]::ReadAllText($bashRc) + if (-not $first.StartsWith($existing)) { throw "existing bashrc content changed" } + if (-not $first.Contains('张三')) { throw "Unicode path was lost" } + Add-ToBashProfile $special $bashRc + Assert-Equal ([IO.File]::ReadAllText($bashRc)) $first + $bytes = [IO.File]::ReadAllBytes($bashRc) + if ($bytes[0] -eq 0xEF) { throw "unexpected UTF-8 BOM" } + + $bash = [IO.Path]::GetFullPath((Join-Path (Split-Path (Get-Command git).Source -Parent) "../bin/bash.exe")) + if (-not (Test-Path -LiteralPath $bash)) { throw "Git Bash is required for quoting regression" } + $env:BSK_TEST_RC = $bashRc + $probe = Join-Path $root 'probe.sh' + [IO.File]::WriteAllText($probe, 'source "$BSK_TEST_RC"; printf "%s\n" "${PATH%%:*}"', $utf8) + $actual = & $bash --noprofile --norc $probe + if ($LASTEXITCODE -ne 0) { throw "Git Bash failed" } + Assert-Equal $actual ('/c' + $special.Substring(2).Replace('\', '/')) + + $env:BSK_HOME = Join-Path $root "home" + $env:BSK_AUTO_UPDATE = "off" + [IO.Directory]::CreateDirectory($env:BSK_HOME) | Out-Null + $sourceDir = Join-Path $root "download" + [IO.Directory]::CreateDirectory($sourceDir) | Out-Null + $source = Join-Path $sourceDir "bsk.exe" + [IO.File]::Copy((Resolve-Path -LiteralPath $BskPath).Path, $source) + $targetDir = Join-Path $root "中文 space [x] & install" + [IO.Directory]::CreateDirectory($targetDir) | Out-Null + $target = Join-Path $targetDir "bsk.exe" + Install-Binary $source $target + Assert-Equal (Get-FileHash -LiteralPath $target).Hash (Get-FileHash -LiteralPath $source).Hash + + $daemon = Start-Process -FilePath $target -ArgumentList @('daemon', 'start', '--foreground', '--port', '0') -WindowStyle Hidden -PassThru + $infoPath = Join-Path $env:BSK_HOME "daemon.json" + $deadline = [DateTime]::UtcNow.AddSeconds(15) + while (-not [IO.File]::Exists($infoPath)) { + if ($daemon.HasExited -or [DateTime]::UtcNow -gt $deadline) { throw "daemon failed to start" } + Start-Sleep -Milliseconds 50 + } + # PE overlays distinguish old/new executables without another build. + $stream = [IO.File]::OpenWrite($source) + try { + $stream.Seek(0, [IO.SeekOrigin]::End) | Out-Null + $marker = $utf8.GetBytes("installer replacement regression") + $stream.Write($marker, 0, $marker.Length) + } finally { $stream.Dispose() } + Install-Binary $source $target + if (-not $daemon.WaitForExit(5000)) { throw "old daemon still running" } + Assert-Equal (Get-FileHash -LiteralPath $target).Hash (Get-FileHash -LiteralPath $source).Hash + & $target --version + if ($LASTEXITCODE -ne 0) { throw "replacement is not executable" } + + # Refuse replacement when daemon identity cannot be verified. + [IO.File]::WriteAllText($infoPath, 'invalid daemon metadata') + $before = (Get-FileHash -LiteralPath $target).Hash + Assert-Fails { Install-Binary $source $target } + Assert-Equal (Get-FileHash -LiteralPath $target).Hash $before + [IO.File]::Delete($infoPath) + + # A remaining lock must fail without truncation or staging debris. + $lock = [IO.File]::Open($target, [IO.FileMode]::Open, [IO.FileAccess]::Read, [IO.FileShare]::None) + try { Assert-Fails { Install-Binary $source $target } } finally { $lock.Dispose() } + Assert-Equal (Get-FileHash -LiteralPath $target).Hash $before + if (@(Get-ChildItem -LiteralPath $targetDir -Filter "*.install-*").Count) { throw "staging files leaked" } + Write-Host "Windows installer regressions passed ($($PSVersionTable.PSVersion))" +} +catch { + Write-Host ($_ | Out-String) + Write-Host $_.ScriptStackTrace + throw +} +finally { + if ($daemon -and -not $daemon.HasExited) { + Stop-Process -Id $daemon.Id -Force + $daemon.WaitForExit(5000) | Out-Null + } + $env:BSK_HOME = $oldBskHome + $env:BSK_AUTO_UPDATE = $oldAutoUpdate + Remove-Item Env:BSK_TEST_RC -ErrorAction SilentlyContinue + # Resolve and verify before recursive cleanup; only this fixture is removed. + $resolvedRoot = [IO.Path]::GetFullPath($root) + $tempRoot = [IO.Path]::GetFullPath([IO.Path]::GetTempPath()).TrimEnd('\') + '\' + if (-not $resolvedRoot.StartsWith($tempRoot, [StringComparison]::OrdinalIgnoreCase) -or + -not ([IO.Path]::GetFileName($resolvedRoot)).StartsWith('bsk-install-test-')) { throw "invalid cleanup path" } + Remove-Item -LiteralPath $resolvedRoot -Recurse -Force +} From 0822c53c65bcd9b50eef69dafd7a4199e451d043 Mon Sep 17 00:00:00 2001 From: Zhang GH <5351707+iuyo5678@users.noreply.github.com> Date: Tue, 8 Sep 2026 20:04:24 +0800 Subject: [PATCH 2/2] fix(windows): preserve transfer cleanup after cancellation --- crates/bsk-cli/src/cli/business_rpc.rs | 42 +++- crates/bsk-cli/src/cli/download.rs | 11 +- crates/bsk-cli/src/cli/upload.rs | 99 ++++---- crates/bsk-cli/tests/windows_parent_cancel.rs | 231 ++++++++++++++++-- .../dsh-plugin-browserskill/src/runner.ts | 5 +- .../tests/runner.test.ts | 2 + 6 files changed, 301 insertions(+), 89 deletions(-) diff --git a/crates/bsk-cli/src/cli/business_rpc.rs b/crates/bsk-cli/src/cli/business_rpc.rs index 46c7efe..ff93c04 100644 --- a/crates/bsk-cli/src/cli/business_rpc.rs +++ b/crates/bsk-cli/src/cli/business_rpc.rs @@ -78,7 +78,7 @@ where return Err(CliError::from_rpc(RpcError { code: ErrorCode::Cancelled, message: "parent closed the cancellation pipe".into(), - data: None, + data: Some(serde_json::json!({"reason": "cancelled_before_dispatch"})), })); } let rpc_id: RpcId = format!("{}-{}", rpc_id_prefix, random_short_id()); @@ -120,6 +120,46 @@ where outcome.map_err(CliError::from_rpc) } +/// Release caller-owned staging even after cancellation. This deliberately +/// exposes only transfer.release, not a general bypass for business RPCs. +/// The entire batch, including connection setup, shares one bounded budget. +pub fn release_transfers<'a>( + sock: &Path, + transfer_ids: impl IntoIterator, +) -> Result<(), CliError> { + use bsk_protocol::tools::{TransferIdParams, TransferReleaseResult}; + const CLEANUP_TIMEOUT: Duration = Duration::from_secs(5); + let ids: Vec<_> = transfer_ids.into_iter().collect(); + if ids.is_empty() { + return Ok(()); + } + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .context("build runtime for transfer cleanup")?; + rt.block_on(async { + tokio::time::timeout(CLEANUP_TIMEOUT, async { + let mut client = IpcClient::connect(sock).await?; + for id in ids { + client + .call::<_, TransferReleaseResult>( + "transfer-release", + Method::TransferRelease, + Some(TransferIdParams { + transfer_id: id.to_string(), + }), + CLEANUP_TIMEOUT, + ) + .await? + .map_err(CliError::from_rpc)?; + } + Ok::<_, CliError>(()) + }) + .await + .context("transfer cleanup exceeded its total time budget")? + }) +} + /// Send a `cancel { rpc_id }` frame over a fresh connection so it /// lands on the daemon while the original call is still parked. /// diff --git a/crates/bsk-cli/src/cli/download.rs b/crates/bsk-cli/src/cli/download.rs index 5a97b28..b1ef44d 100644 --- a/crates/bsk-cli/src/cli/download.rs +++ b/crates/bsk-cli/src/cli/download.rs @@ -9,8 +9,7 @@ use anyhow::Context; use base64::Engine; use bsk_protocol::Method; use bsk_protocol::tools::{ - DownloadParams, DownloadResult, TransferChunkParams, TransferChunkResult, TransferIdParams, - TransferReleaseResult, + DownloadParams, DownloadResult, TransferChunkParams, TransferChunkResult, }; use clap::Args; use uuid::Uuid; @@ -70,13 +69,7 @@ pub fn dispatch(args: DownloadArgs, format: Format) -> Result<(), CliError> { CliError::Local(anyhow::anyhow!("daemon returned no download transfer id")) })?; let write_result = write_transfer(&info.sock_path, &transfer_id, &args.out, args.overwrite); - let _: Result = crate::cli::business_rpc::call( - info.sock_path, - "transfer-release", - Method::TransferRelease, - Some(TransferIdParams { transfer_id }), - Duration::from_secs(5), - ); + let _ = crate::cli::business_rpc::release_transfers(&info.sock_path, [transfer_id.as_str()]); write_result?; match format { Format::Json => { diff --git a/crates/bsk-cli/src/cli/upload.rs b/crates/bsk-cli/src/cli/upload.rs index a602732..7249a8d 100644 --- a/crates/bsk-cli/src/cli/upload.rs +++ b/crates/bsk-cli/src/cli/upload.rs @@ -10,8 +10,7 @@ use base64::Engine; use bsk_protocol::Method; use bsk_protocol::tools::{ TransferBeginParams, TransferBeginResult, TransferChunkParams, TransferChunkResult, - TransferIdParams, TransferReadyResult, TransferReleaseResult, UploadFile, UploadMode, - UploadParams, UploadResult, + TransferIdParams, TransferReadyResult, UploadFile, UploadMode, UploadParams, UploadResult, }; use clap::{Args, ValueEnum}; @@ -69,14 +68,12 @@ pub fn dispatch(args: UploadArgs, format: Format) -> Result<(), CliError> { let (ref_, selector) = split_target(args.target, args.ref_, args.selector)?; let mut staged = Vec::new(); for path in &args.files { - match stage_file(&info.sock_path, &args.session, path) { - Ok(file) => staged.push(file), - Err(err) => { - for (id, _) in &staged { - let _ = release(&info.sock_path, id); - } - return Err(err); - } + if let Err(err) = stage_file(&info.sock_path, &args.session, path, &mut staged) { + let _ = crate::cli::business_rpc::release_transfers( + &info.sock_path, + staged.iter().map(|(id, _)| id.as_str()), + ); + return Err(err); } } let params = UploadParams { @@ -102,6 +99,17 @@ pub fn dispatch(args: UploadArgs, format: Format) -> Result<(), CliError> { Some(params), ipc_timeout(args.timeout), ); + // The helper can reject cancellation before handing any request to the + // daemon. In that case staging still belongs to this CLI. + if result.as_ref().is_err_and(|err| { + err.data() + .is_some_and(|data| data["reason"] == "cancelled_before_dispatch") + }) { + let _ = crate::cli::business_rpc::release_transfers( + &info.sock_path, + staged.iter().map(|(id, _)| id.as_str()), + ); + } // Once tool.upload is dispatched, staging ownership belongs to the // session. A transport timeout cannot prove that Chrome did not attach // the file, so releasing here could invalidate a late successful attach. @@ -118,7 +126,12 @@ pub fn dispatch(args: UploadArgs, format: Format) -> Result<(), CliError> { Ok(()) } -fn stage_file(sock: &Path, session: &str, path: &PathBuf) -> Result<(String, String), CliError> { +fn stage_file( + sock: &Path, + session: &str, + path: &PathBuf, + staged: &mut Vec<(String, String)>, +) -> Result<(), CliError> { let mut file = File::open(path) .with_context(|| format!("open upload file {}", path.display())) .map_err(CliError::Local)?; @@ -149,55 +162,39 @@ fn stage_file(sock: &Path, session: &str, path: &PathBuf) -> Result<(String, Str }), Duration::from_secs(10), )?; - let staged = (|| { - let mut offset = 0u64; - let mut buf = vec![0u8; begin.chunk_size as usize]; - loop { - let n = file.read(&mut buf).map_err(|e| CliError::Local(e.into()))?; - if n == 0 { - break; - } - let reply: TransferChunkResult = crate::cli::business_rpc::call( - sock.to_path_buf(), - "transfer-chunk", - Method::TransferChunk, - Some(TransferChunkParams { - transfer_id: begin.transfer_id.clone(), - offset, - data_base64: base64::engine::general_purpose::STANDARD.encode(&buf[..n]), - }), - Duration::from_secs(30), - )?; - offset = reply.next_offset; + // Record ownership immediately so one error path releases both completed + // files and the current partially staged file. + staged.push((begin.transfer_id.clone(), name)); + let mut offset = 0u64; + let mut buf = vec![0u8; begin.chunk_size as usize]; + loop { + let n = file.read(&mut buf).map_err(|e| CliError::Local(e.into()))?; + if n == 0 { + break; } - let _: TransferReadyResult = crate::cli::business_rpc::call( + let reply: TransferChunkResult = crate::cli::business_rpc::call( sock.to_path_buf(), - "transfer-finish", - Method::TransferFinish, - Some(TransferIdParams { + "transfer-chunk", + Method::TransferChunk, + Some(TransferChunkParams { transfer_id: begin.transfer_id.clone(), + offset, + data_base64: base64::engine::general_purpose::STANDARD.encode(&buf[..n]), }), - Duration::from_secs(10), + Duration::from_secs(30), )?; - Ok::<_, CliError>(()) - })(); - if let Err(err) = staged { - let _ = release(sock, &begin.transfer_id); - return Err(err); + offset = reply.next_offset; } - Ok((begin.transfer_id, name)) -} - -fn release(sock: &Path, id: &str) -> Result { - crate::cli::business_rpc::call( + let _: TransferReadyResult = crate::cli::business_rpc::call( sock.to_path_buf(), - "transfer-release", - Method::TransferRelease, + "transfer-finish", + Method::TransferFinish, Some(TransferIdParams { - transfer_id: id.to_string(), + transfer_id: begin.transfer_id.clone(), }), - Duration::from_secs(5), - ) + Duration::from_secs(10), + )?; + Ok(()) } fn ipc_timeout(timeout_ms: u32) -> Duration { diff --git a/crates/bsk-cli/tests/windows_parent_cancel.rs b/crates/bsk-cli/tests/windows_parent_cancel.rs index 44b4452..77d2738 100644 --- a/crates/bsk-cli/tests/windows_parent_cancel.rs +++ b/crates/bsk-cli/tests/windows_parent_cancel.rs @@ -1,10 +1,12 @@ -//! Real Windows CLI processes cancel over IPC when their parent closes stdin. +//! Real Windows CLI cancellation, including caller-owned transfer cleanup. #![cfg(windows)] +use std::path::Path; use std::sync::{Arc, Mutex}; -use std::time::Duration; +use std::time::{Duration, Instant}; -use bsk::daemon::info::DaemonInfo; +use bsk::daemon::{file_transfer::TransferRegistry, info::DaemonInfo}; +use bsk_protocol::tools::{TransferBeginParams, TransferIdParams}; use bsk_protocol::{ErrorCode, Frame, Method, ResponseBody, ResponseFrame, RpcError}; use serde_json::json; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; @@ -12,9 +14,49 @@ use tokio::net::windows::named_pipe::ServerOptions; use tokio::process::Command; use tokio::sync::{Notify, watch}; -#[tokio::test] -async fn stdin_close_cancels_single_and_multi_rpc_commands_but_is_opt_in() { - for mode in ["wait", "upload", "ordinary"] { +fn response(result: Result) -> ResponseBody { + match result { + Ok(value) => ResponseBody::Ok(serde_json::to_value(value).unwrap()), + Err(err) => ResponseBody::Err(err), + } +} + +#[test] +fn stdin_close_cancels_business_but_releases_resources_with_a_bounded_budget() { + // Resolve the production registry's root before starting any runtime threads. + let registry_home = tempfile::tempdir().unwrap(); + let original_home = std::env::var_os("BSK_HOME"); + unsafe { + std::env::set_var("BSK_HOME", registry_home.path()); + } + let transfers = Arc::new(TransferRegistry::new().unwrap()); + unsafe { + match original_home { + Some(value) => std::env::set_var("BSK_HOME", value), + None => std::env::remove_var("BSK_HOME"), + } + } + transfers.initialize().unwrap(); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(exercise( + transfers, + ®istry_home.path().join("run/transfers"), + )); +} + +async fn exercise(transfers: Arc, staging_root: &Path) { + for mode in [ + "wait", + "upload", + "upload-multi", + "upload-before-dispatch", + "download", + "upload-multi-stall", + "ordinary", + ] { let home = tempfile::tempdir().unwrap(); let pipe_name = format!(r"\\.\pipe\bsk-parent-cancel-{}", uuid::Uuid::new_v4()); let info = DaemonInfo::now(std::process::id(), (&pipe_name).into(), 0, "0.2.0"); @@ -29,9 +71,30 @@ async fn stdin_close_cancels_single_and_multi_rpc_commands_but_is_opt_in() { .unwrap(); let registered = Arc::new(Notify::new()); let pending = Arc::new(Mutex::new(None::)); + let ids = Arc::new(Mutex::new(Vec::::new())); + let released = Arc::new(Mutex::new(Vec::::new())); + let received = Arc::new(Mutex::new(Vec::::new())); + let download_id = if mode == "download" { + let staging = transfers.begin_download("fixture").unwrap(); + let browser_dir = home.path().join("BrowserSkill").join(&staging.transfer_id); + std::fs::create_dir_all(&browser_dir).unwrap(); + let file = browser_dir.join("download.txt"); + std::fs::write(&file, b"download fixture").unwrap(); + transfers + .import_download(&staging.transfer_id, &file) + .unwrap(); + ids.lock().unwrap().push(staging.transfer_id.clone()); + Some(staging.transfer_id) + } else { + None + }; let (cancel, cancelled) = watch::channel(false); let ready = Arc::clone(®istered); let name = pipe_name.clone(); + let registry = Arc::clone(&transfers); + let server_ids = Arc::clone(&ids); + let server_released = Arc::clone(&released); + let server_received = Arc::clone(&received); let server = tokio::spawn(async move { let mut connections = tokio::task::JoinSet::new(); loop { @@ -43,6 +106,11 @@ async fn stdin_close_cancels_single_and_multi_rpc_commands_but_is_opt_in() { let cancel = cancel.clone(); let mut cancelled = cancelled.clone(); let name = name.clone(); + let registry = Arc::clone(®istry); + let ids = Arc::clone(&server_ids); + let released = Arc::clone(&server_released); + let received = Arc::clone(&server_received); + let download_id = download_id.clone(); connections.spawn(async move { let (read, mut write) = tokio::io::split(connection); let mut reader = BufReader::new(read); @@ -53,6 +121,8 @@ async fn stdin_close_cancels_single_and_multi_rpc_commands_but_is_opt_in() { panic!("expected request"); }; line.clear(); + received.lock().unwrap().push(req.method.clone()); + let params = req.params.clone().unwrap_or_default(); let body = match req.method { Method::SystemStatus => ResponseBody::Ok(json!({ "daemon_version": "0.2.0", "protocol_version": "1.1", @@ -60,29 +130,69 @@ async fn stdin_close_cancels_single_and_multi_rpc_commands_but_is_opt_in() { "ws_port": 0, "sock_path": name, "browsers": [], "sessions": [], "version_skew_browsers": [] })), - Method::TransferBegin => ResponseBody::Ok(json!({ - "transfer_id": "fixture", "chunk_size": 512 + Method::TransferBegin => { + let begin = registry + .begin_upload(serde_json::from_value(params).unwrap()) + .unwrap(); + ids.lock().unwrap().push(begin.transfer_id.clone()); + response(Ok(begin)) + } + Method::ToolDownload => ResponseBody::Ok(json!({ + "tab_id": 1, "suggested_filename": "download.txt", + "byte_size": 16, "transfer_id": download_id, })), Method::ToolWaitMs if mode == "ordinary" => { ResponseBody::Ok(json!({"waited_ms": 1})) } - Method::ToolWaitMs | Method::TransferChunk => { + Method::TransferChunk + if mode == "upload-before-dispatch" + || (mode.contains("multi") + && ids.lock().unwrap().len() == 1) => + { + response( + registry.write_chunk(serde_json::from_value(params).unwrap()), + ) + } + Method::TransferFinish if mode != "upload-before-dispatch" => response( + registry.finish_upload(serde_json::from_value(params).unwrap()), + ), + Method::ToolWaitMs + | Method::TransferChunk + | Method::TransferRead + | Method::TransferFinish => { *pending.lock().unwrap() = Some(req.id.clone()); ready.notify_one(); cancelled.wait_for(|value| *value).await.unwrap(); - ResponseBody::Err(RpcError { - code: ErrorCode::Cancelled, - message: "parent cancellation reached daemon".into(), - data: None, - }) + if mode == "upload-before-dispatch" { + // Finish won the race with cancellation, but tool.upload must + // not be dispatched and the CLI still owns the staging. + response( + registry + .finish_upload(serde_json::from_value(params).unwrap()), + ) + } else { + ResponseBody::Err(RpcError { + code: ErrorCode::Cancelled, + message: "parent cancellation reached daemon".into(), + data: None, + }) + } } Method::Cancel => { - let target = - req.params.as_ref().unwrap()["rpc_id"].as_str().unwrap(); + let target = params["rpc_id"].as_str().unwrap(); assert_eq!(pending.lock().unwrap().as_deref(), Some(target)); cancel.send(true).unwrap(); ResponseBody::Ok(json!({"cancelled": true})) } + Method::TransferRelease => { + let params: TransferIdParams = + serde_json::from_value(params).unwrap(); + released.lock().unwrap().push(params.transfer_id.clone()); + if mode.contains("stall") { + std::future::pending::<()>().await; + } + response(Ok(registry.release(params))) + } other => panic!("unexpected request after cancellation: {other:?}"), }; let response = Frame::Response(ResponseFrame { id: req.id, body }); @@ -104,18 +214,37 @@ async fn stdin_close_cancels_single_and_multi_rpc_commands_but_is_opt_in() { .stderr(std::process::Stdio::piped()) .creation_flags(0x0800_0000) .kill_on_drop(true); - if mode == "upload" { + if mode.starts_with("upload") { let source = home.path().join("upload.txt"); - std::fs::write(&source, b"upload fixture").unwrap(); + // Declare 300 MiB without allocating or transferring that much data. + std::fs::File::create(&source) + .unwrap() + .set_len(if mode == "upload-before-dispatch" { + 16 + } else { + 300 * 1024 * 1024 + }) + .unwrap(); + cmd.args(["upload", "--session", "fixture", "--selector", "input"]); + if mode.contains("multi") { + let first = home.path().join("first.txt"); + std::fs::write(&first, b"first file").unwrap(); + cmd.arg("--file").arg(first); + } + cmd.arg("--file").arg(source); + } else if mode == "download" { + let out = home.path().join("out.txt"); + std::fs::write(&out, b"existing output").unwrap(); cmd.args([ - "upload", + "download", "--session", "fixture", "--selector", - "input", - "--file", + "a", + "--overwrite", + "--out", ]) - .arg(source); + .arg(out); } else { cmd.args(["wait-ms", if mode == "ordinary" { "1ms" } else { "60s" }]); } @@ -131,7 +260,6 @@ async fn stdin_close_cancels_single_and_multi_rpc_commands_but_is_opt_in() { tokio::time::timeout(Duration::from_secs(10), registered.notified()) .await .expect("CLI registered business RPC"); - // Normal stdin bytes are not a cancellation; only closing the pipe is. child .stdin .as_mut() @@ -141,9 +269,10 @@ async fn stdin_close_cancels_single_and_multi_rpc_commands_but_is_opt_in() { .unwrap(); drop(child.stdin.take()); } - let output = tokio::time::timeout(Duration::from_secs(5), child.wait_with_output()) + let start = Instant::now(); + let output = tokio::time::timeout(Duration::from_secs(8), child.wait_with_output()) .await - .expect("CLI must settle promptly") + .expect("CLI must settle promptly, including cleanup") .unwrap(); server.abort(); let _ = server.await; @@ -157,7 +286,57 @@ async fn stdin_close_cancels_single_and_multi_rpc_commands_but_is_opt_in() { if mode != "ordinary" { let body: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); assert_eq!(body["code"], "cancelled"); - assert_eq!(body["message"], "parent cancellation reached daemon"); + if mode != "upload-before-dispatch" { + assert_eq!(body["message"], "parent cancellation reached daemon"); + } + let received = received.lock().unwrap(); + let cancel = received.iter().position(|m| *m == Method::Cancel).unwrap(); + assert!( + received[cancel + 1..] + .iter() + .all(|m| *m == Method::TransferRelease), + "only cleanup may follow cancellation: {received:?}" + ); + } + let ids = ids.lock().unwrap(); + if mode.contains("stall") { + assert_eq!(released.lock().unwrap().len(), 1); + assert!(start.elapsed() >= Duration::from_secs(4)); + // The CLI cannot force an unresponsive daemon to clean up. + transfers.release_session("fixture"); + } else { + assert_eq!( + *released.lock().unwrap(), + *ids, + "{mode}: every allocation must be released" + ); + for id in ids.iter() { + assert!(!staging_root.join(id).exists(), "staging leaked: {id}"); + } + if mode.starts_with("upload") { + let retry = transfers + .begin_upload(TransferBeginParams { + session_id: "fixture".into(), + name: "retry.txt".into(), + byte_size: 300 * 1024 * 1024, + }) + .expect("same session must have room for another 300 MiB upload"); + transfers.release(TransferIdParams { + transfer_id: retry.transfer_id, + }); + } + } + if mode == "download" { + assert_eq!( + std::fs::read(home.path().join("out.txt")).unwrap(), + b"existing output" + ); + assert!( + !std::fs::read_dir(home.path()) + .unwrap() + .flatten() + .any(|entry| entry.file_name().to_string_lossy().ends_with(".part")) + ); } } } diff --git a/packages/dsh-plugin-browserskill/src/runner.ts b/packages/dsh-plugin-browserskill/src/runner.ts index c80080e..9ec4adf 100644 --- a/packages/dsh-plugin-browserskill/src/runner.ts +++ b/packages/dsh-plugin-browserskill/src/runner.ts @@ -71,8 +71,9 @@ export interface BskRunner { // Business RPCs translate Ctrl-C / opt-in stdin EOF into cancel(rpc_id). // Allow reconciliation before hard-killing an old or unresponsive CLI. const KILL_GRACE_MS = 3000; -// Windows IPC may spend 5s connecting, then 2s cancelling and 2s settling. -const WINDOWS_KILL_GRACE_MS = 10_000; +// Windows IPC may spend 5s connecting, 2s cancelling, 2s settling, +// and up to 5s releasing the entire batch of caller-owned transfers. +const WINDOWS_KILL_GRACE_MS = 15_000; const SESSION_BUSY_RETRY_DELAY_MS = 100; export function createBskRunner(bskPath: string, spawnImpl: SpawnImpl = spawn): BskRunner { diff --git a/packages/dsh-plugin-browserskill/tests/runner.test.ts b/packages/dsh-plugin-browserskill/tests/runner.test.ts index 8ea7af4..219f290 100644 --- a/packages/dsh-plugin-browserskill/tests/runner.test.ts +++ b/packages/dsh-plugin-browserskill/tests/runner.test.ts @@ -133,6 +133,8 @@ describe("Windows parent cancellation", () => { expect(child.stdin.writableEnded).toBe(true); expect(child.killedWith).toEqual([]); await vi.advanceTimersByTimeAsync(10_000); + expect(child.killedWith).toEqual([]); + await vi.advanceTimersByTimeAsync(5_000); expect(child.killedWith).toEqual(["SIGKILL"]); expect(await result).toMatchObject({ timedOut: true }); });