refactor(coprocessor): enable multi threaded test execution in CI

.github/workflows/coprocessor-cargo-tests.yml

@@ -116,7 +116,7 @@ jobs:
       run: |
         DATABASE_URL=postgresql://postgres:postgres@localhost:5432/coprocessor \
         TEST_GLOBAL_LOCALSTACK=1 \
-        cargo llvm-cov --no-report --workspace --profile coverage -- --test-threads=1
+        cargo llvm-cov --no-report --workspace --profile coverage
       working-directory: coprocessor/fhevm-engine
 
     - name: Generate coverage report

coprocessor/fhevm-engine/test-harness/src/instance.rs

@@ -78,10 +78,13 @@ async fn setup_test_app_existing_localhost(
     })
 }
 
+const POSTGRES_PORT: u16 = 5432;
+
 async fn setup_test_app_custom_docker(
     mode: ImportMode,
 ) -> Result<DBInstance, Box<dyn std::error::Error>> {
     let container = GenericImage::new("postgres", "15.7")
+        .with_exposed_port(POSTGRES_PORT.into())
         .with_wait_for(WaitFor::message_on_stderr(
             "database system is ready to accept connections",
         ))
@@ -94,7 +97,7 @@ async fn setup_test_app_custom_docker(
     info!("Postgres container started");
 
     let cont_host = container.get_host().await?;
-    let cont_port = container.get_host_port_ipv4(5432).await?;
+    let cont_port = container.get_host_port_ipv4(POSTGRES_PORT).await?;
 
     let admin_db_url = format!("postgresql://postgres:postgres@{cont_host}:{cont_port}/postgres");
     let db_url = format!("postgresql://postgres:postgres@{cont_host}:{cont_port}/coprocessor");

coprocessor/fhevm-engine/test-harness/src/localstack.rs

@@ -7,7 +7,7 @@ use base64::Engine;
 use k256::SecretKey;
 use testcontainers::{core::WaitFor, runners::AsyncRunner, ContainerAsync, GenericImage, ImageExt};
 
-fn pick_free_port() -> u16 {
+pub fn pick_free_port() -> u16 {
     TcpListener::bind("127.0.0.1:0")
         .unwrap()
         .local_addr()

coprocessor/fhevm-engine/tfhe-worker/src/tests/health_check.rs

@@ -6,22 +6,22 @@ use tokio::process::Command;
 async fn test_health_check() -> Result<(), Box<dyn std::error::Error>> {
     let app = setup_test_app().await?;
     eprintln!("App started");
-    let url = "http://127.0.0.1:8081";
-    assert!(health_check::wait_alive(url, 10, 1).await);
-    assert!(health_check::wait_healthy(url, 10, 1).await);
+    let url = app.health_check_url();
+    assert!(health_check::wait_alive(&url, 10, 1).await);
+    assert!(health_check::wait_healthy(&url, 10, 1).await);
     tokio::time::sleep(tokio::time::Duration::from_secs(20)).await;
-    assert!(health_check::wait_alive(url, 10, 1).await);
-    assert!(health_check::wait_healthy(url, 10, 1).await);
+    assert!(health_check::wait_alive(&url, 10, 1).await);
+    assert!(health_check::wait_healthy(&url, 10, 1).await);
     eprintln!("Pausing database");
     let db_id = app
         .db_docker_id()
         .expect("Database Docker ID should be set");
     Command::new("docker").args(["pause", &db_id]).spawn()?;
     tokio::time::sleep(tokio::time::Duration::from_secs(15)).await;
-    assert!(!health_check::wait_healthy(url, 10, 1).await);
+    assert!(!health_check::wait_healthy(&url, 10, 1).await);
     tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
     eprintln!("Unpausing database");
     Command::new("docker").args(["unpause", &db_id]).spawn()?;
-    assert!(health_check::wait_healthy(url, 10, 1).await);
+    assert!(health_check::wait_healthy(&url, 10, 1).await);
     Ok(())
 }

coprocessor/fhevm-engine/tfhe-worker/src/tests/utils.rs

@@ -16,6 +16,7 @@ pub struct TestInstance {
     // send message to this on destruction to stop the app
     app_close_channel: Option<tokio::sync::watch::Sender<bool>>,
     db_url: String,
+    health_check_port: u16,
 }
 
 impl Drop for TestInstance {
@@ -35,6 +36,10 @@ impl TestInstance {
     pub fn db_docker_id(&self) -> Option<String> {
         self._container.as_ref().map(|c| c.id().to_string())
     }
+
+    pub fn health_check_url(&self) -> String {
+        format!("http://127.0.0.1:{}", self.health_check_port)
+    }
 }
 
 pub fn default_dependence_cache_size() -> u16 {
@@ -53,15 +58,17 @@ const LOCAL_DB_URL: &str = "postgresql://postgres:postgres@127.0.0.1:5432/coproc
 
 async fn setup_test_app_existing_db() -> Result<TestInstance, Box<dyn std::error::Error>> {
     let (app_close_channel, rx) = tokio::sync::watch::channel(false);
-    start_coprocessor(rx, LOCAL_DB_URL).await;
+    let health_check_port = start_coprocessor(rx, LOCAL_DB_URL).await;
     Ok(TestInstance {
         _container: None,
         app_close_channel: Some(app_close_channel),
         db_url: LOCAL_DB_URL.to_string(),
+        health_check_port,
     })
 }
 
-async fn start_coprocessor(rx: Receiver<bool>, db_url: &str) {
+async fn start_coprocessor(rx: Receiver<bool>, db_url: &str) -> u16 {
+    let health_check_port = test_harness::localstack::pick_free_port();
     let args: Args = Args {
         run_bg_worker: true,
         worker_polling_interval_ms: 1000,
@@ -76,7 +83,7 @@ async fn start_coprocessor(rx: Receiver<bool>, db_url: &str) {
         database_url: Some(db_url.into()),
         service_name: "coprocessor".to_string(),
         log_level: Level::INFO,
-        health_check_port: 8081,
+        health_check_port,
         metric_rerand_batch_latency: MetricsConfig::default(),
         metric_fhe_batch_latency: MetricsConfig::default(),
         worker_id: None,
@@ -95,6 +102,7 @@ async fn start_coprocessor(rx: Receiver<bool>, db_url: &str) {
 
     // wait until worker starts
     tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
+    health_check_port
 }
 
 async fn setup_test_app_custom_docker() -> Result<TestInstance, Box<dyn std::error::Error>> {
@@ -133,11 +141,12 @@ async fn setup_test_app_custom_docker() -> Result<TestInstance, Box<dyn std::err
     println!("DB prepared");
 
     let (app_close_channel, rx) = tokio::sync::watch::channel(false);
-    start_coprocessor(rx, &db_url).await;
+    let health_check_port = start_coprocessor(rx, &db_url).await;
     Ok(TestInstance {
         _container: Some(container),
         app_close_channel: Some(app_close_channel),
         db_url,
+        health_check_port,
     })
 }