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

@@ -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;

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<dyn std::error::Error + Send + Sync>>;
+
+/// 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<TcpStream> {
+    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<String> {
+    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<TcpStream> {
+    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(())
+}

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;

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<Arc<TlsAcceptorHolder>>,
     context: ConnectionContext,
     graceful: Watcher,
+    connection_permit: Option<OwnedSemaphorePermit>,
 ) {
     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,