Merge pull request #60 from logos-blockchain/fix/async-binary-provider-resolution

fix(tf): resolve local binaries asynchronously

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",

examples/multi_app/fixture/src/lib.rs

@@ -123,7 +123,7 @@ impl AppDeployment<AppHostEnv> 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(),

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(),

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"

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<Self, BinaryProviderError> {
+    pub(super) async fn acquire(path: &Path) -> Result<Self, BinaryProviderError> {
         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 {

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<Option<PathBuf>, BinaryProviderError>;
+    async fn try_resolve(&self) -> Result<Option<PathBuf>, 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<PathBuf, BinaryProviderError> {
+    async fn resolve(&self) -> Result<PathBuf, BinaryProviderError> {
         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<PathBuf, BinaryProviderError> {
-        if let Some(path) = self.try_resolve()? {
+    async fn resolve_uncached(&self) -> Result<PathBuf, BinaryProviderError> {
+        if let Some(path) = self.try_resolve().await? {
             return Ok(path);
         }
 

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<Option<PathBuf>, BinaryProviderError> {
+    async fn try_resolve(&self) -> Result<Option<PathBuf>, 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,
@@ -83,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
     }
 

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<Option<PathBuf>, BinaryProviderError> {
+    async fn try_resolve(&self) -> Result<Option<PathBuf>, 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<Vec<u8>, BinaryProviderError> {
+    async fn download_bytes(&self, url: &str) -> Result<Vec<u8>, 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(),

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<Option<PathBuf>, BinaryProviderError> {
+    async fn try_resolve(&self) -> Result<Option<PathBuf>, BinaryProviderError> {
         let Some(path) = env::var_os(&self.env_var).map(PathBuf::from) else {
             return Ok(None);
         };

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<Option<PathBuf>, BinaryProviderError> {
+    async fn try_resolve(&self) -> Result<Option<PathBuf>, 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),

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<Option<PathBuf>, BinaryProviderError> {
+    async fn try_resolve(&self) -> Result<Option<PathBuf>, BinaryProviderError> {
         if !self.path.is_absolute() {
             return Err(BinaryProviderError::RelativePath {
                 path: self.path.clone(),

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};
@@ -19,30 +22,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 +57,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 +85,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 +112,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 +135,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 +154,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 +163,36 @@ fn fails_when_build_command_does_not_create_output() {
     ));
 }
 
-#[test]
-fn downloads_binary_from_minimal_http_server() {
+#[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");
     let body = b"downloaded-node";
     let server = SingleResponseServer::start(body);
@@ -160,13 +203,63 @@ 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 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");
     let server = SingleResponseServer::start(b"downloaded-node");
     let provider = DownloadBinaryProvider {
@@ -178,6 +271,7 @@ fn rejects_download_checksum_mismatch() {
 
     let error = provider
         .resolve()
+        .await
         .expect_err("checksum mismatch is rejected");
 
     assert!(matches!(
@@ -186,8 +280,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 +303,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 +315,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 +330,7 @@ fn rejects_processor_that_does_not_create_output() {
 
     let error = provider
         .resolve()
+        .await
         .expect_err("missing processed output is rejected");
 
     assert!(matches!(
@@ -245,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))
 }
@@ -263,8 +398,9 @@ impl CountingBinaryProvider {
     }
 }
 
+#[async_trait::async_trait]
 impl BinaryProvider for CountingBinaryProvider {
-    fn try_resolve(&self) -> Result<Option<PathBuf>, BinaryProviderError> {
+    async fn try_resolve(&self) -> Result<Option<PathBuf>, BinaryProviderError> {
         self.resolve_count.fetch_add(1, Ordering::SeqCst);
 
         Ok(Some(self.path.clone()))
@@ -288,17 +424,29 @@ fn sha256_hex(bytes: &[u8]) -> String {
 
 struct SingleResponseServer {
     addr: String,
+    request_started: Option<tokio::sync::oneshot::Receiver<()>>,
+    release_response: Option<std::sync::mpsc::Sender<()>>,
 }
 
 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()
@@ -310,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");
+    }
 }

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<T: Serialize>(
+pub async fn yaml_config_launch_spec<T: Serialize>(
     config: &T,
     spec: &LocalProcessSpec,
 ) -> Result<LaunchSpec, DynError> {
     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<E>(
+pub async fn build_launch_spec_with_args<E>(
     config: &<E as Application>::NodeConfig,
     dir: &std::path::Path,
     label: &str,
@@ -392,21 +392,21 @@ pub fn build_launch_spec_with_args<E>(
 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<Vec<u8>>,
     spec: &LocalProcessSpec,
 ) -> Result<LaunchSpec, DynError> {
-    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<T: Serialize>(
+pub async fn default_yaml_launch_spec<T: Serialize>(
     config: &T,
     binary_env_var: &str,
     rust_log: &str,
@@ -415,6 +415,7 @@ pub fn default_yaml_launch_spec<T: Serialize>(
         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<u8>>) -> Vec<u8> {
     rendered_config.into()
 }
 
-pub(crate) fn rendered_config_launch_spec(
+pub(crate) async fn rendered_config_launch_spec(
     rendered_config: Vec<u8>,
     spec: &LocalProcessSpec,
 ) -> Result<LaunchSpec, DynError> {
-    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<String> {
 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"]);
     }

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: &<Self as Application>::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<E: LocalDeployerEnv>(
     ProcessNode::spawn(
         &label,
         config,
-        move |config, dir, label| build_launch_spec_with_args::<E>(config, dir, label, &extra_args),
+        move |config, dir, label| {
+            let extra_args = extra_args.clone();
+            Box::pin(async move {
+                build_launch_spec_with_args::<E>(config, dir, label, &extra_args).await
+            })
+        },
         E::node_endpoints,
         keep_tempdir,
         persist_dir,

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,

testing-framework/deployers/local/src/node_control/mod.rs

@@ -292,6 +292,7 @@ impl<E: LocalDeployerEnv> NodeManager<E> {
             name,
             &options.args,
         )
+        .await
         .map_err(|source| NodeManagerError::Config { source })?;
 
         if let Err(source) = node.restart_with_launch(launch).await {

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<Config: Clone + Send + Sync + 'static, Client: Clone + Send + Sync + 'stati
     pub async fn spawn(
         label: &str,
         config: Config,
-        build_launch_spec: impl FnOnce(&Config, &Path, &str) -> Result<LaunchSpec, DynError>,
+        build_launch_spec: impl for<'a> FnOnce(
+            &'a Config,
+            &'a Path,
+            &'a str,
+        ) -> Pin<
+            Box<dyn Future<Output = Result<LaunchSpec, DynError>> + Send + 'a>,
+        >,
         endpoints_from_config: impl FnOnce(&Config) -> Result<NodeEndpoints, DynError>,
         keep_tempdir: bool,
         persist_dir: Option<&Path>,
@@ -245,6 +253,7 @@ impl<Config: Clone + Send + Sync + 'static, Client: Clone + Send + Sync + 'stati
         }
 
         let launch = build_launch_spec(&config, tempdir.path(), label)
+            .await
             .map_err(|source| ProcessSpawnError::Config { source })?;
         let endpoints = endpoints_from_config(&config)
             .map_err(|source| ProcessSpawnError::Config { source })?;
@@ -481,10 +490,12 @@ mod tests {
             "node",
             (),
             |_, _, _| {
-                Ok(LaunchSpec {
-                    binary: "/bin/sleep".into(),
-                    args: vec!["60".into()],
-                    ..LaunchSpec::default()
+                Box::pin(async {
+                    Ok(LaunchSpec {
+                        binary: "/bin/sleep".into(),
+                        args: vec!["60".into()],
+                        ..LaunchSpec::default()
+                    })
                 })
             },
             |_| Ok(NodeEndpoints::default()),

testing-framework/deployers/local/src/provisioner.rs

@@ -327,7 +327,7 @@ mod tests {
             Ok(Vec::new())
         }
 
-        fn build_launch_spec(
+        async fn build_launch_spec(
             _config: &EmptyConfig,
             _dir: &Path,
             _label: &str,