From 9ecca30ef32f0e646fa9e7b898d035fbd687c257 Mon Sep 17 00:00:00 2001 From: andrussal Date: Mon, 3 Aug 2026 17:30:58 +0200 Subject: [PATCH 1/2] fix(tf): resolve local binaries asynchronously --- Cargo.lock | 16 +--- examples/multi_app/fixture/src/lib.rs | 2 +- testing-framework/app/src/process.rs | 2 +- testing-framework/deployers/local/Cargo.toml | 2 +- .../deployers/local/src/binary/lock.rs | 7 +- .../deployers/local/src/binary/mod.rs | 12 +-- .../local/src/binary/providers/build.rs | 13 +-- .../local/src/binary/providers/download.rs | 15 ++-- .../local/src/binary/providers/env.rs | 4 +- .../local/src/binary/providers/fallback.rs | 7 +- .../local/src/binary/providers/path.rs | 4 +- .../deployers/local/src/binary/tests.rs | 81 ++++++++++++------- .../deployers/local/src/env/helpers.rs | 35 ++++---- .../deployers/local/src/env/mod.rs | 11 ++- .../deployers/local/src/env/tests.rs | 2 +- .../deployers/local/src/node_control/mod.rs | 1 + .../deployers/local/src/process.rs | 21 +++-- .../deployers/local/src/provisioner.rs | 2 +- 18 files changed, 142 insertions(+), 95 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c9d44c3..1f9fddd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -589,15 +589,6 @@ dependencies = [ "tokio-util", ] -[[package]] -name = "concurrent-queue" -version = "2.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973" -dependencies = [ - "crossbeam-utils", -] - [[package]] name = "const-oid" version = "0.9.6" @@ -882,11 +873,10 @@ dependencies = [ [[package]] name = "event-listener" -version = "5.4.1" +version = "5.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e13b66accf52311f30a0db42147dadea9850cb48cd070028831ae5f5d4b856ab" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" dependencies = [ - "concurrent-queue", "parking", "pin-project-lite", ] @@ -3015,9 +3005,7 @@ checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" dependencies = [ "base64", "bytes", - "futures-channel", "futures-core", - "futures-util", "http", "http-body", "http-body-util", diff --git a/examples/multi_app/fixture/src/lib.rs b/examples/multi_app/fixture/src/lib.rs index f07e56e..c2b3812 100644 --- a/examples/multi_app/fixture/src/lib.rs +++ b/examples/multi_app/fixture/src/lib.rs @@ -123,7 +123,7 @@ impl AppDeployment for JobWorkerApp { let health_port = allocate_available_port()?; let client = WorkerClient::new(health_port)?; let launch = LaunchSpec { - binary: worker_binary_provider().resolve()?, + binary: worker_binary_provider().resolve().await?, args: vec![ "--queue-url".to_owned(), self.queue_url.to_string(), diff --git a/testing-framework/app/src/process.rs b/testing-framework/app/src/process.rs index 99e4346..13f434a 100644 --- a/testing-framework/app/src/process.rs +++ b/testing-framework/app/src/process.rs @@ -123,7 +123,7 @@ where let mut process = ProcessNode::spawn( &label, (), - move |(), _working_dir, _label| Ok(launch), + move |(), _working_dir, _label| Box::pin(async move { Ok(launch) }), move |()| Ok(endpoints), keep_tempdir, persist_dir.as_deref(), diff --git a/testing-framework/deployers/local/Cargo.toml b/testing-framework/deployers/local/Cargo.toml index b83b792..f47bf5b 100644 --- a/testing-framework/deployers/local/Cargo.toml +++ b/testing-framework/deployers/local/Cargo.toml @@ -15,7 +15,7 @@ workspace = true [dependencies] async-trait = "0.1" fs_extra = "1.3" -reqwest = { features = ["blocking", "rustls-tls"], workspace = true } +reqwest = { features = ["rustls-tls"], workspace = true } serde = { workspace = true } serde_yaml = { workspace = true } sha2 = "0.10" diff --git a/testing-framework/deployers/local/src/binary/lock.rs b/testing-framework/deployers/local/src/binary/lock.rs index 220da35..abc20f0 100644 --- a/testing-framework/deployers/local/src/binary/lock.rs +++ b/testing-framework/deployers/local/src/binary/lock.rs @@ -7,10 +7,11 @@ use std::{ fs, io, path::{Path, PathBuf}, - thread, time::{Duration, Instant}, }; +use tokio::time::sleep; + use super::types::BinaryProviderError; const LOCK_RETRY_DELAY: Duration = Duration::from_millis(200); @@ -23,7 +24,7 @@ pub(super) struct BinaryProviderLock { } impl BinaryProviderLock { - pub(super) fn acquire(path: &Path) -> Result { + pub(super) async fn acquire(path: &Path) -> Result { if let Some(parent) = path.parent() { fs::create_dir_all(parent).map_err(|source| BinaryProviderError::Io { path: parent.to_owned(), @@ -50,7 +51,7 @@ impl BinaryProviderLock { }); } - thread::sleep(LOCK_RETRY_DELAY); + sleep(LOCK_RETRY_DELAY).await; } Err(source) => { return Err(BinaryProviderError::Io { diff --git a/testing-framework/deployers/local/src/binary/mod.rs b/testing-framework/deployers/local/src/binary/mod.rs index d8111f6..60d5074 100644 --- a/testing-framework/deployers/local/src/binary/mod.rs +++ b/testing-framework/deployers/local/src/binary/mod.rs @@ -13,6 +13,7 @@ mod types; use std::path::PathBuf; +use async_trait::async_trait; use cache::BinaryCache; pub(super) use types::optional_path_display; pub use types::{ @@ -28,8 +29,9 @@ pub use types::{ /// binary in the current environment. The default [`resolve`](Self::resolve) /// method turns that into a launch error, while [`FallbackBinaryProvider`] uses /// it to try several providers in order. +#[async_trait] pub trait BinaryProvider: Send + Sync { - fn try_resolve(&self) -> Result, BinaryProviderError>; + async fn try_resolve(&self) -> Result, BinaryProviderError>; fn display(&self) -> String; @@ -40,21 +42,21 @@ pub trait BinaryProvider: Send + Sync { /// Resolution is cached per process so repeated node starts using the same /// provider config do not rebuild, redownload, or rediscover the same /// binary. - fn resolve(&self) -> Result { + async fn resolve(&self) -> Result { let cache_key = self.cache_key(); if let Some(path) = BinaryCache::get(&cache_key) { return Ok(path); } - let path = self.resolve_uncached()?; + let path = self.resolve_uncached().await?; BinaryCache::insert(cache_key, path.clone()); Ok(path) } - fn resolve_uncached(&self) -> Result { - if let Some(path) = self.try_resolve()? { + async fn resolve_uncached(&self) -> Result { + if let Some(path) = self.try_resolve().await? { return Ok(path); } diff --git a/testing-framework/deployers/local/src/binary/providers/build.rs b/testing-framework/deployers/local/src/binary/providers/build.rs index 8c153ef..267f86b 100644 --- a/testing-framework/deployers/local/src/binary/providers/build.rs +++ b/testing-framework/deployers/local/src/binary/providers/build.rs @@ -8,10 +8,11 @@ use std::{ env, path::{Path, PathBuf}, - process::Command, }; +use async_trait::async_trait; use sha2::{Digest as _, Sha256}; +use tokio::process::Command; use tracing::info; use crate::binary::{ @@ -19,12 +20,13 @@ use crate::binary::{ optional_path_display, }; +#[async_trait] impl BinaryProvider for BuildBinaryProvider { - fn try_resolve(&self) -> Result, BinaryProviderError> { + async fn try_resolve(&self) -> Result, BinaryProviderError> { let output_path = self.output_path(); - let _lock = BinaryProviderLock::acquire(&self.lock_path())?; + let _lock = BinaryProviderLock::acquire(&self.lock_path()).await?; - self.run_build()?; + self.run_build().await?; self.ensure_output_exists(&output_path)?; Ok(Some(output_path)) @@ -45,7 +47,7 @@ impl BinaryProvider for BuildBinaryProvider { } impl BuildBinaryProvider { - fn run_build(&self) -> Result<(), BinaryProviderError> { + async fn run_build(&self) -> Result<(), BinaryProviderError> { info!( command = self.command.display(), workspace = %self.workspace_dir().display(), @@ -55,6 +57,7 @@ impl BuildBinaryProvider { let status = self .command() .status() + .await .map_err(|source| BinaryProviderError::Io { path: self.workspace_dir(), source, diff --git a/testing-framework/deployers/local/src/binary/providers/download.rs b/testing-framework/deployers/local/src/binary/providers/download.rs index c24825c..1c06304 100644 --- a/testing-framework/deployers/local/src/binary/providers/download.rs +++ b/testing-framework/deployers/local/src/binary/providers/download.rs @@ -12,7 +12,7 @@ use std::{ path::{Path, PathBuf}, }; -use reqwest::blocking; +use async_trait::async_trait; use sha2::{Digest as _, Sha256}; use tracing::info; @@ -21,17 +21,18 @@ use crate::binary::{ lock::BinaryProviderLock, optional_path_display, }; +#[async_trait] impl BinaryProvider for DownloadBinaryProvider { - fn try_resolve(&self) -> Result, BinaryProviderError> { + async fn try_resolve(&self) -> Result, BinaryProviderError> { let url = self.url.resolve()?; let path = self.cached_binary_path(&url)?; - let _lock = BinaryProviderLock::acquire(&self.lock_path(&url))?; + let _lock = BinaryProviderLock::acquire(&self.lock_path(&url)).await?; if path.is_file() { return Ok(Some(path)); } - let bytes = self.download_bytes(&url)?; + let bytes = self.download_bytes(&url).await?; self.verify_checksum(&path, &bytes)?; self.prepare_binary(&path, &bytes)?; @@ -86,10 +87,11 @@ impl DownloadBinaryProvider { Ok(cache_dir.join(self.download_file_name(url))) } - fn download_bytes(&self, url: &str) -> Result, BinaryProviderError> { + async fn download_bytes(&self, url: &str) -> Result, BinaryProviderError> { info!(url, "downloading binary"); - blocking::get(url) + reqwest::get(url) + .await .map_err(|source| BinaryProviderError::Download { url: url.to_owned(), source, @@ -100,6 +102,7 @@ impl DownloadBinaryProvider { source, })? .bytes() + .await .map(|bytes| bytes.to_vec()) .map_err(|source| BinaryProviderError::Download { url: url.to_owned(), diff --git a/testing-framework/deployers/local/src/binary/providers/env.rs b/testing-framework/deployers/local/src/binary/providers/env.rs index d2d279b..b613a59 100644 --- a/testing-framework/deployers/local/src/binary/providers/env.rs +++ b/testing-framework/deployers/local/src/binary/providers/env.rs @@ -6,12 +6,14 @@ use std::{env, path::PathBuf}; +use async_trait::async_trait; use tracing::{debug, info}; use crate::binary::{BinaryProvider, BinaryProviderError, EnvBinaryProvider}; +#[async_trait] impl BinaryProvider for EnvBinaryProvider { - fn try_resolve(&self) -> Result, BinaryProviderError> { + async fn try_resolve(&self) -> Result, BinaryProviderError> { let Some(path) = env::var_os(&self.env_var).map(PathBuf::from) else { return Ok(None); }; diff --git a/testing-framework/deployers/local/src/binary/providers/fallback.rs b/testing-framework/deployers/local/src/binary/providers/fallback.rs index fe2b2e3..895bf66 100644 --- a/testing-framework/deployers/local/src/binary/providers/fallback.rs +++ b/testing-framework/deployers/local/src/binary/providers/fallback.rs @@ -6,12 +6,15 @@ use std::path::PathBuf; +use async_trait::async_trait; + use crate::binary::{BinaryProvider, BinaryProviderError, FallbackBinaryProvider}; +#[async_trait] impl BinaryProvider for FallbackBinaryProvider { - fn try_resolve(&self) -> Result, BinaryProviderError> { + async fn try_resolve(&self) -> Result, BinaryProviderError> { for provider in &self.providers { - match provider.resolve() { + match provider.resolve().await { Ok(path) => return Ok(Some(path)), Err(BinaryProviderError::NotFound { .. }) => continue, Err(error) => return Err(error), diff --git a/testing-framework/deployers/local/src/binary/providers/path.rs b/testing-framework/deployers/local/src/binary/providers/path.rs index ef5463e..eb5b902 100644 --- a/testing-framework/deployers/local/src/binary/providers/path.rs +++ b/testing-framework/deployers/local/src/binary/providers/path.rs @@ -6,12 +6,14 @@ use std::path::PathBuf; +use async_trait::async_trait; use tracing::info; use crate::binary::{BinaryProvider, BinaryProviderError, PathBinaryProvider}; +#[async_trait] impl BinaryProvider for PathBinaryProvider { - fn try_resolve(&self) -> Result, BinaryProviderError> { + async fn try_resolve(&self) -> Result, BinaryProviderError> { if !self.path.is_absolute() { return Err(BinaryProviderError::RelativePath { path: self.path.clone(), diff --git a/testing-framework/deployers/local/src/binary/tests.rs b/testing-framework/deployers/local/src/binary/tests.rs index 15c1a26..12d2514 100644 --- a/testing-framework/deployers/local/src/binary/tests.rs +++ b/testing-framework/deployers/local/src/binary/tests.rs @@ -19,30 +19,32 @@ use super::{ PathBinaryProvider, }; -#[test] -fn resolves_configured_absolute_path() { +#[tokio::test] +async fn resolves_configured_absolute_path() { let temp = TempDir::new().expect("temp dir"); let binary = temp.path().join("node"); write_file(&binary, b"binary"); let path = PathBinaryProvider::new(&binary) .resolve() + .await .expect("path provider resolves"); assert_eq!(path, binary); } -#[test] -fn rejects_relative_configured_path() { +#[tokio::test] +async fn rejects_relative_configured_path() { let error = PathBinaryProvider::new("relative-node") .resolve() + .await .expect_err("relative path is rejected"); assert!(matches!(error, BinaryProviderError::RelativePath { .. })); } -#[test] -fn resolves_first_available_fallback_provider() { +#[tokio::test] +async fn resolves_first_available_fallback_provider() { let temp = TempDir::new().expect("temp dir"); let binary = temp.path().join("node"); write_file(&binary, b"binary"); @@ -52,13 +54,16 @@ fn resolves_first_available_fallback_provider() { Arc::new(PathBinaryProvider::new(&binary)), ]; let provider = FallbackBinaryProvider::new(providers); - let path = provider.resolve().expect("fallback provider resolves"); + let path = provider + .resolve() + .await + .expect("fallback provider resolves"); assert_eq!(path, binary); } -#[test] -fn fallback_reuses_inner_provider_cache() { +#[tokio::test] +async fn fallback_reuses_inner_provider_cache() { let temp = TempDir::new().expect("temp dir"); let binary = temp.path().join("node"); write_file(&binary, b"binary"); @@ -77,13 +82,19 @@ fn fallback_reuses_inner_provider_cache() { cached_provider, ]); - assert_eq!(first.resolve().expect("first fallback resolves"), binary); - assert_eq!(second.resolve().expect("second fallback resolves"), binary); + assert_eq!( + first.resolve().await.expect("first fallback resolves"), + binary + ); + assert_eq!( + second.resolve().await.expect("second fallback resolves"), + binary + ); assert_eq!(resolve_count.load(Ordering::SeqCst), 1); } -#[test] -fn runs_build_command_and_returns_output_path() { +#[tokio::test] +async fn runs_build_command_and_returns_output_path() { let temp = TempDir::new().expect("temp dir"); let output = temp.path().join("built-node"); let script = temp.path().join("build.sh"); @@ -98,14 +109,14 @@ fn runs_build_command_and_returns_output_path() { working_dir: Some(temp.path().to_owned()), lock_dir: Some(temp.path().join("locks")), }; - let path = provider.resolve().expect("build provider resolves"); + let path = provider.resolve().await.expect("build provider resolves"); assert_eq!(path, output); assert_eq!(fs::read(path).expect("built file"), b"built"); } -#[test] -fn build_provider_runs_even_when_output_exists() { +#[tokio::test] +async fn build_provider_runs_even_when_output_exists() { let temp = TempDir::new().expect("temp dir"); let output = temp.path().join("built-node"); let script = temp.path().join("build.sh"); @@ -121,14 +132,14 @@ fn build_provider_runs_even_when_output_exists() { working_dir: Some(temp.path().to_owned()), lock_dir: Some(temp.path().join("locks")), }; - let path = provider.resolve().expect("build provider resolves"); + let path = provider.resolve().await.expect("build provider resolves"); assert_eq!(path, output); assert_eq!(fs::read(path).expect("built file"), b"new"); } -#[test] -fn fails_when_build_command_does_not_create_output() { +#[tokio::test] +async fn fails_when_build_command_does_not_create_output() { let temp = TempDir::new().expect("temp dir"); let output = temp.path().join("missing-node"); let provider = BuildBinaryProvider { @@ -140,6 +151,7 @@ fn fails_when_build_command_does_not_create_output() { let error = provider .resolve() + .await .expect_err("missing build output is rejected"); assert!(matches!( @@ -148,8 +160,8 @@ fn fails_when_build_command_does_not_create_output() { )); } -#[test] -fn downloads_binary_from_minimal_http_server() { +#[tokio::test] +async fn downloads_binary_from_minimal_http_server() { let temp = TempDir::new().expect("temp dir"); let body = b"downloaded-node"; let server = SingleResponseServer::start(body); @@ -160,13 +172,16 @@ fn downloads_binary_from_minimal_http_server() { cache_dir: Some(temp.path().join("cache")), processor: None, }; - let path = provider.resolve().expect("download provider resolves"); + let path = provider + .resolve() + .await + .expect("download provider resolves"); assert_eq!(fs::read(path).expect("downloaded file"), body); } -#[test] -fn rejects_download_checksum_mismatch() { +#[tokio::test] +async fn rejects_download_checksum_mismatch() { let temp = TempDir::new().expect("temp dir"); let server = SingleResponseServer::start(b"downloaded-node"); let provider = DownloadBinaryProvider { @@ -178,6 +193,7 @@ fn rejects_download_checksum_mismatch() { let error = provider .resolve() + .await .expect_err("checksum mismatch is rejected"); assert!(matches!( @@ -186,8 +202,8 @@ fn rejects_download_checksum_mismatch() { )); } -#[test] -fn processes_downloaded_artifact_before_publishing_binary() { +#[tokio::test] +async fn processes_downloaded_artifact_before_publishing_binary() { let temp = TempDir::new().expect("temp dir"); let body = b"archive:downloaded-node"; let server = SingleResponseServer::start(body); @@ -209,7 +225,10 @@ fn processes_downloaded_artifact_before_publishing_binary() { Ok(()) }); - let path = provider.resolve().expect("processed download resolves"); + let path = provider + .resolve() + .await + .expect("processed download resolves"); assert_eq!( fs::read(path).expect("processed binary"), @@ -218,8 +237,8 @@ fn processes_downloaded_artifact_before_publishing_binary() { assert_eq!(process_count.load(Ordering::SeqCst), 1); } -#[test] -fn rejects_processor_that_does_not_create_output() { +#[tokio::test] +async fn rejects_processor_that_does_not_create_output() { let temp = TempDir::new().expect("temp dir"); let body = b"archive"; let server = SingleResponseServer::start(body); @@ -233,6 +252,7 @@ fn rejects_processor_that_does_not_create_output() { let error = provider .resolve() + .await .expect_err("missing processed output is rejected"); assert!(matches!( @@ -263,8 +283,9 @@ impl CountingBinaryProvider { } } +#[async_trait::async_trait] impl BinaryProvider for CountingBinaryProvider { - fn try_resolve(&self) -> Result, BinaryProviderError> { + async fn try_resolve(&self) -> Result, BinaryProviderError> { self.resolve_count.fetch_add(1, Ordering::SeqCst); Ok(Some(self.path.clone())) diff --git a/testing-framework/deployers/local/src/env/helpers.rs b/testing-framework/deployers/local/src/env/helpers.rs index d639b15..edf130b 100644 --- a/testing-framework/deployers/local/src/env/helpers.rs +++ b/testing-framework/deployers/local/src/env/helpers.rs @@ -375,15 +375,15 @@ where } /// Serializes a config as YAML and builds a launch spec for `spec`. -pub fn yaml_config_launch_spec( +pub async fn yaml_config_launch_spec( config: &T, spec: &LocalProcessSpec, ) -> Result { let config_yaml = serde_yaml::to_string(config)?; - rendered_config_launch_spec(config_yaml.into_bytes(), spec) + rendered_config_launch_spec(config_yaml.into_bytes(), spec).await } -pub fn build_launch_spec_with_args( +pub async fn build_launch_spec_with_args( config: &::NodeConfig, dir: &std::path::Path, label: &str, @@ -392,21 +392,21 @@ pub fn build_launch_spec_with_args( where E: crate::env::LocalDeployerEnv, { - let mut launch = E::build_launch_spec(config, dir, label)?; + let mut launch = E::build_launch_spec(config, dir, label).await?; launch.args.extend(extra_args.iter().cloned()); Ok(launch) } /// Uses an already rendered text config to build a launch spec for `spec`. -pub fn text_config_launch_spec( +pub async fn text_config_launch_spec( rendered_config: impl Into>, spec: &LocalProcessSpec, ) -> Result { - rendered_config_launch_spec(rendered_config.into(), spec) + rendered_config_launch_spec(rendered_config.into(), spec).await } /// Uses the standard binary+config launch shape for a YAML-rendered config. -pub fn default_yaml_launch_spec( +pub async fn default_yaml_launch_spec( config: &T, binary_env_var: &str, rust_log: &str, @@ -415,6 +415,7 @@ pub fn default_yaml_launch_spec( config, &LocalProcessSpec::new(binary_env_var).with_rust_log(rust_log), ) + .await } /// Serializes a node config as YAML bytes. @@ -427,11 +428,11 @@ pub fn text_node_config(rendered_config: impl Into>) -> Vec { rendered_config.into() } -pub(crate) fn rendered_config_launch_spec( +pub(crate) async fn rendered_config_launch_spec( rendered_config: Vec, spec: &LocalProcessSpec, ) -> Result { - let binary = spec.binary.resolve()?; + let binary = spec.binary.resolve().await?; let mut args = config_file_args(spec); args.extend(spec.extra_args.iter().cloned()); @@ -456,19 +457,21 @@ fn config_file_args(spec: &LocalProcessSpec) -> Vec { mod tests { use super::{LocalProcessSpec, text_config_launch_spec}; - #[test] - fn launch_spec_uses_flag_config_by_default() { + #[tokio::test] + async fn launch_spec_uses_flag_config_by_default() { let temp = tempfile::tempdir().expect("temp dir"); let binary = temp.path().join("app"); std::fs::write(&binary, b"binary").expect("test binary"); let spec = LocalProcessSpec::new("APP_BIN").with_binary_path(binary); - let launch = text_config_launch_spec("config", &spec).expect("launch spec"); + let launch = text_config_launch_spec("config", &spec) + .await + .expect("launch spec"); assert_eq!(launch.args, ["--config", "config.yaml"]); } - #[test] - fn launch_spec_can_use_positional_config_path() { + #[tokio::test] + async fn launch_spec_can_use_positional_config_path() { let temp = tempfile::tempdir().expect("temp dir"); let binary = temp.path().join("app"); std::fs::write(&binary, b"binary").expect("test binary"); @@ -476,7 +479,9 @@ mod tests { .with_binary_path(binary) .with_positional_config_file("app.json") .with_args(["--port".to_owned(), "8080".to_owned()]); - let launch = text_config_launch_spec("config", &spec).expect("launch spec"); + let launch = text_config_launch_spec("config", &spec) + .await + .expect("launch spec"); assert_eq!(launch.args, ["app.json", "--port", "8080"]); } diff --git a/testing-framework/deployers/local/src/env/mod.rs b/testing-framework/deployers/local/src/env/mod.rs index 34993cd..6cbdcb0 100644 --- a/testing-framework/deployers/local/src/env/mod.rs +++ b/testing-framework/deployers/local/src/env/mod.rs @@ -274,7 +274,7 @@ where } /// Builds the full launch spec for a local node process. - fn build_launch_spec( + async fn build_launch_spec( config: &::NodeConfig, _dir: &Path, label: &str, @@ -283,7 +283,7 @@ where std::io::Error::other("build_launch_spec is not implemented for this app") })?; let rendered = Self::render_local_config(config)?; - helpers::rendered_config_launch_spec(rendered, &spec) + helpers::rendered_config_launch_spec(rendered, &spec).await } /// Returns the main HTTP API port from a node config when the app follows @@ -660,7 +660,12 @@ pub async fn spawn_node_from_config( ProcessNode::spawn( &label, config, - move |config, dir, label| build_launch_spec_with_args::(config, dir, label, &extra_args), + move |config, dir, label| { + let extra_args = extra_args.clone(); + Box::pin(async move { + build_launch_spec_with_args::(config, dir, label, &extra_args).await + }) + }, E::node_endpoints, keep_tempdir, persist_dir, diff --git a/testing-framework/deployers/local/src/env/tests.rs b/testing-framework/deployers/local/src/env/tests.rs index 044adc1..7439642 100644 --- a/testing-framework/deployers/local/src/env/tests.rs +++ b/testing-framework/deployers/local/src/env/tests.rs @@ -60,7 +60,7 @@ impl LocalDeployerEnv for DummyEnv { build_dummy_initial_nodes() } - fn build_launch_spec( + async fn build_launch_spec( config: &DummyConfig, dir: &std::path::Path, label: &str, diff --git a/testing-framework/deployers/local/src/node_control/mod.rs b/testing-framework/deployers/local/src/node_control/mod.rs index 20d1ab0..0a6f1af 100644 --- a/testing-framework/deployers/local/src/node_control/mod.rs +++ b/testing-framework/deployers/local/src/node_control/mod.rs @@ -292,6 +292,7 @@ impl NodeManager { name, &options.args, ) + .await .map_err(|source| NodeManagerError::Config { source })?; if let Err(source) = node.restart_with_launch(launch).await { diff --git a/testing-framework/deployers/local/src/process.rs b/testing-framework/deployers/local/src/process.rs index 485de73..142484e 100644 --- a/testing-framework/deployers/local/src/process.rs +++ b/testing-framework/deployers/local/src/process.rs @@ -1,10 +1,12 @@ use std::{ collections::HashMap, env, fs, + future::Future, io::{self, Error, ErrorKind}, mem, net::{Ipv4Addr, SocketAddr}, path::{Path, PathBuf}, + pin::Pin, process::Stdio, thread, time::{Duration, Instant}, @@ -231,7 +233,13 @@ impl Result, + build_launch_spec: impl for<'a> FnOnce( + &'a Config, + &'a Path, + &'a str, + ) -> Pin< + Box> + Send + 'a>, + >, endpoints_from_config: impl FnOnce(&Config) -> Result, keep_tempdir: bool, persist_dir: Option<&Path>, @@ -245,6 +253,7 @@ impl Date: Tue, 4 Aug 2026 09:04:36 +0200 Subject: [PATCH 2/2] 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"); + } }