From 1bbb9c237c69ae2bf7051b48267b0bf8990b4c0e Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 13 Sep 2026 07:46:58 +0000 Subject: [PATCH] Share connection semaphore across listeners (review #010 C4) --- README.md | 2 +- docs/architecture/config.md | 8 +- docs/reviews/010-emfile-accept-loop-flood.md | 37 ++++---- src/main.rs | 12 ++- src/server.rs | 58 ++++++++++++- tests/integration_test.rs | 91 +++++++++++++++++++- 6 files changed, 183 insertions(+), 25 deletions(-) diff --git a/README.md b/README.md index 7898c10..49c0b0b 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,7 @@ Configuration uses TOML and is split into **static** (requires restart) and | `shutdown_timeout_secs` | `30` | Graceful shutdown timeout | | `connection_idle_timeout_secs` | `60` | Server-side idle timeout for client TLS connections (prevents FD exhaustion from abandoned connections) | | `tls_handshake_timeout_secs` | `10` | Max seconds to complete the TLS handshake; stalled handshakes are closed (prevents slowloris FD/permit exhaustion) | -| `max_connections` | `1024` | Max concurrent client TLS connections (backpressure via semaphore) | +| `max_connections` | `1024` | Max concurrent client TLS connections, shared across all listeners (global semaphore) | | `logging.level` | `"info"` | Log level | | `logging.format` | `"text"` | Log format (`"text"` or `"json"`) | | `logging.log_file_path` | (not set) | Path to log file for fail2ban | diff --git a/docs/architecture/config.md b/docs/architecture/config.md index 1dcf9d3..0bbdb51 100644 --- a/docs/architecture/config.md +++ b/docs/architecture/config.md @@ -92,7 +92,7 @@ Immutable after startup. Changes require a process restart. | `shutdown_timeout_secs` | `u64` | Maximum seconds to wait for in-flight requests during graceful shutdown (default: `30`) | | `connection_idle_timeout_secs` | `u64` | Server-side idle timeout for client TLS connections. Idle HTTP/2 connections are closed after this duration (with keep-alive pings at 15s intervals to detect dead peers). HTTP/1.1 connections are closed if the client doesn't send a complete request header within this duration. Prevents FD exhaustion from abandoned connections (default: `60`; must be > 0; see review #007 C1) | | `tls_handshake_timeout_secs` | `u64` | Maximum seconds a client may take to complete the TLS handshake. Stalled handshakes (e.g. crawlers/slowloris clients that open a TCP connection but never send a ClientHello) are closed after this duration, releasing the FD and connection slot. Without it, a stalled handshake holds an FD + connection semaphore permit indefinitely (default: `10`; must be > 0; see review #010 C3) | -| `max_connections` | `usize` | Maximum number of concurrent client TLS connections. When the limit is reached, new connections wait in the OS TCP backlog until a slot frees (default: `1024`; must be > 0; see review #007 C2) | +| `max_connections` | `usize` | Maximum number of concurrent client TLS connections **process-wide**. The connection semaphore is shared across all HTTPS listeners, so this is a global cap — not a per-listener cap. When the limit is reached, new connections wait in the OS TCP backlog until a slot frees (default: `1024`; must be > 0; see review #007 C2, review #010 C4) | | `logging` | `LoggingConfig` | Logging configuration (see below) | **LoggingConfig** (nested in `[logging]` TOML section): @@ -316,7 +316,7 @@ health_check_port = 9900 # Local health check (0 to disable) admin_key_path = "/etc/reverse-proxy/admin-key" # Empty string to disable # connection_idle_timeout_secs = 60 # Server-side idle timeout (default: 60) # tls_handshake_timeout_secs = 10 # TLS handshake timeout (default: 10) -# max_connections = 1024 # Max concurrent TLS connections (default: 1024) +# max_connections = 1024 # Max concurrent TLS connections, process-wide across all listeners (default: 1024) [logging] level = "info" @@ -463,7 +463,9 @@ On startup, the config is validated: the server-side idle timeout, reintroducing the FD exhaustion bug from review #007 C1. 22. `max_connections` must be > 0. A zero value would deadlock the connection - semaphore, preventing any client connection from being accepted. + semaphore, preventing any client connection from being accepted. The + semaphore is shared across all listeners (review #010 C4), so + `max_connections` is the process-wide cap rather than per-listener. 23. `tls_handshake_timeout_secs` must be > 0. A zero value would immediately kill every TLS handshake, preventing any client connection from completing (review #010 C3). diff --git a/docs/reviews/010-emfile-accept-loop-flood.md b/docs/reviews/010-emfile-accept-loop-flood.md index b29f80d..a1c0c2a 100644 --- a/docs/reviews/010-emfile-accept-loop-flood.md +++ b/docs/reviews/010-emfile-accept-loop-flood.md @@ -24,9 +24,9 @@ fixes: default 10s) wraps tls_acceptor.accept() in src/server.rs; stalled handshakes release FD + permit. - >- - C4: OPEN — connection semaphore is per-listener (src/server.rs:338), so - the effective cap is max_connections × listeners; sequence before C2 so - the RLIMIT cross-check uses the real FD budget. + C4: FIXED 2026-09-13 — connection semaphore is shared across all + listeners (ConnectionSemaphore, src/server.rs); max_connections is now a + process-wide cap, not max_connections × listeners. - >- C2: OPEN — no startup cross-check that max_connections fits under RLIMIT_NOFILE with headroom. Sequenced after C3/C4. @@ -187,11 +187,13 @@ Notes from implementation: (default 10s) wrapping the accept in `tokio::time::timeout`. This was the likely actual FD-exhaustion vector for slow/held crawler handshakes, and a slowloris amplifier. - - **C4 (open)**: `conn_sem` is per-listener (`main.rs` spawns one - `serve_https_listener` per listener, each creating its own - semaphore), so the effective connection cap is - `max_connections × listeners`. Any C2 RLIMIT cross-check must - account for this (or the semaphore should be shared). + - ~~**C4 (open)**~~ — **FIXED 2026-09-13**: `conn_sem` was per-listener + (`main.rs` spawns one `serve_https_listener` per listener, each creating + its own semaphore), so the effective connection cap was + `max_connections × listeners`. Fixed via a single `ConnectionSemaphore` + (`Arc` wrapper, src/server.rs) created in `main.rs` and shared + by every listener; `max_connections` is now a process-wide cap. This is + the topology the C2 RLIMIT cross-check should assume. ### Original finding (pre-fix, preserved for context) @@ -265,7 +267,7 @@ observed limit). Residual risk after mitigation: none identified for FD exhaustion at current traffic (peak concurrent connections observed ≪ 800); the code -findings C4/C2 remain the durable fix (C1 + C3 landed 2026-09-13). +findings C2 remains the durable fix (C1 + C3 + C4 landed 2026-09-13). ## Traffic-analysis side note (from the same investigation) @@ -298,13 +300,18 @@ behavior, which is exactly what made the EMFILE state reachable. future is dropped, releasing the TCP FD and the semaphore permit; the idle watchdog never needs to run for a stalled handshake. Closes the crawler slow-handshake vector described under "Trigger conditions". -3. Land C4 (shared connection semaphore across listeners) so - `max_connections` is a global cap rather than per-listener. Sequenced - before C2 because the RLIMIT cross-check's FD budget depends on the - final semaphore topology (shared vs per-listener). +3. ~~Land C4 (shared connection semaphore across listeners) so + `max_connections` is a global cap rather than per-listener~~ — DONE + 2026-09-13. `serve_https_listener()` no longer creates its own semaphore; + it receives an `Arc` (new public wrapper in + src/server.rs) built once in `main.rs` and cloned into every listener + task. `max_connections` is now the process-wide concurrent TLS connection + cap; the effective cap no longer scales with listener count. This is the + topology C2's RLIMIT cross-check should assume. 4. Land C2 (RLIMIT cross-check at startup) with a prominent warning or - hard validation error. Must account for C4 (per-listener semaphore - multiplication → shared after C4 lands) when computing the FD budget. + hard validation error. The FD budget assumes a shared semaphore + (landed, see step 3): connection FDs are capped at `max_connections` + process-wide. Lower urgency after M1: with the deploy baseline (`nofile 8192`, `max_connections 800`) the ceiling is ~10% of the limit, so C2 is a validation guard, not an active exposure. diff --git a/src/main.rs b/src/main.rs index 1b827a6..126927c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -17,7 +17,9 @@ use reverse_proxy::health; use reverse_proxy::logging; use reverse_proxy::proxy::{build_router, create_http_client, create_https_client, ProxyState}; use reverse_proxy::rate_limit::{start_eviction_task, RateLimiter}; -use reverse_proxy::server::{drain_in_flight, serve_https_listener, InFlightCounter}; +use reverse_proxy::server::{ + drain_in_flight, serve_https_listener, ConnectionSemaphore, InFlightCounter, +}; use reverse_proxy::shutdown::GracefulShutdown; use reverse_proxy::tls::acceptor::{setup_tls, TlsMode}; use reverse_proxy::tls::redirect; @@ -217,6 +219,12 @@ async fn run_server(loaded_config: cli::LoadedConfig, config_path: &str) -> Resu let app = build_router(proxy_state.clone(), config_arc.clone(), rate_limiter); let in_flight = InFlightCounter::new(); + let conn_sem = ConnectionSemaphore::new(loaded_config.static_config.max_connections); + + info!( + max_connections = conn_sem.max_connections(), + "global connection semaphore created (shared across all listeners)" + ); let mut https_server_handles = Vec::new(); @@ -237,7 +245,7 @@ async fn run_server(loaded_config: cli::LoadedConfig, config_path: &str) -> Resu std::time::Duration::from_secs( loaded_config.static_config.tls_handshake_timeout_secs, ), - loaded_config.static_config.max_connections, + conn_sem.clone(), )); info!( diff --git a/src/server.rs b/src/server.rs index 25e21e6..e7e72ef 100644 --- a/src/server.rs +++ b/src/server.rs @@ -171,6 +171,33 @@ impl InFlightCounter { } } +/// Connection semaphore shared across all HTTPS listeners so that +/// `max_connections` acts as a global cap rather than a per-listener cap +/// (review #010 C4). +pub struct ConnectionSemaphore { + semaphore: Arc, + max_connections: usize, +} + +impl ConnectionSemaphore { + pub fn new(max_connections: usize) -> Arc { + Arc::new(Self { + semaphore: Arc::new(Semaphore::new(max_connections)), + max_connections, + }) + } + + pub fn max_connections(&self) -> usize { + self.max_connections + } + + pub async fn acquire_owned( + &self, + ) -> Result { + self.semaphore.clone().acquire_owned().await + } +} + #[derive(Debug)] struct IdleState { last_activity: Mutex, @@ -354,10 +381,9 @@ pub async fn serve_https_listener( in_flight: Arc, connection_idle_timeout: Duration, tls_handshake_timeout: Duration, - max_connections: usize, + conn_sem: Arc, ) { let local_addr = tcp_listener.local_addr(); - let conn_sem = Arc::new(Semaphore::new(max_connections)); let accept_errors = Arc::new(AcceptErrorReporter::new()); loop { @@ -374,7 +400,6 @@ pub async fn serve_https_listener( let tls_acceptor = tls_acceptor.clone(); let router = router.clone(); let in_flight = in_flight.clone(); - let conn_sem = conn_sem.clone(); let permit = match conn_sem.acquire_owned().await { Ok(permit) => permit, @@ -799,6 +824,33 @@ mod tests { assert_eq!(state.in_flight(), 0); } + #[tokio::test] + async fn connection_semaphore_reports_cap() { + let sem = ConnectionSemaphore::new(7); + assert_eq!(sem.max_connections(), 7); + } + + #[tokio::test] + async fn connection_semaphore_permits_are_shared_across_clones() { + let sem = ConnectionSemaphore::new(2); + + let p1 = sem.acquire_owned().await.unwrap(); + let p2 = sem.acquire_owned().await.unwrap(); + + let clone = Arc::clone(&sem); + let blocked = tokio::time::timeout(Duration::from_millis(100), clone.acquire_owned()).await; + assert!( + blocked.is_err(), + "third acquire through a clone must block once the shared pool of 2 is exhausted" + ); + + drop(p1); + drop(p2); + + let p3 = sem.acquire_owned().await.expect("permits released, acquire succeeds"); + drop(p3); + } + #[test] fn idle_state_touch_updates_last_activity() { let state = IdleState::new(); diff --git a/tests/integration_test.rs b/tests/integration_test.rs index 4cfd290..dbd8d8f 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -1176,6 +1176,21 @@ mod idle_timeout_tests { Arc, tokio::task::JoinHandle<()>, tokio::sync::watch::Sender, + ) { + let conn_sem = reverse_proxy::server::ConnectionSemaphore::new(1024); + spawn_test_https_server(idle_timeout, handshake_timeout, upstream, conn_sem).await + } + + async fn spawn_test_https_server( + idle_timeout: Duration, + handshake_timeout: Duration, + upstream: String, + conn_sem: Arc, + ) -> ( + std::net::SocketAddr, + Arc, + tokio::task::JoinHandle<()>, + tokio::sync::watch::Sender, ) { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); @@ -1195,7 +1210,7 @@ mod idle_timeout_tests { in_flight_clone, idle_timeout, handshake_timeout, - 1024, + conn_sem, ) .await; }); @@ -1539,4 +1554,78 @@ mod idle_timeout_tests { let _ = upstream.shutdown_tx.send(()); } + + // Reproducer for review #010 C4: the connection semaphore must be shared + // across listeners. Two listeners with a shared cap of 2 must admit only + // 2 concurrent connections total — the third connection, hitting either + // listener, must wait until a slot frees. + #[tokio::test] + async fn connection_semaphore_is_shared_across_listeners() { + let conn_sem = reverse_proxy::server::ConnectionSemaphore::new(2); + + let upstream = helpers::http_test_helper::TestUpstream::spawn_ok().await; + let upstream_addr = format!("127.0.0.1:{}", upstream.addr.port()); + + let (addr1, _in_flight1, _h1, _s1) = spawn_test_https_server( + Duration::from_secs(60), + Duration::from_secs(10), + upstream_addr.clone(), + conn_sem.clone(), + ) + .await; + let (addr2, _in_flight2, _h2, _s2) = spawn_test_https_server( + Duration::from_secs(60), + Duration::from_secs(10), + upstream_addr, + conn_sem.clone(), + ) + .await; + + // Two live connections saturate the shared pool: one per listener. + let tls1 = connect_tls(addr1).await; + let tls2 = connect_tls(addr2).await; + + // A third connection (racing against listener 1) must not complete its + // handshake: the accept loop parks it on the shared semaphore before + // TLS ever starts. + let (third_tx, third_rx) = tokio::sync::oneshot::channel(); + let third_task = tokio::spawn(async move { + let stream = connect_tls(addr1).await; + let _ = third_tx.send(stream); + }); + + tokio::time::sleep(Duration::from_millis(500)).await; + assert!( + third_rx.is_empty(), + "third connection must be capped by the shared semaphore" + ); + + // Release one slot; the parked third connection must now get through. + drop(tls2); + + let mut tls3 = tokio::time::timeout(Duration::from_secs(5), third_rx) + .await + .expect("third connection should complete after a permit is released") + .expect("third connection task should not panic"); + + tls3.write_all(b"GET / HTTP/1.1\r\nHost: test.local\r\nConnection: close\r\n\r\n") + .await + .unwrap(); + + let mut buf = vec![0u8; 4096]; + let n = tokio::time::timeout(Duration::from_secs(5), tls3.read(&mut buf)) + .await + .expect("timeout waiting for response") + .expect("read error"); + let response = String::from_utf8_lossy(&buf[..n]); + assert!( + response.starts_with("HTTP/1.1 200 OK"), + "expected a real 200 round-trip on the capped connection, got: {response}" + ); + + third_task.abort(); + drop(tls1); + drop(tls3); + let _ = upstream.shutdown_tx.send(()); + } }