From 12d604bd3231eb041697cc898e30a5c7e6dda8ca Mon Sep 17 00:00:00 2001 From: andrussal Date: Tue, 4 Aug 2026 09:04:36 +0200 Subject: [PATCH] fix(tf): harden async binary resolution --- .../local/src/binary/providers/build.rs | 3 +- .../deployers/local/src/binary/tests.rs | 149 +++++++++++++++++- 2 files changed, 150 insertions(+), 2 deletions(-) diff --git a/testing-framework/deployers/local/src/binary/providers/build.rs b/testing-framework/deployers/local/src/binary/providers/build.rs index 267f86b..465e08d 100644 --- a/testing-framework/deployers/local/src/binary/providers/build.rs +++ b/testing-framework/deployers/local/src/binary/providers/build.rs @@ -86,7 +86,8 @@ impl BuildBinaryProvider { let mut command = Command::new(&self.command.program); command .current_dir(self.workspace_dir()) - .args(&self.command.args); + .args(&self.command.args) + .kill_on_drop(true); command } diff --git a/testing-framework/deployers/local/src/binary/tests.rs b/testing-framework/deployers/local/src/binary/tests.rs index 12d2514..08b4ee6 100644 --- a/testing-framework/deployers/local/src/binary/tests.rs +++ b/testing-framework/deployers/local/src/binary/tests.rs @@ -1,3 +1,5 @@ +#[cfg(unix)] +use std::time::Instant; use std::{ fs, io::{Read as _, Write as _}, @@ -8,6 +10,7 @@ use std::{ atomic::{AtomicUsize, Ordering}, }, thread, + time::Duration, }; use sha2::{Digest as _, Sha256}; @@ -160,6 +163,34 @@ async fn fails_when_build_command_does_not_create_output() { )); } +#[cfg(unix)] +#[tokio::test] +async fn cancelling_build_resolution_stops_the_build_process() { + let temp = TempDir::new().expect("temp dir"); + let output = temp.path().join("built-node"); + let pid_file = temp.path().join("build.pid"); + + let provider = BuildBinaryProvider { + command: BuildCommand::new("sh") + .with_args(["-c", "printf '%s' \"$$\" > build.pid; exec sleep 60"]), + output_path: output, + working_dir: Some(temp.path().to_owned()), + lock_dir: Some(temp.path().join("locks")), + }; + let resolution = tokio::spawn(async move { provider.resolve().await }); + let pid = wait_for_pid(&pid_file).await; + + resolution.abort(); + let error = resolution + .await + .expect_err("resolution should be cancelled"); + assert!(error.is_cancelled()); + assert!( + wait_until_process_exits(pid, Duration::from_secs(2)), + "build process {pid} should exit when resolution is cancelled" + ); +} + #[tokio::test] async fn downloads_binary_from_minimal_http_server() { let temp = TempDir::new().expect("temp dir"); @@ -180,6 +211,53 @@ async fn downloads_binary_from_minimal_http_server() { assert_eq!(fs::read(path).expect("downloaded file"), body); } +#[tokio::test] +async fn concurrent_downloads_share_one_cached_artifact() { + let temp = TempDir::new().expect("temp dir"); + let body = b"downloaded-node"; + + let mut server = SingleResponseServer::start_paused(body); + let url = server.url(); + let cache_dir = temp.path().join("cache"); + let checksum = sha256_hex(body); + + let first_provider = DownloadBinaryProvider { + url: DownloadUrl::Fixed(url.clone()), + sha256: Some(DownloadChecksum::Fixed(checksum.clone())), + cache_dir: Some(cache_dir.clone()), + processor: None, + }; + + let second_provider = DownloadBinaryProvider { + url: DownloadUrl::Fixed(url), + sha256: Some(DownloadChecksum::Fixed(checksum)), + cache_dir: Some(cache_dir), + processor: None, + }; + + let first = tokio::spawn(async move { first_provider.resolve().await }); + server.wait_for_request().await; + + let second = tokio::spawn(async move { second_provider.resolve().await }); + tokio::time::sleep(Duration::from_millis(50)).await; + assert!(!second.is_finished(), "second resolution should wait"); + + server.release_response(); + let (first, second) = tokio::time::timeout(Duration::from_secs(2), async { + (first.await, second.await) + }) + .await + .expect("concurrent resolutions timed out"); + + let first = first.expect("first task failed").expect("first resolution"); + let second = second + .expect("second task failed") + .expect("second resolution"); + + assert_eq!(first, second); + assert_eq!(fs::read(first).expect("downloaded file"), body); +} + #[tokio::test] async fn rejects_download_checksum_mismatch() { let temp = TempDir::new().expect("temp dir"); @@ -265,6 +343,43 @@ fn write_file(path: &Path, contents: &[u8]) { fs::write(path, contents).expect("write file"); } +#[cfg(unix)] +async fn wait_for_pid(path: &Path) -> u32 { + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + loop { + if let Ok(pid) = fs::read_to_string(path) { + return pid.parse().expect("valid process id"); + } + assert!( + tokio::time::Instant::now() < deadline, + "build process did not publish its id" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } +} + +#[cfg(unix)] +fn wait_until_process_exits(pid: u32, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if !process_exists(pid) { + return true; + } + thread::sleep(Duration::from_millis(20)); + } + !process_exists(pid) +} + +#[cfg(unix)] +fn process_exists(pid: u32) -> bool { + std::process::Command::new("kill") + .arg("-0") + .arg(pid.to_string()) + .stderr(std::process::Stdio::null()) + .status() + .is_ok_and(|status| status.success()) +} + fn missing_binary_provider(path: PathBuf) -> BinaryProviderRef { Arc::new(PathBinaryProvider::new(path)) } @@ -309,17 +424,29 @@ fn sha256_hex(bytes: &[u8]) -> String { struct SingleResponseServer { addr: String, + request_started: Option>, + release_response: Option>, } impl SingleResponseServer { fn start(body: &'static [u8]) -> Self { + let mut server = Self::start_paused(body); + server.release_response(); + server + } + + fn start_paused(body: &'static [u8]) -> Self { let listener = TcpListener::bind("127.0.0.1:0").expect("bind test http server"); let addr = listener.local_addr().expect("server addr").to_string(); + let (request_started_tx, request_started_rx) = tokio::sync::oneshot::channel(); + let (release_response_tx, release_response_rx) = std::sync::mpsc::channel(); thread::spawn(move || { let (mut stream, _) = listener.accept().expect("accept one request"); let mut buffer = [0; 1024]; let _ = stream.read(&mut buffer); + let _ = request_started_tx.send(()); + release_response_rx.recv().expect("release response"); let response = format!( "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len() @@ -331,10 +458,30 @@ impl SingleResponseServer { stream.write_all(body).expect("write body"); }); - Self { addr } + Self { + addr, + request_started: Some(request_started_rx), + release_response: Some(release_response_tx), + } } fn url(&self) -> String { format!("http://{}/binary", self.addr) } + + async fn wait_for_request(&mut self) { + self.request_started + .take() + .expect("request receiver") + .await + .expect("request started"); + } + + fn release_response(&mut self) { + self.release_response + .take() + .expect("response sender") + .send(()) + .expect("release response"); + } }