Skip to content

apollo_infra: concurrent tests #11500

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
victorkstarkware merged 1 commit into from
Jan 8, 2026
Merged

apollo_infra: concurrent tests #11500

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .config/nextest.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions crates/apollo_infra/src/tests/concurrent_servers_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,8 @@ use crate::component_server::{
RemoteComponentServer,
};
use crate::tests::{
available_ports_factory,
dummy_remote_server_config,
AVAILABLE_PORTS,
TEST_LOCAL_CLIENT_METRICS,
TEST_LOCAL_SERVER_METRICS,
TEST_REMOTE_CLIENT_METRICS,
Expand Down Expand Up @@ -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_factory(7).get_next_local_host_socket();
let remote_client_config = RemoteClientConfig::default();

let max_concurrency = 10;
Expand Down
15 changes: 8 additions & 7 deletions crates/apollo_infra/src/tests/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -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 to allow running tests concurrently in different processes, which do not have a
// shared memory.
fn available_ports_factory(instance_index: u16) -> AvailablePorts {
AvailablePorts::new(TestIdentifier::InfraUnitTests.into(), instance_index)
}

#[derive(Serialize, Deserialize, Clone, AsRefStr, EnumDiscriminants)]
#[strum_discriminants(
Expand Down
23 changes: 13 additions & 10 deletions crates/apollo_infra/src/tests/remote_component_client_server_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ use crate::component_server::{
};
use crate::serde_utils::SerdeWrapper;
use crate::tests::{
available_ports_factory,
dummy_remote_server_config,
test_a_b_functionality,
ComponentA,
Expand All @@ -55,7 +56,6 @@ use crate::tests::{
ResultB,
ValueA,
ValueB,
AVAILABLE_PORTS,
TEST_LOCAL_CLIENT_METRICS,
TEST_LOCAL_SERVER_METRICS,
TEST_REMOTE_CLIENT_METRICS,
Expand Down Expand Up @@ -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_factory(0).get_next_local_host_socket();
task::spawn(async move {
async fn handler<T>(
_http_request: Request<Body>,
Expand Down Expand Up @@ -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 available_ports = available_ports_factory(1);
let a_socket = available_ports.get_next_local_host_socket();
let b_socket = available_ports.get_next_local_host_socket();

// Shared semaphore used inside ComponentA::handle_request
let semaphore = Arc::new(Semaphore::new(0));
Expand Down Expand Up @@ -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 available_ports = available_ports_factory(2);
let a_socket = available_ports.get_next_local_host_socket();
let b_socket = available_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();
Expand All @@ -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 available_ports = available_ports_factory(3);
let a_socket = available_ports.get_next_local_host_socket();
let b_socket = available_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;
Expand Down Expand Up @@ -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_factory(4).get_next_local_host_socket();
let client = ComponentAClient::new(
FAST_FAILING_CLIENT_CONFIG,
&socket.ip().to_string(),
Expand Down Expand Up @@ -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_factory(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));
Expand Down
4 changes: 2 additions & 2 deletions crates/apollo_infra/src/tests/server_metrics_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,8 @@ use crate::component_server::{
RemoteComponentServer,
};
use crate::tests::{
available_ports_factory,
dummy_remote_server_config,
AVAILABLE_PORTS,
TEST_LOCAL_CLIENT_METRICS,
TEST_LOCAL_SERVER_METRICS,
TEST_REMOTE_CLIENT_METRICS,
Expand Down Expand Up @@ -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_factory(6).get_next_local_host_socket();
let config = RemoteClientConfig::default();

let mut remote_server = RemoteComponentServer::new(
Expand Down