apollo_infra: concurrent tests

victorkstarkware · victorkstarkware · commit 4205ac59f480 · 2026-01-07T17:14:52.000+02:00
diff --git a/.config/nextest.toml b/.config/nextest.toml
@@ -7,7 +7,7 @@ max-threads = 1
 # serialized tests (mostly integration tests)
 [[profile.default.overrides]]
 # TODO(victork): evaluate whether these are indeed required to run sequentially
-filter = '(package(apollo_integration_tests) & kind(test)) | package(apollo_infra) | package(apollo_network) | (package(apollo_l1_provider) & kind(test))'
+filter = '(package(apollo_integration_tests) & kind(test)) | package(apollo_network) | (package(apollo_l1_provider) & kind(test))'
 test-group = 'serialized'
 
 # slow test exception whitelist
diff --git a/crates/apollo_infra/src/tests/concurrent_servers_test.rs b/crates/apollo_infra/src/tests/concurrent_servers_test.rs
@@ -31,8 +31,8 @@ use crate::component_server::{
     RemoteComponentServer,
 };
 use crate::tests::{
+    available_ports,
     dummy_remote_server_config,
-    AVAILABLE_PORTS,
     TEST_LOCAL_CLIENT_METRICS,
     TEST_LOCAL_SERVER_METRICS,
     TEST_REMOTE_CLIENT_METRICS,
@@ -157,7 +157,7 @@ async fn setup_concurrent_local_test() -> LocalConcurrentComponentClient {
 
 async fn setup_concurrent_remote_test() -> RemoteConcurrentComponentClient {
     let local_client = setup_concurrent_local_test().await;
-    let socket = AVAILABLE_PORTS.lock().await.get_next_local_host_socket();
+    let socket = available_ports(7).get_next_local_host_socket();
     let remote_client_config = RemoteClientConfig::default();
 
     let max_concurrency = 10;
diff --git a/crates/apollo_infra/src/tests/mod.rs b/crates/apollo_infra/src/tests/mod.rs
@@ -17,12 +17,11 @@ use apollo_metrics::metrics::{
     MetricScope,
 };
 use async_trait::async_trait;
-use once_cell::sync::Lazy;
 use serde::{Deserialize, Serialize};
 use starknet_types_core::felt::Felt;
 use strum::{EnumVariantNames, VariantNames};
 use strum_macros::{AsRefStr, EnumDiscriminants, EnumIter, IntoStaticStr};
-use tokio::sync::{Mutex, Semaphore};
+use tokio::sync::Semaphore;
 
 use crate::component_client::ClientResult;
 use crate::component_definitions::{ComponentRequestHandler, ComponentStarter, PrioritizedRequest};
@@ -185,11 +184,13 @@ const TEST_LOCAL_CLIENT_RESPONSE_TIMES: LabeledMetricHistogram = LabeledMetricHi
 pub(crate) const TEST_LOCAL_CLIENT_METRICS: LocalClientMetrics =
     LocalClientMetrics::new(&TEST_LOCAL_CLIENT_RESPONSE_TIMES);
 
-// Define the shared fixture
-pub static AVAILABLE_PORTS: Lazy<Arc<Mutex<AvailablePorts>>> = Lazy::new(|| {
-    let available_ports = AvailablePorts::new(TestIdentifier::InfraUnitTests.into(), 0);
-    Arc::new(Mutex::new(available_ports))
-});
+/// Creates an `AvailablePorts` instance with a unique `instance_index`.
+/// Each test that binds ports should use a different instance_index to get disjoint port ranges.
+/// This is necessary because nextest runs each test in its own process, so a shared static
+/// counter doesn't work - each process would start from the same base port.
+pub fn available_ports(instance_index: u16) -> AvailablePorts {
+    AvailablePorts::new(TestIdentifier::InfraUnitTests.into(), instance_index)
+}
 
 #[derive(Serialize, Deserialize, Clone, AsRefStr, EnumDiscriminants)]
 #[strum_discriminants(
diff --git a/crates/apollo_infra/src/tests/remote_component_client_server_test.rs b/crates/apollo_infra/src/tests/remote_component_client_server_test.rs
@@ -41,6 +41,7 @@ use crate::component_server::{
 };
 use crate::serde_utils::SerdeWrapper;
 use crate::tests::{
+    available_ports,
     dummy_remote_server_config,
     test_a_b_functionality,
     ComponentA,
@@ -55,7 +56,6 @@ use crate::tests::{
     ResultB,
     ValueA,
     ValueB,
-    AVAILABLE_PORTS,
     TEST_LOCAL_CLIENT_METRICS,
     TEST_LOCAL_SERVER_METRICS,
     TEST_REMOTE_CLIENT_METRICS,
@@ -135,7 +135,7 @@ async fn create_client_and_faulty_server<T>(body: T) -> ComponentAClient
 where
     T: Serialize + DeserializeOwned + Debug + Send + Sync + 'static + Clone,
 {
-    let socket = AVAILABLE_PORTS.lock().await.get_next_local_host_socket();
+    let socket = available_ports(0).get_next_local_host_socket();
     task::spawn(async move {
         async fn handler<T>(
             _http_request: Request<Body>,
@@ -183,8 +183,9 @@ async fn remote_connection_concurrency() {
     const MAX_ATTEMPTS: usize = 50;
 
     let setup_value: ValueB = Felt::from(90);
-    let a_socket = AVAILABLE_PORTS.lock().await.get_next_local_host_socket();
-    let b_socket = AVAILABLE_PORTS.lock().await.get_next_local_host_socket();
+    let mut ports = available_ports(1);
+    let a_socket = ports.get_next_local_host_socket();
+    let b_socket = ports.get_next_local_host_socket();
 
     // Shared semaphore used inside ComponentA::handle_request
     let semaphore = Arc::new(Semaphore::new(0));
@@ -401,8 +402,9 @@ async fn setup_for_tests(
 #[tokio::test]
 async fn proper_setup() {
     let setup_value: ValueB = Felt::from(90);
-    let a_socket = AVAILABLE_PORTS.lock().await.get_next_local_host_socket();
-    let b_socket = AVAILABLE_PORTS.lock().await.get_next_local_host_socket();
+    let mut ports = available_ports(2);
+    let a_socket = ports.get_next_local_host_socket();
+    let b_socket = ports.get_next_local_host_socket();
 
     setup_for_tests(setup_value, a_socket, b_socket, MAX_CONCURRENCY, None).await;
     let a_client_config = RemoteClientConfig::default();
@@ -426,8 +428,9 @@ async fn proper_setup() {
 
 #[tokio::test]
 async fn faulty_client_setup() {
-    let a_socket = AVAILABLE_PORTS.lock().await.get_next_local_host_socket();
-    let b_socket = AVAILABLE_PORTS.lock().await.get_next_local_host_socket();
+    let mut ports = available_ports(3);
+    let a_socket = ports.get_next_local_host_socket();
+    let b_socket = ports.get_next_local_host_socket();
     // Todo(uriel): Find a better way to pass expected value to the setup
     // 123 is some arbitrary value, we don't check it anyway.
     setup_for_tests(Felt::from(123), a_socket, b_socket, MAX_CONCURRENCY, None).await;
@@ -461,7 +464,7 @@ async fn faulty_client_setup() {
 
 #[tokio::test]
 async fn unconnected_server() {
-    let socket = AVAILABLE_PORTS.lock().await.get_next_local_host_socket();
+    let socket = available_ports(4).get_next_local_host_socket();
     let client = ComponentAClient::new(
         FAST_FAILING_CLIENT_CONFIG,
         &socket.ip().to_string(),
@@ -494,7 +497,7 @@ async fn faulty_server(
 
 #[tokio::test]
 async fn retry_request() {
-    let socket = AVAILABLE_PORTS.lock().await.get_next_local_host_socket();
+    let socket = available_ports(5).get_next_local_host_socket();
     // Spawn a server that responses with OK every other request.
     task::spawn(async move {
         let should_send_ok = Arc::new(Mutex::new(false));
diff --git a/crates/apollo_infra/src/tests/server_metrics_test.rs b/crates/apollo_infra/src/tests/server_metrics_test.rs
@@ -33,8 +33,8 @@ use crate::component_server::{
     RemoteComponentServer,
 };
 use crate::tests::{
+    available_ports,
     dummy_remote_server_config,
-    AVAILABLE_PORTS,
     TEST_LOCAL_CLIENT_METRICS,
     TEST_LOCAL_SERVER_METRICS,
     TEST_REMOTE_CLIENT_METRICS,
@@ -173,7 +173,7 @@ async fn setup_remote_server_test(
     max_concurrency: usize,
 ) -> (Arc<Semaphore>, RemoteTestComponentClient) {
     let (test_sem, local_client) = setup_local_server_test().await;
-    let socket = AVAILABLE_PORTS.lock().await.get_next_local_host_socket();
+    let socket = available_ports(6).get_next_local_host_socket();
     let config = RemoteClientConfig::default();
 
     let mut remote_server = RemoteComponentServer::new(
