mirror of
https://github.com/Tencent/BrowserSkill.git
synced 2026-09-10 13:01:23 +08:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0822c53c65 | ||
|
|
caea98cd52 | ||
|
|
e8137b42cc |
@@ -38,7 +38,7 @@ jobs:
|
|||||||
run: cargo test --workspace --locked
|
run: cargo test --workspace --locked
|
||||||
|
|
||||||
windows-daemon:
|
windows-daemon:
|
||||||
name: Windows daemon and update regression tests
|
name: Windows daemon, installer and cancellation regression tests
|
||||||
runs-on: windows-latest
|
runs-on: windows-latest
|
||||||
|
|
||||||
steps:
|
steps:
|
||||||
@@ -55,8 +55,16 @@ jobs:
|
|||||||
- name: Run Windows update tests
|
- name: Run Windows update tests
|
||||||
run: cargo test -p bsk --lib --locked cli::update::tests
|
run: cargo test -p bsk --lib --locked cli::update::tests
|
||||||
|
|
||||||
- name: Run Windows update process tests
|
- name: Run Windows update and cancellation process tests
|
||||||
run: cargo test -p bsk --test windows_update --locked
|
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:
|
frontend:
|
||||||
name: Frontend lint, typecheck, tests, build
|
name: Frontend lint, typecheck, tests, build
|
||||||
|
|||||||
@@ -73,6 +73,14 @@ where
|
|||||||
P: Serialize + Send + 'static,
|
P: Serialize + Send + 'static,
|
||||||
R: DeserializeOwned + 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: Some(serde_json::json!({"reason": "cancelled_before_dispatch"})),
|
||||||
|
}));
|
||||||
|
}
|
||||||
let rpc_id: RpcId = format!("{}-{}", rpc_id_prefix, random_short_id());
|
let rpc_id: RpcId = format!("{}-{}", rpc_id_prefix, random_short_id());
|
||||||
let mut client = IpcClient::connect(sock.clone()).await?;
|
let mut client = IpcClient::connect(sock.clone()).await?;
|
||||||
let rpc_id_for_call = rpc_id.clone();
|
let rpc_id_for_call = rpc_id.clone();
|
||||||
@@ -88,16 +96,16 @@ where
|
|||||||
let outcome = tokio::select! {
|
let outcome = tokio::select! {
|
||||||
biased;
|
biased;
|
||||||
res = &mut main_fut => res?,
|
res = &mut main_fut => res?,
|
||||||
sig = wait_for_sigint() => {
|
sig = wait_for_cancel(parent_cancel) => {
|
||||||
sig.context("install SIGINT handler").map_err(CliError::Local)?;
|
sig.context("wait for cancellation").map_err(CliError::Local)?;
|
||||||
debug!(rpc_id = %rpc_id_for_cancel, "SIGINT: forwarding cancel to daemon");
|
debug!(rpc_id = %rpc_id_for_cancel, "forwarding cancel to daemon");
|
||||||
let _ = send_cancel(&sock, &rpc_id_for_cancel).await;
|
let _ = send_cancel(&sock, &rpc_id_for_cancel).await;
|
||||||
match tokio::time::timeout(CANCEL_RESPONSE_GRACE, &mut main_fut).await {
|
match tokio::time::timeout(CANCEL_RESPONSE_GRACE, &mut main_fut).await {
|
||||||
Ok(res) => res?,
|
Ok(res) => res?,
|
||||||
Err(_) => {
|
Err(_) => {
|
||||||
debug!(
|
debug!(
|
||||||
rpc_id = %rpc_id_for_cancel,
|
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 {
|
return Err(CliError::from_rpc(RpcError {
|
||||||
code: ErrorCode::Cancelled,
|
code: ErrorCode::Cancelled,
|
||||||
@@ -112,6 +120,46 @@ where
|
|||||||
outcome.map_err(CliError::from_rpc)
|
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<Item = &'a str>,
|
||||||
|
) -> 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
|
/// Send a `cancel { rpc_id }` frame over a fresh connection so it
|
||||||
/// lands on the daemon while the original call is still parked.
|
/// lands on the daemon while the original call is still parked.
|
||||||
///
|
///
|
||||||
@@ -146,8 +194,57 @@ fn random_short_id() -> String {
|
|||||||
bytes.iter().map(|b| format!("{b:02x}")).collect()
|
bytes.iter().map(|b| format!("{b:02x}")).collect()
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Wait for either Ctrl-C or process SIGTERM. Returning here means
|
/// Opt-in control pipe for parent processes that cannot send Ctrl-C (Node on
|
||||||
/// the user has asked us to stop.
|
/// 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<tokio::sync::watch::Receiver<bool>> {
|
||||||
|
if std::env::var("BSK_CANCEL_ON_STDIN_CLOSE").as_deref() != Ok("1") {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
static CANCEL: std::sync::OnceLock<tokio::sync::watch::Receiver<bool>> =
|
||||||
|
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<tokio::sync::watch::Receiver<bool>>,
|
||||||
|
) -> 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<()> {
|
async fn wait_for_sigint() -> anyhow::Result<()> {
|
||||||
tokio::signal::ctrl_c().await.context("listen for SIGINT")?;
|
tokio::signal::ctrl_c().await.context("listen for SIGINT")?;
|
||||||
Ok(())
|
Ok(())
|
||||||
|
|||||||
@@ -9,8 +9,7 @@ use anyhow::Context;
|
|||||||
use base64::Engine;
|
use base64::Engine;
|
||||||
use bsk_protocol::Method;
|
use bsk_protocol::Method;
|
||||||
use bsk_protocol::tools::{
|
use bsk_protocol::tools::{
|
||||||
DownloadParams, DownloadResult, TransferChunkParams, TransferChunkResult, TransferIdParams,
|
DownloadParams, DownloadResult, TransferChunkParams, TransferChunkResult,
|
||||||
TransferReleaseResult,
|
|
||||||
};
|
};
|
||||||
use clap::Args;
|
use clap::Args;
|
||||||
use uuid::Uuid;
|
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"))
|
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 write_result = write_transfer(&info.sock_path, &transfer_id, &args.out, args.overwrite);
|
||||||
let _: Result<TransferReleaseResult, CliError> = crate::cli::business_rpc::call(
|
let _ = crate::cli::business_rpc::release_transfers(&info.sock_path, [transfer_id.as_str()]);
|
||||||
info.sock_path,
|
|
||||||
"transfer-release",
|
|
||||||
Method::TransferRelease,
|
|
||||||
Some(TransferIdParams { transfer_id }),
|
|
||||||
Duration::from_secs(5),
|
|
||||||
);
|
|
||||||
write_result?;
|
write_result?;
|
||||||
match format {
|
match format {
|
||||||
Format::Json => {
|
Format::Json => {
|
||||||
|
|||||||
@@ -10,8 +10,7 @@ use base64::Engine;
|
|||||||
use bsk_protocol::Method;
|
use bsk_protocol::Method;
|
||||||
use bsk_protocol::tools::{
|
use bsk_protocol::tools::{
|
||||||
TransferBeginParams, TransferBeginResult, TransferChunkParams, TransferChunkResult,
|
TransferBeginParams, TransferBeginResult, TransferChunkParams, TransferChunkResult,
|
||||||
TransferIdParams, TransferReadyResult, TransferReleaseResult, UploadFile, UploadMode,
|
TransferIdParams, TransferReadyResult, UploadFile, UploadMode, UploadParams, UploadResult,
|
||||||
UploadParams, UploadResult,
|
|
||||||
};
|
};
|
||||||
use clap::{Args, ValueEnum};
|
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 (ref_, selector) = split_target(args.target, args.ref_, args.selector)?;
|
||||||
let mut staged = Vec::new();
|
let mut staged = Vec::new();
|
||||||
for path in &args.files {
|
for path in &args.files {
|
||||||
match stage_file(&info.sock_path, &args.session, path) {
|
if let Err(err) = stage_file(&info.sock_path, &args.session, path, &mut staged) {
|
||||||
Ok(file) => staged.push(file),
|
let _ = crate::cli::business_rpc::release_transfers(
|
||||||
Err(err) => {
|
&info.sock_path,
|
||||||
for (id, _) in &staged {
|
staged.iter().map(|(id, _)| id.as_str()),
|
||||||
let _ = release(&info.sock_path, id);
|
);
|
||||||
}
|
return Err(err);
|
||||||
return Err(err);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
let params = UploadParams {
|
let params = UploadParams {
|
||||||
@@ -102,6 +99,17 @@ pub fn dispatch(args: UploadArgs, format: Format) -> Result<(), CliError> {
|
|||||||
Some(params),
|
Some(params),
|
||||||
ipc_timeout(args.timeout),
|
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
|
// Once tool.upload is dispatched, staging ownership belongs to the
|
||||||
// session. A transport timeout cannot prove that Chrome did not attach
|
// session. A transport timeout cannot prove that Chrome did not attach
|
||||||
// the file, so releasing here could invalidate a late successful 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(())
|
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)
|
let mut file = File::open(path)
|
||||||
.with_context(|| format!("open upload file {}", path.display()))
|
.with_context(|| format!("open upload file {}", path.display()))
|
||||||
.map_err(CliError::Local)?;
|
.map_err(CliError::Local)?;
|
||||||
@@ -149,55 +162,39 @@ fn stage_file(sock: &Path, session: &str, path: &PathBuf) -> Result<(String, Str
|
|||||||
}),
|
}),
|
||||||
Duration::from_secs(10),
|
Duration::from_secs(10),
|
||||||
)?;
|
)?;
|
||||||
let staged = (|| {
|
// Record ownership immediately so one error path releases both completed
|
||||||
let mut offset = 0u64;
|
// files and the current partially staged file.
|
||||||
let mut buf = vec![0u8; begin.chunk_size as usize];
|
staged.push((begin.transfer_id.clone(), name));
|
||||||
loop {
|
let mut offset = 0u64;
|
||||||
let n = file.read(&mut buf).map_err(|e| CliError::Local(e.into()))?;
|
let mut buf = vec![0u8; begin.chunk_size as usize];
|
||||||
if n == 0 {
|
loop {
|
||||||
break;
|
let n = file.read(&mut buf).map_err(|e| CliError::Local(e.into()))?;
|
||||||
}
|
if n == 0 {
|
||||||
let reply: TransferChunkResult = crate::cli::business_rpc::call(
|
break;
|
||||||
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;
|
|
||||||
}
|
}
|
||||||
let _: TransferReadyResult = crate::cli::business_rpc::call(
|
let reply: TransferChunkResult = crate::cli::business_rpc::call(
|
||||||
sock.to_path_buf(),
|
sock.to_path_buf(),
|
||||||
"transfer-finish",
|
"transfer-chunk",
|
||||||
Method::TransferFinish,
|
Method::TransferChunk,
|
||||||
Some(TransferIdParams {
|
Some(TransferChunkParams {
|
||||||
transfer_id: begin.transfer_id.clone(),
|
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>(())
|
offset = reply.next_offset;
|
||||||
})();
|
|
||||||
if let Err(err) = staged {
|
|
||||||
let _ = release(sock, &begin.transfer_id);
|
|
||||||
return Err(err);
|
|
||||||
}
|
}
|
||||||
Ok((begin.transfer_id, name))
|
let _: TransferReadyResult = crate::cli::business_rpc::call(
|
||||||
}
|
|
||||||
|
|
||||||
fn release(sock: &Path, id: &str) -> Result<TransferReleaseResult, CliError> {
|
|
||||||
crate::cli::business_rpc::call(
|
|
||||||
sock.to_path_buf(),
|
sock.to_path_buf(),
|
||||||
"transfer-release",
|
"transfer-finish",
|
||||||
Method::TransferRelease,
|
Method::TransferFinish,
|
||||||
Some(TransferIdParams {
|
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 {
|
fn ipc_timeout(timeout_ms: u32) -> Duration {
|
||||||
|
|||||||
@@ -0,0 +1,342 @@
|
|||||||
|
//! Real Windows CLI cancellation, including caller-owned transfer cleanup.
|
||||||
|
#![cfg(windows)]
|
||||||
|
|
||||||
|
use std::path::Path;
|
||||||
|
use std::sync::{Arc, Mutex};
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
|
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};
|
||||||
|
use tokio::net::windows::named_pipe::ServerOptions;
|
||||||
|
use tokio::process::Command;
|
||||||
|
use tokio::sync::{Notify, watch};
|
||||||
|
|
||||||
|
fn response<T: serde::Serialize>(result: Result<T, RpcError>) -> 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<TransferRegistry>, 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");
|
||||||
|
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::<String>));
|
||||||
|
let ids = Arc::new(Mutex::new(Vec::<String>::new()));
|
||||||
|
let released = Arc::new(Mutex::new(Vec::<String>::new()));
|
||||||
|
let received = Arc::new(Mutex::new(Vec::<Method>::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 {
|
||||||
|
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();
|
||||||
|
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);
|
||||||
|
let mut line = String::new();
|
||||||
|
while reader.read_line(&mut line).await.unwrap_or(0) > 0 {
|
||||||
|
let Frame::Request(req) = serde_json::from_str::<Frame>(&line).unwrap()
|
||||||
|
else {
|
||||||
|
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",
|
||||||
|
"pid": std::process::id(), "uptime_secs": 0,
|
||||||
|
"ws_port": 0, "sock_path": name,
|
||||||
|
"browsers": [], "sessions": [], "version_skew_browsers": []
|
||||||
|
})),
|
||||||
|
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::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();
|
||||||
|
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 = 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 });
|
||||||
|
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.starts_with("upload") {
|
||||||
|
let source = home.path().join("upload.txt");
|
||||||
|
// 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([
|
||||||
|
"download",
|
||||||
|
"--session",
|
||||||
|
"fixture",
|
||||||
|
"--selector",
|
||||||
|
"a",
|
||||||
|
"--overwrite",
|
||||||
|
"--out",
|
||||||
|
])
|
||||||
|
.arg(out);
|
||||||
|
} 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");
|
||||||
|
child
|
||||||
|
.stdin
|
||||||
|
.as_mut()
|
||||||
|
.unwrap()
|
||||||
|
.write_all(b"still connected")
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
drop(child.stdin.take());
|
||||||
|
}
|
||||||
|
let start = Instant::now();
|
||||||
|
let output = tokio::time::timeout(Duration::from_secs(8), child.wait_with_output())
|
||||||
|
.await
|
||||||
|
.expect("CLI must settle promptly, including cleanup")
|
||||||
|
.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");
|
||||||
|
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"))
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
+41
-15
@@ -21,6 +21,7 @@ $ErrorActionPreference = "Stop"
|
|||||||
|
|
||||||
$Repo = if ($env:BSK_REPO) { $env:BSK_REPO } else { "Tencent/BrowserSkill" }
|
$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 = 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}"
|
$GitHub = "https://github.com/${Repo}"
|
||||||
|
|
||||||
function Write-Log {
|
function Write-Log {
|
||||||
@@ -64,7 +65,7 @@ function Get-PlatformTriple {
|
|||||||
function Add-ToUserPath {
|
function Add-ToUserPath {
|
||||||
param([string]$Dir)
|
param([string]$Dir)
|
||||||
|
|
||||||
$currentUserPath = [Environment]::GetEnvironmentVariable("PATH", "User") -split ";" | Where-Object { $_ }
|
$currentUserPath = @([Environment]::GetEnvironmentVariable("PATH", "User") -split ";" | Where-Object { $_ })
|
||||||
|
|
||||||
if ($currentUserPath -contains $Dir) {
|
if ($currentUserPath -contains $Dir) {
|
||||||
Write-Log "$Dir is already in your user PATH"
|
Write-Log "$Dir is already in your user PATH"
|
||||||
@@ -90,30 +91,59 @@ function Add-ToSessionPath {
|
|||||||
# ── Git Bash (bash environment) PATH helper ──────────────────────────────────
|
# ── Git Bash (bash environment) PATH helper ──────────────────────────────────
|
||||||
|
|
||||||
function Add-ToBashProfile {
|
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)
|
# Convert Windows path (e.g. C:\Users\foo\.local\bin) to Git-Bash Unix-style (/c/Users/foo/.local/bin)
|
||||||
$unixPath = $Dir -replace '\\', '/'
|
$unixPath = $Dir -replace '\\', '/'
|
||||||
if ($unixPath -match '^([A-Z]):(.*)$') {
|
if ($unixPath -match '^([A-Z]):(.*)$') {
|
||||||
$unixPath = '/' + $matches[1].ToLower() + $matches[2]
|
$unixPath = '/' + $matches[1].ToLower() + $matches[2]
|
||||||
}
|
}
|
||||||
$bashRc = Join-Path $HOME ".bashrc"
|
# Single-quote the literal directory; only the existing PATH is expanded.
|
||||||
$exportLine = "export PATH=""${unixPath}:`$PATH"" # bsk CLI"
|
$shellQuote = "'" + [char]34 + "'" + [char]34 + "'"
|
||||||
|
$quotedPath = "'" + $unixPath.Replace("'", $shellQuote) + "'"
|
||||||
|
$exportLine = "export PATH=${quotedPath}:`"`$PATH`" # bsk CLI"
|
||||||
|
|
||||||
if (Test-Path $bashRc) {
|
if (Test-Path -LiteralPath $BashRc) {
|
||||||
$content = Get-Content $bashRc -Raw -ErrorAction SilentlyContinue
|
$content = [System.IO.File]::ReadAllText($BashRc)
|
||||||
if ($content -match [regex]::Escape($unixPath)) {
|
if ($content.Contains($exportLine)) {
|
||||||
Write-Log "$unixPath is already in ~/.bashrc"
|
Write-Log "$unixPath is already in ~/.bashrc"
|
||||||
return
|
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"
|
Write-Log "added ${unixPath} to ~/.bashrc"
|
||||||
}
|
}
|
||||||
|
|
||||||
# ── Main ──────────────────────────────────────────────────────────────────────
|
# ── 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 {
|
function Main {
|
||||||
$platform = Get-PlatformTriple
|
$platform = Get-PlatformTriple
|
||||||
|
|
||||||
@@ -184,7 +214,7 @@ function Main {
|
|||||||
New-Item -ItemType Directory -Path $InstallDir -Force | Out-Null
|
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"
|
Write-Log "installed bsk to $InstallDir\bsk.exe"
|
||||||
|
|
||||||
@@ -199,12 +229,8 @@ function Main {
|
|||||||
|
|
||||||
# Verify
|
# Verify
|
||||||
$bskPath = Join-Path $InstallDir "bsk.exe"
|
$bskPath = Join-Path $InstallDir "bsk.exe"
|
||||||
if (Get-Command bsk -ErrorAction SilentlyContinue) {
|
& $bskPath --version
|
||||||
& bsk --version
|
if ($LASTEXITCODE -ne 0) { throw "installed bsk failed verification" }
|
||||||
}
|
|
||||||
else {
|
|
||||||
Write-Log "verify install: & ""$bskPath"" --version"
|
|
||||||
}
|
|
||||||
|
|
||||||
Write-Log "done"
|
Write-Log "done"
|
||||||
Write-Host ""
|
Write-Host ""
|
||||||
|
|||||||
@@ -5,7 +5,7 @@
|
|||||||
* map the CLI's JSON error envelope onto a thrown `BskError`.
|
* 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. */
|
/** Shape of the JSON error envelope `bsk --json` prints on failure. */
|
||||||
export interface BskErrorBody {
|
export interface BskErrorBody {
|
||||||
@@ -53,7 +53,11 @@ export interface BskRunOptions {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/** Minimal spawn signature so tests can substitute a fake child process. */
|
/** 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 {
|
export interface BskRunner {
|
||||||
/** Run `bsk <args...> --json` and collect its output. */
|
/** Run `bsk <args...> --json` and collect its output. */
|
||||||
@@ -64,29 +68,66 @@ export interface BskRunner {
|
|||||||
killFor(tag: string): number;
|
killFor(tag: string): number;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Business RPCs translate SIGINT into the daemon's cancel(rpc_id) protocol.
|
// Business RPCs translate Ctrl-C / opt-in stdin EOF into cancel(rpc_id).
|
||||||
// Give that bounded reconciliation path time to settle before the hard kill.
|
// Allow reconciliation before hard-killing an old or unresponsive CLI.
|
||||||
const KILL_GRACE_MS = 3000;
|
const KILL_GRACE_MS = 3000;
|
||||||
|
// 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;
|
const SESSION_BUSY_RETRY_DELAY_MS = 100;
|
||||||
|
|
||||||
export function createBskRunner(bskPath: string, spawnImpl: SpawnImpl = spawn): BskRunner {
|
export function createBskRunner(bskPath: string, spawnImpl: SpawnImpl = spawn): BskRunner {
|
||||||
const live = new Map<ChildProcess, string | undefined>();
|
const live = new Map<ChildProcess, string | undefined>();
|
||||||
|
const windows = process.platform === "win32";
|
||||||
|
const cancelling = new Set<ChildProcess>();
|
||||||
|
|
||||||
function killChild(child: ChildProcess): void {
|
function killChild(child: ChildProcess): void {
|
||||||
if (child.exitCode !== null || child.signalCode !== null) return;
|
if (child.exitCode !== null || child.signalCode !== null || cancelling.has(child)) return;
|
||||||
child.kill("SIGINT");
|
cancelling.add(child);
|
||||||
const force = setTimeout(() => {
|
// Node kills Windows children outright for SIGINT. EOF asks the CLI to
|
||||||
if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL");
|
// send its existing cancel RPC and wait for browser reconciliation.
|
||||||
}, KILL_GRACE_MS);
|
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();
|
force.unref();
|
||||||
|
child.once("close", () => {
|
||||||
|
clearTimeout(force);
|
||||||
|
cancelling.delete(child);
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
return {
|
return {
|
||||||
run(args, options = {}) {
|
run(args, options = {}) {
|
||||||
|
if (options.signal?.aborted) {
|
||||||
|
return Promise.resolve({
|
||||||
|
code: null,
|
||||||
|
stdout: "",
|
||||||
|
stderr: "",
|
||||||
|
timedOut: false,
|
||||||
|
aborted: true,
|
||||||
|
});
|
||||||
|
}
|
||||||
return new Promise<BskRunResult>((resolve, reject) => {
|
return new Promise<BskRunResult>((resolve, reject) => {
|
||||||
let child: ChildProcess;
|
let child: ChildProcess;
|
||||||
try {
|
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) {
|
} catch (error) {
|
||||||
reject(error);
|
reject(error);
|
||||||
return;
|
return;
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
import type { ChildProcess } from "node:child_process";
|
import type { ChildProcess } from "node:child_process";
|
||||||
import { EventEmitter } from "node:events";
|
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 {
|
import {
|
||||||
type BskRunResult,
|
type BskRunResult,
|
||||||
createBskRunner,
|
createBskRunner,
|
||||||
@@ -92,6 +93,77 @@ 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([]);
|
||||||
|
await vi.advanceTimersByTimeAsync(5_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", () => {
|
describe("parseBskJson", () => {
|
||||||
const base: BskRunResult = { code: 0, stdout: "", stderr: "", timedOut: false, aborted: false };
|
const base: BskRunResult = { code: 0, stdout: "", stderr: "", timedOut: false, aborted: false };
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user