Multiple test utils relays

Cargo-minimal.lock

@@ -2914,6 +2914,7 @@ dependencies = [
  "bitcoin 0.32.100",
  "bitcoin-ohttp",
  "corepc-node",
+ "futures",
  "http",
  "once_cell",
  "payjoin",

Cargo-recent.lock

@@ -2890,6 +2890,7 @@ dependencies = [
  "bitcoin 0.32.100",
  "bitcoin-ohttp",
  "corepc-node",
+ "futures",
  "http",
  "once_cell",
  "payjoin",

payjoin-cli/tests/e2e.rs

@@ -221,7 +221,7 @@ mod e2e {
         type Result<T> = std::result::Result<T, BoxError>;
 
         init_tracing();
-        let mut services = TestServices::initialize().await?;
+        let mut services = TestServices::initialize_with_relays(3).await?;
         let temp_dir = tempdir()?;
 
         let result = tokio::select! {
@@ -250,7 +250,7 @@ mod e2e {
             let payjoin_cli = env!("CARGO_BIN_EXE_payjoin-cli");
 
             let directory = &services.directory_url();
-            let ohttp_relay = &services.ohttp_relay_url();
+            let ohttp_relays = &services.ohttp_relay_urls();
 
             let cli_receive_initiator = Command::new(payjoin_cli)
                 .arg("--root-certificate")
@@ -262,7 +262,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&receiver_db_path)
                 .arg("--ohttp-relays")
-                .arg(ohttp_relay)
+                .arg(ohttp_relays)
                 .arg("receive")
                 .arg(RECEIVE_SATS)
                 .arg("--pj-directory")
@@ -284,7 +284,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&sender_db_path)
                 .arg("--ohttp-relays")
-                .arg(ohttp_relay)
+                .arg(ohttp_relays)
                 .arg("send")
                 .arg(&bip21)
                 .arg("--fee-rate")
@@ -305,7 +305,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&receiver_db_path)
                 .arg("--ohttp-relays")
-                .arg(ohttp_relay)
+                .arg(ohttp_relays)
                 .arg("resume")
                 .stdout(Stdio::piped())
                 .stderr(Stdio::inherit())
@@ -323,7 +323,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&receiver_db_path)
                 .arg("--ohttp-relays")
-                .arg(ohttp_relay)
+                .arg(ohttp_relays)
                 .arg("resume")
                 .stdout(Stdio::piped())
                 .stderr(Stdio::inherit())
@@ -341,7 +341,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&sender_db_path)
                 .arg("--ohttp-relays")
-                .arg(ohttp_relay)
+                .arg(ohttp_relays)
                 .arg("send")
                 .arg(&bip21)
                 .arg("--fee-rate")
@@ -372,7 +372,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&receiver_db_path)
                 .arg("--ohttp-relays")
-                .arg(ohttp_relay)
+                .arg(ohttp_relays)
                 .arg("resume")
                 .stdout(Stdio::piped())
                 .stderr(Stdio::inherit())
@@ -391,7 +391,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&receiver_db_path)
                 .arg("--ohttp-relays")
-                .arg(ohttp_relay)
+                .arg(ohttp_relays)
                 .arg("resume")
                 .stdout(Stdio::piped())
                 .stderr(Stdio::inherit())
@@ -408,7 +408,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&sender_db_path)
                 .arg("--ohttp-relays")
-                .arg(ohttp_relay)
+                .arg(ohttp_relays)
                 .arg("resume")
                 .stdout(Stdio::piped())
                 .stderr(Stdio::inherit())
@@ -508,7 +508,7 @@ mod e2e {
         type Result<T> = std::result::Result<T, BoxError>;
 
         init_tracing();
-        let services = TestServices::initialize().await?;
+        let services = TestServices::initialize_with_relays(3).await?;
         let temp_dir = tempdir()?;
 
         let result = send_v2_receive_v1_async(&services, &temp_dir).await;
@@ -605,7 +605,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&sender_db_path)
                 .arg("--ohttp-relays")
-                .arg(services.ohttp_relay_url())
+                .arg(services.ohttp_relay_urls())
                 .arg("send")
                 .arg(&bip21)
                 .arg("--fee-rate")
@@ -665,7 +665,7 @@ mod e2e {
         type Result<T> = std::result::Result<T, BoxError>;
 
         init_tracing();
-        let mut services = TestServices::initialize().await?;
+        let mut services = TestServices::initialize_with_relays(3).await?;
         let temp_dir = tempdir()?;
 
         let result = tokio::select! {
@@ -692,7 +692,7 @@ mod e2e {
             let cookie_file = &bitcoind.params.cookie_file;
             let payjoin_cli = env!("CARGO_BIN_EXE_payjoin-cli");
             let directory = &services.directory_url();
-            let ohttp_relay = &services.ohttp_relay_url();
+            let ohttp_relays = &services.ohttp_relay_urls();
 
             // Get a BIP21 from a receiver then kill it so the sender can never complete payjoin
             let cli_receiver = Command::new(payjoin_cli)
@@ -705,7 +705,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&receiver_db_path)
                 .arg("--ohttp-relays")
-                .arg(ohttp_relay)
+                .arg(ohttp_relays)
                 .arg("receive")
                 .arg(RECEIVE_SATS)
                 .arg("--pj-directory")
@@ -729,7 +729,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&sender_db_path)
                 .arg("--ohttp-relays")
-                .arg(ohttp_relay)
+                .arg(ohttp_relays)
                 .arg("send")
                 .arg(&bip21)
                 .arg("--fee-rate")
@@ -755,7 +755,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&sender_db_path)
                 .arg("--ohttp-relays")
-                .arg(ohttp_relay)
+                .arg(ohttp_relays)
                 .arg("cancel")
                 .arg(session_id.to_string())
                 .stdout(Stdio::piped())
@@ -798,7 +798,7 @@ mod e2e {
         type Result<T> = std::result::Result<T, BoxError>;
 
         init_tracing();
-        let mut services = TestServices::initialize().await?;
+        let mut services = TestServices::initialize_with_relays(3).await?;
         let temp_dir = tempdir()?;
 
         let result = tokio::select! {
@@ -823,7 +823,7 @@ mod e2e {
             let cookie_file = &bitcoind.params.cookie_file;
             let payjoin_cli = env!("CARGO_BIN_EXE_payjoin-cli");
             let directory = &services.directory_url();
-            let ohttp_relay = &services.ohttp_relay_url();
+            let ohttp_relays = &services.ohttp_relay_urls();
 
             // Start a receiver and capture its BIP21 so a session is persisted,
             // then leave it parked at Initialized waiting for a proposal.
@@ -837,7 +837,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&receiver_db_path)
                 .arg("--ohttp-relays")
-                .arg(ohttp_relay)
+                .arg(ohttp_relays)
                 .arg("receive")
                 .arg(RECEIVE_SATS)
                 .arg("--pj-directory")
@@ -865,7 +865,7 @@ mod e2e {
                 .arg("--db-path")
                 .arg(&receiver_db_path)
                 .arg("--ohttp-relays")
-                .arg(ohttp_relay)
+                .arg(ohttp_relays)
                 .arg("cancel")
                 .arg(session_id.to_string())
                 .arg("--role")

payjoin-test-utils/Cargo.toml

@@ -28,6 +28,7 @@ axum-server = { version = "0.8", features = [
 ], optional = true }
 bitcoin = { version = "0.32.7", features = ["base64"] }
 corepc-node = { version = "0.10.0", features = ["download", "29_0"] }
+futures = { version = "0.3", default-features = false, features = ["std"] }
 http = { version = "1.3.1", optional = true }
 ohttp = { package = "bitcoin-ohttp", version = "0.6.0", optional = true }
 once_cell = "1.21.3"

payjoin-test-utils/src/v2.rs

@@ -2,6 +2,8 @@ use std::result::Result;
 use std::sync::Arc;
 use std::time::Duration;
 
+type RelayEntry = (u16, Option<JoinHandle<Result<(), BoxSendSyncError>>>);
+
 use axum_server::tls_rustls::RustlsConfig;
 use http::StatusCode;
 use ohttp::hpke::{Aead, Kdf, Kem};
@@ -21,12 +23,16 @@ use crate::BoxSendSyncError;
 pub struct TestServices {
     cert: Certificate,
     directory: (u16, Option<JoinHandle<Result<(), BoxSendSyncError>>>),
-    ohttp_relay: (u16, Option<JoinHandle<Result<(), BoxSendSyncError>>>),
+    ohttp_relays: Vec<RelayEntry>,
     http_agent: Arc<Client>,
 }
 
 impl TestServices {
     pub async fn initialize() -> Result<Self, BoxSendSyncError> {
+        Self::initialize_with_relays(2).await
+    }
+
+    pub async fn initialize_with_relays(num_relays: u8) -> Result<Self, BoxSendSyncError> {
         // TODO add a UUID, and cleanup guard to delete after on successful run
         let cert = local_cert_key();
         let cert_der = cert.cert.der().to_vec();
@@ -37,14 +43,19 @@ impl TestServices {
         root_store.add(CertificateDer::from(cert.cert.der().to_vec())).unwrap();
 
         let directory = init_directory(cert_key, root_store.clone()).await?;
-        let ohttp_relay = init_ohttp_relay(root_store, None).await?;
+
+        let mut ohttp_relays = Vec::with_capacity(num_relays as usize);
+        for _ in 0..num_relays {
+            let relay = init_ohttp_relay(root_store.clone(), None).await?;
+            ohttp_relays.push((relay.0, Some(relay.1)));
+        }
 
         let http_agent: Arc<Client> = Arc::new(http_agent(cert_der)?);
 
         Ok(Self {
             cert: cert.cert,
             directory: (directory.0, Some(directory.1)),
-            ohttp_relay: (ohttp_relay.0, Some(ohttp_relay.1)),
+            ohttp_relays,
             http_agent,
         })
     }
@@ -57,20 +68,43 @@ impl TestServices {
         self.directory.1.take().expect("directory handle not found")
     }
 
-    pub fn ohttp_relay_url(&self) -> String { format!("http://localhost:{}", self.ohttp_relay.0) }
+    pub fn ohttp_relay_url(&self) -> String {
+        format!("http://localhost:{}", self.ohttp_relays[0].0)
+    }
+
+    pub fn ohttp_relay_urls(&self) -> String {
+        self.ohttp_relays
+            .iter()
+            .map(|r| format!("http://localhost:{}", r.0))
+            .collect::<Vec<_>>()
+            .join(",")
+    }
 
     pub fn ohttp_gateway_url(&self) -> String {
         format!("{}/.well-known/ohttp-gateway", self.directory_url())
     }
 
     pub fn take_ohttp_relay_handle(&mut self) -> JoinHandle<Result<(), BoxSendSyncError>> {
-        self.ohttp_relay.1.take().expect("ohttp relay handle not found")
+        let handles: Vec<_> = self
+            .ohttp_relays
+            .iter_mut()
+            .map(|r| r.1.take().expect("ohttp relay handle not found"))
+            .collect();
+        tokio::spawn(async move {
+            match futures::future::select_all(handles).await {
+                (Ok(inner), _idx, _rest) => inner,
+                (Err(e), _idx, _rest) => Err(e.into()),
+            }
+        })
     }
 
     pub fn http_agent(&self) -> Arc<Client> { self.http_agent.clone() }
 
     pub async fn wait_for_services_ready(&self) -> Result<(), &'static str> {
-        wait_for_service_ready(&self.ohttp_relay_url(), self.http_agent()).await?;
+        for relay in &self.ohttp_relays {
+            wait_for_service_ready(&format!("http://localhost:{}", relay.0), self.http_agent())
+                .await?;
+        }
         wait_for_service_ready(&self.directory_url(), self.http_agent()).await?;
         Ok(())
     }