24 file desc per test

Author: enisdenjo

Cargo.lock

@@ -40,3 +40,4 @@ mockito = { workspace = true }
 tempfile = "3.23.0"
 hex = "0.4"
 tiny_http = "0.12"
+libc = "0.2"

e2e/Cargo.toml

@@ -1,6 +1,7 @@
 use axum;
 use bytes::Bytes;
 use dashmap::DashMap;
+use lazy_static::lazy_static;
 use ntex::{
     client::ClientResponse,
     io::Sealed,
@@ -14,9 +15,36 @@ use std::{
     time::Duration,
 };
 use tempfile::NamedTempFile;
-use tokio::{net::TcpListener, sync::oneshot};
+use tokio::{
+    net::TcpListener,
+    sync::{oneshot, Semaphore},
+};
 use tracing::{info, warn};
 
+lazy_static! {
+    /// Limits concurrent test routers to avoid hitting the OS open-file-descriptor limit.
+    /// Hive Router and the subgraphs opens about 24 file descriptors per instance (tokio
+    /// for the event queue, dyn libs, port bindings, domain sockets), so we divide the system's
+    /// `RLIMIT_NOFILE` by that number to get a safe concurrency limit. Increasing the OS ulimit
+    /// will increase the concurrency of tests.
+    static ref CONCURRENCY_SEMAPHORE: Arc<Semaphore> = {
+        let limit = {
+            let mut rlimit = libc::rlimit {
+                rlim_cur: 0,
+                rlim_max: 0,
+            };
+            let nofile = if unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut rlimit) } == 0 {
+                rlimit.rlim_cur as usize
+            } else {
+                256 // fallback, about the default ulimit on many sysstms
+            };
+            (nofile / 24).max(1)
+        };
+        info!("Concurrency semaphore initialized with {} permits", limit);
+        Arc::new(Semaphore::new(limit))
+    };
+}
+
 use hive_router::{
     background_tasks::BackgroundTasksManager, configure_app_from_config, configure_ntex_app,
     init_rustls_crypto_provider, telemetry::Telemetry,
@@ -393,6 +421,11 @@ pub struct TestRouter<'subgraphs, State> {
 
 impl<'subgraphs> TestRouter<'subgraphs, Built> {
     pub async fn start(mut self) -> TestRouter<'subgraphs, Started> {
+        let permit = Arc::clone(&CONCURRENCY_SEMAPHORE)
+            .acquire_owned()
+            .await
+            .expect("concurrency semaphore closed");
+
         init_rustls_crypto_provider();
         let config = self.config.take().unwrap();
         let (telemetry, subscriber) =
@@ -467,6 +500,7 @@ impl<'subgraphs> TestRouter<'subgraphs, Built> {
 
         let mut hold_until_drop = self._hold_until_drop;
         hold_until_drop.push(Box::new(subscription_guard));
+        hold_until_drop.push(Box::new(permit));
 
         TestRouter {
             graphql_path: self.graphql_path,