apollo_infra: add tcp keepalive to remote server accepted sockets

Mirror the client-side SO_KEEPALIVE behaviour on the server: set TCP
keepalive probes on each accepted socket via a configurable idle time
(tcp_keepalive_idle_time_ms). This lets the OS detect dead clients that
disappear without sending FIN/RST.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Itay-Tsabary-Starkware

crates/apollo_infra/Cargo.toml

@@ -28,6 +28,7 @@ metrics-exporter-prometheus.workspace = true
 rand.workspace = true
 rstest.workspace = true
 serde = { workspace = true, features = ["derive"] }
+socket2.workspace = true
 serde_json.workspace = true
 starknet_api.workspace = true
 static_assertions.workspace = true

crates/apollo_infra/src/component_client/remote_component_client.rs

@@ -18,7 +18,6 @@ use hyper_util::client::legacy::Client;
 use hyper_util::rt::TokioExecutor;
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
-use static_assertions::const_assert;
 use tokio::sync::Mutex;
 use tokio::time::Instant;
 use tracing::field::{display, Empty};
@@ -32,6 +31,7 @@ use crate::component_definitions::{
     ServerError,
     APPLICATION_OCTET_STREAM,
     REQUEST_ID_HEADER,
+    TCP_KEEPALIVE_FACTOR,
 };
 use crate::metrics::RemoteClientMetrics;
 use crate::requests::LabeledRequest;
@@ -45,10 +45,6 @@ pub const DEFAULT_RETRIES: usize = 15;
 pub const REQUEST_TIMEOUT_ERROR_MESSAGE: &str = "request timed out";
 
 const DEFAULT_IDLE_CONNECTIONS: usize = 10;
-pub(crate) const TCP_IDLE_TIMEOUT_FACTOR: f64 = 1.5;
-// Ensure tcp connection timeout is greater than http2 connection timeout by requiring a factor
-// greater than 1.
-const_assert!(TCP_IDLE_TIMEOUT_FACTOR > 1.0);
 
 // 8 MiB — bounds memory materialized from a single response as defense in depth.
 const DEFAULT_MAX_RESPONSE_BODY_BYTES: usize = 8 * 1024 * 1024;
@@ -66,8 +62,8 @@ pub struct RemoteClientConfig {
     pub retries: usize,
     pub idle_connections: usize,
     // Determines client connection timeouts. Used plainly for HTTP/2 connections, and with a
-    // `TCP_IDLE_TIMEOUT_FACTOR` for TCP connections.
-    #[validate(custom(function = "validate_tcp_exceeds_http_keepalive"))]
+    // `TCP_KEEPALIVE_FACTOR` for TCP connections.
+    #[validate(custom(function = "validate_keepalive_timeout_ms"))]
     pub keepalive_timeout_ms: u64,
     pub attempts_per_log: usize,
     pub initial_retry_delay_ms: u64,
@@ -96,24 +92,39 @@ impl Default for RemoteClientConfig {
 }
 
 /// Validates that the TCP keepalive duration (at second granularity, as the OS stores
-/// `TCP_KEEPIDLE` in whole seconds) is greater than or equal to the HTTP keepalive duration
-/// (millisecond granularity). If the configured `keepalive_timeout_ms * TCP_IDLE_TIMEOUT_FACTOR` is
-/// less than 1 second, truncation to whole seconds yields 0 s, making the TCP keepalive shorter
-/// than the HTTP keepalive.
-fn validate_tcp_exceeds_http_keepalive(keepalive_timeout_ms: u64) -> Result<(), ValidationError> {
+/// `TCP_KEEPIDLE` in whole seconds) is positive and greater than or equal to the HTTP keepalive
+/// duration (millisecond granularity). If the configured
+/// `keepalive_timeout_ms * TCP_KEEPALIVE_FACTOR` rounds to 0 s, the kernel rejects the socket
+/// option with `EINVAL`. If it is less than `keepalive_timeout_ms`, the TCP keepalive fires before
+/// the HTTP-level timeout.
+pub(crate) fn validate_keepalive_timeout_ms(
+    keepalive_timeout_ms: u64,
+) -> Result<(), ValidationError> {
     let http_keepalive = Duration::from_millis(keepalive_timeout_ms);
-    let tcp_keepalive_raw = http_keepalive.mul_f64(TCP_IDLE_TIMEOUT_FACTOR);
+    let tcp_keepalive_raw = http_keepalive.mul_f64(TCP_KEEPALIVE_FACTOR);
     // TCP_KEEPIDLE is stored in whole seconds; fractional seconds are truncated by the OS.
-    let tcp_keepalive = Duration::from_secs(tcp_keepalive_raw.as_secs());
+    let tcp_keepalive_secs = tcp_keepalive_raw.as_secs();
+    if tcp_keepalive_secs == 0 {
+        return Err(create_validation_error(
+            format!(
+                "TCP keepalive rounds to 0 s (keepalive_timeout_ms={keepalive_timeout_ms}, \
+                 factor={TCP_KEEPALIVE_FACTOR}): increase keepalive_timeout_ms so that \
+                 keepalive_timeout_ms * {TCP_KEEPALIVE_FACTOR} is at least 1 s",
+            ),
+            "tcp_keepalive_zero",
+            "TCP keepalive (second granularity) must be > 0 s.",
+        ));
+    }
+    let tcp_keepalive = Duration::from_secs(tcp_keepalive_secs);
     if tcp_keepalive >= http_keepalive {
         Ok(())
     } else {
         Err(create_validation_error(
             format!(
-                "TCP keepalive ({} s) is shorter than HTTP keepalive ({keepalive_timeout_ms} ms): \
-                 increase keepalive_timeout_ms so that keepalive_timeout_ms * \
-                 {TCP_IDLE_TIMEOUT_FACTOR} rounds to at least {keepalive_timeout_ms} ms",
-                tcp_keepalive.as_secs(),
+                "TCP keepalive ({tcp_keepalive_secs} s) is shorter than HTTP keepalive \
+                 ({keepalive_timeout_ms} ms): increase keepalive_timeout_ms so that \
+                 keepalive_timeout_ms * {TCP_KEEPALIVE_FACTOR} rounds to at least \
+                 {keepalive_timeout_ms} ms",
             ),
             "tcp_keepalive_shorter_than_http_keepalive",
             "TCP keepalive (second granularity) must be >= HTTP keepalive (ms granularity).",
@@ -230,7 +241,7 @@ where
         let mut connector = HttpConnector::new();
         connector.set_nodelay(config.set_tcp_nodelay);
         connector.set_connect_timeout(Some(Duration::from_millis(config.connection_timeout_ms)));
-        connector.set_keepalive(Some(idle_timeout.mul_f64(TCP_IDLE_TIMEOUT_FACTOR)));
+        connector.set_keepalive(Some(idle_timeout.mul_f64(TCP_KEEPALIVE_FACTOR)));
 
         // Create the HTTP/2 client.
         let client = Client::builder(TokioExecutor::new())

crates/apollo_infra/src/component_client/remote_component_client_test.rs

@@ -18,8 +18,8 @@ fn tcp_keepalive_validation_passes_when_tcp_equals_http_keepalive() {
     assert!(config.validate().is_ok());
 }
 
-// keepalive_timeout_ms = 100 ms: tcp_raw = 150 ms, tcp_whole_secs = 0 s = 0 ms, which is less than
-// http_keepalive = 100 ms.
+// keepalive_timeout_ms = 100 ms: tcp_raw = 150 ms, tcp_whole_secs = 0 s = 0 ms, which is less
+// than http_keepalive = 100 ms.
 #[test]
 fn tcp_keepalive_validation_fails_when_tcp_truncated_below_http_keepalive() {
     let config = RemoteClientConfig { keepalive_timeout_ms: 100, ..Default::default() };

crates/apollo_infra/src/component_definitions.rs

@@ -6,6 +6,7 @@ use derive_more::{Display, FromStr};
 use rand::random;
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
+use static_assertions::const_assert;
 use thiserror::Error;
 use tokio::sync::mpsc::{Receiver, Sender};
 use tokio::time::Instant;
@@ -17,6 +18,11 @@ use crate::requests::LabeledRequest;
 pub(crate) const APPLICATION_OCTET_STREAM: &str = "application/octet-stream";
 pub const BUSY_PREVIOUS_REQUESTS_MSG: &str = "Server is busy addressing previous requests";
 
+pub(crate) const TCP_KEEPALIVE_FACTOR: f64 = 1.5;
+// Ensure tcp connection timeout is greater than http2 connection timeout by requiring a factor
+// greater than 1.
+const_assert!(TCP_KEEPALIVE_FACTOR > 1.0);
+
 #[async_trait]
 pub trait ComponentRequestHandler<Request, Response> {
     async fn handle_request(&mut self, request: Request) -> Response;

crates/apollo_infra/src/component_server/remote_component_server.rs

@@ -20,11 +20,13 @@ use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
 use hyper_util::server::conn::auto::Builder as ServerBuilder;
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
+use socket2::{SockRef, TcpKeepalive};
 use tokio::net::TcpListener;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 use tracing::{debug, error, instrument, trace, warn};
 use validator::Validate;
 
+use crate::component_client::remote_component_client::validate_keepalive_timeout_ms;
 use crate::component_client::{ClientError, LocalComponentClient};
 use crate::component_definitions::{
     ComponentClient,
@@ -33,6 +35,7 @@ use crate::component_definitions::{
     APPLICATION_OCTET_STREAM,
     BUSY_PREVIOUS_REQUESTS_MSG,
     REQUEST_ID_HEADER,
+    TCP_KEEPALIVE_FACTOR,
 };
 use crate::component_server::ComponentServerStarter;
 use crate::metrics::RemoteServerMetrics;
@@ -80,6 +83,7 @@ pub struct RemoteServerConfig {
     pub max_concurrency: usize,
     pub max_request_body_bytes: usize,
     pub keepalive_interval_ms: u64,
+    #[validate(custom(function = "validate_keepalive_timeout_ms"))]
     pub keepalive_timeout_ms: u64,
 }
 
@@ -393,6 +397,10 @@ where
             panic!("Failed to bind remote component server socket {:#?}: {e}", bind_socket)
         });
 
+        let max_streams = self.config.max_streams_per_connection;
+        let keepalive_interval = Duration::from_millis(self.config.keepalive_interval_ms);
+        let keepalive_timeout = Duration::from_millis(self.config.keepalive_timeout_ms);
+
         loop {
             let (stream, peer_addr) = match listener.accept().await {
                 Ok(conn) => conn,
@@ -407,10 +415,13 @@ where
                 warn!("Failed to set TCP_NODELAY: {e}");
             }
 
+            let tcp_keepalive =
+                TcpKeepalive::new().with_time(keepalive_timeout.mul_f64(TCP_KEEPALIVE_FACTOR));
+            if let Err(e) = SockRef::from(&stream).set_tcp_keepalive(&tcp_keepalive) {
+                error!("Failed to set TCP keepalive: {e}");
+            }
+
             let io = TokioIo::new(stream);
-            let max_streams = self.config.max_streams_per_connection;
-            let keepalive_interval = Duration::from_millis(self.config.keepalive_interval_ms);
-            let keepalive_timeout = Duration::from_millis(self.config.keepalive_timeout_ms);
 
             tokio::spawn(per_connection_service(
                 io,