apollo_infra: add tcp keepalive to remote client

Set SO_KEEPALIVE on outbound sockets via a configurable idle time
(tcp_keepalive_idle_time_ms). This lets the OS detect dead peers that
disappear without sending FIN/RST.

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

Author: Itay-Tsabary-Starkware

Cargo.lock

@@ -30,6 +30,7 @@ rstest.workspace = true
 serde = { workspace = true, features = ["derive"] }
 serde_json.workspace = true
 starknet_api.workspace = true
+static_assertions.workspace = true
 thiserror.workspace = true
 time = { workspace = true, features = ["macros"] }
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
@@ -38,6 +39,7 @@ tracing.workspace = true
 tracing-subscriber = { workspace = true, features = ["env-filter", "json", "time"] }
 validator.workspace = true
 
+
 [dev-dependencies]
 apollo_infra_utils = { workspace = true, features = ["testing"] }
 apollo_metrics = { workspace = true, features = ["testing"] }
@@ -47,5 +49,6 @@ 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/Cargo.toml

@@ -4,6 +4,7 @@ use std::marker::PhantomData;
 use std::time::Duration;
 
 use apollo_config::dumping::{ser_param, SerializeConfig};
+use apollo_config::validators::create_validation_error;
 use apollo_config::{ParamPath, ParamPrivacyInput, SerializedParam};
 use async_trait::async_trait;
 use bytes::Bytes;
@@ -17,11 +18,12 @@ 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};
 use tracing::{debug, instrument, trace, warn};
-use validator::Validate;
+use validator::{Validate, ValidationError};
 
 use super::definitions::{ClientError, ClientResult};
 use crate::component_definitions::{
@@ -35,10 +37,19 @@ use crate::metrics::RemoteClientMetrics;
 use crate::requests::LabeledRequest;
 use crate::serde_utils::SerdeWrapper;
 
+#[cfg(test)]
+#[path = "remote_component_client_test.rs"]
+mod remote_component_client_test;
+
 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;
 const DEFAULT_IDLE_TIMEOUT_MS: u64 = 30000;
@@ -54,6 +65,9 @@ const DEFAULT_REQUEST_TIMEOUT_MS: u64 = 30_000;
 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"))]
     pub idle_timeout_ms: u64,
     pub attempts_per_log: usize,
     pub initial_retry_delay_ms: u64,
@@ -81,6 +95,32 @@ 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 `idle_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(idle_timeout_ms: u64) -> Result<(), ValidationError> {
+    let http_keepalive = Duration::from_millis(idle_timeout_ms);
+    let tcp_keepalive_raw = http_keepalive.mul_f64(TCP_IDLE_TIMEOUT_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());
+    if tcp_keepalive >= http_keepalive {
+        Ok(())
+    } else {
+        Err(create_validation_error(
+            format!(
+                "TCP keepalive ({} s) is shorter than HTTP keepalive ({idle_timeout_ms} ms): \
+                 increase idle_timeout_ms so that idle_timeout_ms * {TCP_IDLE_TIMEOUT_FACTOR} \
+                 rounds to at least {idle_timeout_ms} ms",
+                tcp_keepalive.as_secs(),
+            ),
+            "tcp_keepalive_shorter_than_http_keepalive",
+            "TCP keepalive (second granularity) must be >= HTTP keepalive (ms granularity).",
+        ))
+    }
+}
+
 impl SerializeConfig for RemoteClientConfig {
     fn dump(&self) -> BTreeMap<ParamPath, SerializedParam> {
         BTreeMap::from_iter([
@@ -184,13 +224,19 @@ where
         metrics: &'static RemoteClientMetrics,
     ) -> Self {
         let uri = format!("http://{url}:{port}/").parse().unwrap();
+        let idle_timeout = Duration::from_millis(config.idle_timeout_ms);
+
+        // Create the tcp connector.
         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)));
+
+        // Create the HTTP/2 client.
         let client = Client::builder(TokioExecutor::new())
             .http2_only(true)
             .pool_max_idle_per_host(config.idle_connections)
-            .pool_idle_timeout(Duration::from_millis(config.idle_timeout_ms))
+            .pool_idle_timeout(idle_timeout)
             .build(connector);
 
         debug!("RemoteComponentClient created with URI: {uri:?}");

crates/apollo_infra/src/component_client/remote_component_client.rs

@@ -0,0 +1,27 @@
+use validator::Validate;
+
+use crate::component_client::RemoteClientConfig;
+
+// idle_timeout_ms = 2000 ms: tcp_raw = 3000 ms, tcp_whole_secs = 3 s = 3000 ms, which is strictly
+// greater than http_keepalive = 2000 ms.
+#[test]
+fn tcp_keepalive_validation_passes_when_tcp_strictly_greater_than_http_keepalive() {
+    let config = RemoteClientConfig { idle_timeout_ms: 2000, ..Default::default() };
+    assert!(config.validate().is_ok());
+}
+
+// idle_timeout_ms = 1000 ms: tcp_raw = 1500 ms, tcp_whole_secs = 1 s = 1000 ms, which equals
+// http_keepalive = 1000 ms.
+#[test]
+fn tcp_keepalive_validation_passes_when_tcp_equals_http_keepalive() {
+    let config = RemoteClientConfig { idle_timeout_ms: 1000, ..Default::default() };
+    assert!(config.validate().is_ok());
+}
+
+// idle_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 { idle_timeout_ms: 100, ..Default::default() };
+    assert!(config.validate().is_err());
+}

crates/apollo_infra/src/component_client/remote_component_client_test.rs

@@ -3,6 +3,7 @@ mod local_component_client_server_test;
 mod local_request_prioritization;
 mod remote_component_client_server_test;
 mod server_metrics_test;
+mod test_utils;
 
 use std::net::IpAddr;
 use std::sync::Arc;

crates/apollo_infra/src/tests/mod.rs

@@ -31,6 +31,7 @@ use tokio::sync::{Mutex, Semaphore};
 use tokio::task;
 use tokio::time::{sleep, timeout};
 
+use crate::component_client::remote_component_client::TCP_IDLE_TIMEOUT_FACTOR;
 use crate::component_client::{
     ClientError,
     ClientResult,
@@ -56,6 +57,7 @@ use crate::component_server::{
     RemoteServerConfig,
 };
 use crate::serde_utils::SerdeWrapper;
+use crate::tests::test_utils::client_socket_keepalive_time;
 use crate::tests::{
     available_ports_factory,
     dummy_remote_server_config,
@@ -734,3 +736,52 @@ async fn zombie_connection_is_evicted() {
          connection is still open"
     );
 }
+
+/// Verifies that `TCP_KEEPIDLE` on the client's outbound socket equals
+/// `idle_timeout_ms * TCP_IDLE_TIMEOUT_FACTOR`, confirming the socket is armed to probe after
+/// exactly the expected idle period.
+///
+/// Internally, hyper's `HttpConnector::set_keepalive` calls `SockRef::set_tcp_keepalive` via
+/// socket2 only when it establishes a TCP connection for a request. The socket therefore only
+/// exists — and the option is only applied — after the first `send`. The test triggers that path,
+/// then reads the option back through the raw file-descriptor scan to confirm config value was
+/// properly applied on the socket.
+///
+/// Note: an end-to-end test that the OS closes the socket after unanswered probes would require
+/// packet-level manipulation (e.g. iptables DROP on loopback) and is out of scope for unit tests.
+#[tokio::test]
+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;
+    let expected_keepalive_idle =
+        Duration::from_millis(IDLE_TIMEOUT_MS).mul_f64(TCP_IDLE_TIMEOUT_FACTOR);
+    assert_eq!(
+        expected_keepalive_idle.subsec_nanos(),
+        0,
+        "IDLE_TIMEOUT_MS * TCP_IDLE_TIMEOUT_FACTOR must be a whole number of seconds"
+    );
+
+    let mut ports = available_ports_factory(unique_u16!());
+    let a_socket = ports.get_next_local_host_socket();
+    let b_socket = ports.get_next_local_host_socket();
+
+    setup_for_tests(VALID_VALUE_A, a_socket, b_socket, MAX_CONCURRENCY, None).await;
+
+    let client = ComponentAClient::new(
+        RemoteClientConfig { idle_timeout_ms: IDLE_TIMEOUT_MS, ..Default::default() },
+        &a_socket.ip().to_string(),
+        a_socket.port(),
+        &TEST_REMOTE_CLIENT_METRICS,
+    );
+
+    // Trigger the lazy TCP connect so the socket exists and keepalive options are applied.
+    client.a_get_value().await.expect("request should succeed");
+
+    let actual_keepalive_idle = client_socket_keepalive_time(a_socket)
+        .expect("SO_KEEPALIVE should be set and TCP_KEEPIDLE should be readable");
+    assert_eq!(
+        actual_keepalive_idle, expected_keepalive_idle,
+        "TCP_KEEPIDLE should equal idle_timeout_ms * TCP_IDLE_TIMEOUT_FACTOR"
+    );
+}

crates/apollo_infra/src/tests/remote_component_client_server_test.rs

@@ -0,0 +1,26 @@
+use std::net::SocketAddr;
+use std::os::unix::io::BorrowedFd;
+use std::time::Duration;
+
+use socket2::SockRef;
+
+/// Returns the `TCP_KEEPIDLE` duration of the outbound socket in this process that is connected
+/// to `server_addr`, or `None` if no such socket is found or `SO_KEEPALIVE` is not enabled.
+pub(crate) fn client_socket_keepalive_time(server_addr: SocketAddr) -> Option<Duration> {
+    for fd in 0_i32..4096 {
+        // SAFETY: We only borrow the fd transiently to read socket options; `SockRef` does
+        // not take ownership of or close the fd. Invalid fds produce errors from `peer_addr`
+        // and `keepalive`, which we handle gracefully via `.ok()` / `.unwrap_or`.
+        let borrowed = unsafe { BorrowedFd::borrow_raw(fd) };
+        let sock = SockRef::from(&borrowed);
+        if sock
+            .peer_addr()
+            .ok()
+            .and_then(|a: socket2::SockAddr| a.as_socket())
+            .is_some_and(|a| a == server_addr)
+        {
+            return sock.keepalive().unwrap_or(false).then(|| sock.keepalive_time().ok()).flatten();
+        }
+    }
+    None
+}