From 40c089f31beab39e5e3b7f9a768fa64ecd4ca2ff Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sat, 18 Jul 2026 10:41:51 +0800 Subject: [PATCH] feat(server): add opt-in global connection cap to the S3 accept loop (#4957) Follow-up to backlog#1191 (optional sub-item). The main accept loop spawned one task per socket with no global bound, so a connection flood could exhaust file descriptors and memory and take existing traffic down with it. - New RUSTFS_API_MAX_CONNECTIONS (default 0 = unlimited, no semaphore constructed, accept loop unchanged). - When set, the loop acquires an owned semaphore permit BEFORE accept(): at saturation it simply stops accepting and lets the kernel backlog absorb bursts (TCP-native backpressure) instead of accept-then-close churn. The permit moves into the connection task and is released by RAII on any exit path, including TLS handshake failures. - Saturation is observable via the rustfs_http_server_connection_cap_saturated_total counter, a connection_cap_state startup event, and a per-wait debug event; shutdown stays responsive while parked on the semaphore. - The cap covers everything on the main listener (S3, admin, console, internode gRPC); the constant docs carry sizing guidance. - E2E coverage: ten sequential Connection-close requests against cap 2 prove permits never leak; two stalled connections saturating cap 2 leave a third unserved until their permits are released, after which the queued request is accepted and answered. --- crates/config/src/constants/api.rs | 17 +++ crates/e2e_test/src/connection_cap_test.rs | 127 +++++++++++++++++++++ crates/e2e_test/src/lib.rs | 4 + rustfs/src/server/http.rs | 77 ++++++++++++- 4 files changed, 224 insertions(+), 1 deletion(-) create mode 100644 crates/e2e_test/src/connection_cap_test.rs diff --git a/crates/config/src/constants/api.rs b/crates/config/src/constants/api.rs index 7ba69efdd..e6fb4be46 100644 --- a/crates/config/src/constants/api.rs +++ b/crates/config/src/constants/api.rs @@ -73,3 +73,20 @@ pub const ENV_API_RATE_LIMIT_BUCKET_BURST: &str = "RUSTFS_API_RATE_LIMIT_BUCKET_ /// Default for `RUSTFS_API_RATE_LIMIT_BUCKET_BURST` (`0` = same as bucket RPM). pub const DEFAULT_API_RATE_LIMIT_BUCKET_BURST: u32 = 0; + +/// Maximum concurrently served connections on the main API listener. +/// +/// `0` (the default) means unlimited. When set, the accept loop stops +/// accepting once the cap is reached and lets the kernel backlog absorb +/// bursts, releasing capacity as connections close. This bounds file +/// descriptor and memory usage under a connection flood. +/// +/// The cap covers everything on the main listener — S3, admin, console, +/// and internode gRPC — so size it well above peer-node count plus the +/// expected client concurrency. +/// Environment variable: RUSTFS_API_MAX_CONNECTIONS +/// Example: RUSTFS_API_MAX_CONNECTIONS=10000 +pub const ENV_API_MAX_CONNECTIONS: &str = "RUSTFS_API_MAX_CONNECTIONS"; + +/// Default for `RUSTFS_API_MAX_CONNECTIONS` (`0` = unlimited). +pub const DEFAULT_API_MAX_CONNECTIONS: usize = 0; diff --git a/crates/e2e_test/src/connection_cap_test.rs b/crates/e2e_test/src/connection_cap_test.rs new file mode 100644 index 000000000..bb835a712 --- /dev/null +++ b/crates/e2e_test/src/connection_cap_test.rs @@ -0,0 +1,127 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! E2E coverage for the opt-in global connection cap on the main API listener +//! (backlog#1191 follow-up, `RUSTFS_API_MAX_CONNECTIONS`): permits must be +//! released when connections close (no leak), and the cap must actually bound +//! concurrency — a queued connection is served only after a held one closes. + +use crate::common::{RustFSTestEnvironment, init_logging}; +use serial_test::serial; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::time::timeout; + +type TestResult = Result<(), Box>; + +/// Open a TCP connection and write one unauthenticated `GET /` (any response, +/// e.g. 403, proves the connection was accepted and served). +async fn open_and_request(addr: &str, connection: &str) -> std::io::Result { + let mut stream = TcpStream::connect(addr).await?; + let request = format!("GET / HTTP/1.1\r\nHost: {addr}\r\nConnection: {connection}\r\n\r\n"); + stream.write_all(request.as_bytes()).await?; + Ok(stream) +} + +/// Read until the response head is complete, or `None` on timeout/close — +/// a `None` on an open socket means the connection sits unaccepted in the +/// kernel backlog behind the cap. +async fn read_response_head(stream: &mut TcpStream, dur: Duration) -> Option { + let deadline = tokio::time::Instant::now() + dur; + let mut buf = vec![0u8; 4096]; + let mut collected = String::new(); + loop { + let remaining = deadline.checked_duration_since(tokio::time::Instant::now())?; + match timeout(remaining, stream.read(&mut buf)).await { + Ok(Ok(0)) | Ok(Err(_)) | Err(_) => return None, + Ok(Ok(n)) => { + collected.push_str(&String::from_utf8_lossy(&buf[..n])); + if collected.contains("\r\n\r\n") { + return Some(collected); + } + } + } + } +} + +#[tokio::test] +#[serial] +async fn connection_cap_releases_permits_on_close() -> TestResult { + init_logging(); + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server_with_env(vec![], &[("RUSTFS_API_MAX_CONNECTIONS", "2")]) + .await?; + + // Ten sequential connections against cap 2: if permits leaked, the third + // request would already hang in the backlog and time out. + for i in 0..10 { + let mut stream = open_and_request(&env.address, "close").await?; + let head = read_response_head(&mut stream, Duration::from_secs(10)) + .await + .unwrap_or_else(|| panic!("request {i} got no response — a connection permit leaked")); + assert!(head.starts_with("HTTP/1.1"), "request {i} unexpected response: {head}"); + } + + Ok(()) +} + +/// Open a TCP connection and send an INCOMPLETE request head. Once accepted +/// it pins a connection permit: hyper waits for the rest of the head (75s +/// default header timeout) until we close the socket. +async fn open_and_stall(addr: &str) -> std::io::Result { + let mut stream = TcpStream::connect(addr).await?; + stream + .write_all(format!("GET / HTTP/1.1\r\nHost: {addr}\r\n").as_bytes()) + .await?; + Ok(stream) +} + +#[tokio::test] +#[serial] +async fn connection_cap_blocks_excess_connections_until_permits_free() -> TestResult { + init_logging(); + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server_with_env(vec![], &[("RUSTFS_API_MAX_CONNECTIONS", "2")]) + .await?; + // Let the readiness poller's pooled connection close and free its permit. + tokio::time::sleep(Duration::from_secs(1)).await; + + // Two stalled connections saturate cap 2 (a served-and-closed connection + // would release its permit immediately, so stalling is what makes the + // occupancy deterministic). + let stalled_a = open_and_stall(&env.address).await?; + let stalled_b = open_and_stall(&env.address).await?; + tokio::time::sleep(Duration::from_millis(300)).await; + + // A complete request now sits in the kernel backlog: connect() succeeds + // but no permit is available, so no response arrives. + let mut blocked = open_and_request(&env.address, "close").await?; + assert!( + read_response_head(&mut blocked, Duration::from_secs(3)).await.is_none(), + "cap 2 with two stalled connections must leave the third unserved" + ); + + // Dropping the stalled connections releases their permits (hyper sees + // EOF while reading the head); the queued request's bytes already sit in + // the socket buffer, so it must now be accepted and served. + drop(stalled_a); + drop(stalled_b); + let head = read_response_head(&mut blocked, Duration::from_secs(10)) + .await + .expect("queued connection must be served after permits are released"); + assert!(head.starts_with("HTTP/1.1"), "unexpected response: {head}"); + + Ok(()) +} diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index aeb77892e..9e0ae89c2 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -83,6 +83,10 @@ mod security_boundary_test; #[cfg(test)] mod api_rate_limit_test; +// Opt-in global connection cap on the main listener (backlog#1191 follow-up) +#[cfg(test)] +mod connection_cap_test; + // Admin authorization gate: non-admin denial + root-credential lifecycle (backlog#1151 sec-4) #[cfg(test)] mod admin_auth_test; diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 2b873fd51..3ac438caa 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -71,6 +71,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::task::{Context, Poll}; use std::time::Duration; use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use tonic::service::interceptor::InterceptedService; use tonic::{Request, Status}; use tower::{Service, ServiceBuilder}; @@ -92,6 +93,7 @@ const METRIC_HTTP_SERVER_REQUEST_BODY_BYTES_TOTAL: &str = "rustfs_http_server_re const METRIC_HTTP_SERVER_REQUEST_BODY_SIZE_BYTES: &str = "rustfs_http_server_request_body_size_bytes"; const METRIC_HTTP_SERVER_RESPONSE_BODY_BYTES_TOTAL: &str = "rustfs_http_server_response_body_bytes_total"; const METRIC_HTTP_SERVER_RESPONSE_BODY_SIZE_BYTES: &str = "rustfs_http_server_response_body_size_bytes"; +const METRIC_HTTP_SERVER_CONNECTION_CAP_SATURATED_TOTAL: &str = "rustfs_http_server_connection_cap_saturated_total"; /// Cached handle for the per-response-body-chunk byte counter. A streamed GET /// emits many chunks, so resolving the `counter!` registry entry once — the @@ -113,6 +115,7 @@ const EVENT_HTTP_STARTUP_ENDPOINTS: &str = "http_startup_endpoints"; const EVENT_HTTP_HOST_ROUTING: &str = "http_host_routing"; const EVENT_HTTP_COMPRESSION_STATE: &str = "http_compression_state"; const EVENT_API_RATE_LIMIT_STATE: &str = "api_rate_limit_state"; +const EVENT_CONNECTION_CAP_STATE: &str = "connection_cap_state"; const EVENT_HTTP_TRANSPORT_PARAMETERS: &str = "http_transport_parameters"; const EVENT_HTTP_ACCEPT_LOOP_STATE: &str = "http_accept_loop_state"; const EVENT_HTTP_CONNECTION_DRAIN: &str = "http_connection_drain"; @@ -737,6 +740,33 @@ pub async fn start_http_server( } } + // Global connection cap (backlog#1191 follow-up sub-item): bounds the + // number of concurrently served connections on this listener so a + // connection flood cannot exhaust file descriptors or memory. `None` + // (the default, RUSTFS_API_MAX_CONNECTIONS=0) leaves the accept loop + // unchanged. + let max_connections = + rustfs_utils::get_env_usize(rustfs_config::ENV_API_MAX_CONNECTIONS, rustfs_config::DEFAULT_API_MAX_CONNECTIONS); + let connection_limiter = (max_connections > 0).then(|| Arc::new(Semaphore::new(max_connections.min(Semaphore::MAX_PERMITS)))); + if connection_limiter.is_some() { + info!( + event = EVENT_CONNECTION_CAP_STATE, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_TRANSPORT, + state = "enabled", + max_connections, + "Connection cap state changed" + ); + } else { + debug!( + event = EVENT_CONNECTION_CAP_STATE, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_TRANSPORT, + state = "disabled", + "Connection cap state changed" + ); + } + let is_console = config.console_enable; let server_domains_configured = !config.server_domains.is_empty(); let task_handle = tokio::spawn(async move { @@ -830,6 +860,47 @@ pub async fn start_http_server( loop { trace!("Waiting for new connection"); + + // Connection cap: acquire a permit BEFORE accepting, so that at + // saturation the loop stops accepting and the kernel backlog + // absorbs bursts (TCP-native backpressure) instead of accept-then- + // close churn. The permit travels into the connection task and is + // released by RAII when the connection ends. + let connection_permit = match &connection_limiter { + None => None, + Some(semaphore) => match semaphore.clone().try_acquire_owned() { + Ok(permit) => Some(permit), + Err(_) => { + counter!(METRIC_HTTP_SERVER_CONNECTION_CAP_SATURATED_TOTAL).increment(1); + debug!( + event = EVENT_CONNECTION_CAP_STATE, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_TRANSPORT, + state = "saturated", + "Connection cap state changed" + ); + tokio::select! { + permit = semaphore.clone().acquire_owned() => match permit { + Ok(permit) => Some(permit), + // The semaphore is never closed; fail safe by + // stopping the accept loop if it ever is. + Err(_) => break, + }, + _ = shutdown_rx.recv() => { + info!( + event = EVENT_HTTP_ACCEPT_LOOP_STATE, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_TRANSPORT, + state = "shutdown_signal_received", + "HTTP accept loop state changed" + ); + break; + } + } + } + }, + }; + let (socket, _) = { #[cfg(unix)] { @@ -930,7 +1001,7 @@ pub async fn start_http_server( rate_limit_layer: api_rate_limit_layer.clone(), }; - process_connection(socket, tls_acceptor.clone(), connection_ctx, graceful.watcher()); + process_connection(socket, tls_acceptor.clone(), connection_ctx, graceful.watcher(), connection_permit); } let active_connections = graceful.count(); @@ -1160,8 +1231,12 @@ fn process_connection( tls_acceptor: Option>, context: ConnectionContext, graceful: Watcher, + connection_permit: Option, ) { tokio::spawn(async move { + // Hold the connection-cap permit for the lifetime of this task; RAII + // release covers TLS handshake failures and normal close alike. + let _connection_permit = connection_permit; let ConnectionContext { http_server, s3_service,