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
 starknet-types-core.workspace = true
 strum = { workspace = true, features = ["derive"] }

crates/apollo_infra/Cargo.toml

@@ -17,6 +17,7 @@ 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};
@@ -39,6 +40,11 @@ pub const DEFAULT_RETRIES: usize = 15;
 pub const REQUEST_TIMEOUT_ERROR_MESSAGE: &str = "request timed out";
 
 const DEFAULT_IDLE_CONNECTIONS: usize = 10;
+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 +60,8 @@ 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.
     pub idle_timeout_ms: u64,
     pub attempts_per_log: usize,
     pub initial_retry_delay_ms: u64,
@@ -184,13 +192,20 @@ 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

@@ -2,6 +2,7 @@ use std::convert::Infallible;
 use std::fmt::Debug;
 use std::future::ready;
 use std::net::{Ipv4Addr, SocketAddr};
+use std::os::unix::io::BorrowedFd;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -23,6 +24,7 @@ use metrics_exporter_prometheus::PrometheusBuilder;
 use rstest::rstest;
 use serde::de::DeserializeOwned;
 use serde::Serialize;
+use socket2::SockRef;
 use starknet_types_core::felt::Felt;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::{TcpListener, TcpStream};
@@ -734,3 +736,54 @@ async fn zombie_connection_is_evicted() {
          connection is still open"
     );
 }
+
+/// Returns whether the outbound socket in this process that is connected to `server_addr`
+/// has `SO_KEEPALIVE` enabled. Returns `false` if no such socket is found.
+fn client_socket_has_keepalive(server_addr: SocketAddr) -> bool {
+    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);
+        }
+    }
+    false
+}
+
+/// Verifies that `SO_KEEPALIVE` on the client's outbound socket is set.
+#[rstest]
+#[tokio::test]
+async fn tcp_keepalive_socket_is_set() {
+    const ARBITRARY_IDLE_TIMEOUT_MS: u64 = 100;
+
+    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();
+
+    // Hyper connects lazily on first request, so after setup no socket is connected to
+    // a_socket yet — our test client's connection will be the only one to check.
+    setup_for_tests(VALID_VALUE_A, a_socket, b_socket, MAX_CONCURRENCY, None).await;
+
+    let client = ComponentAClient::new(
+        RemoteClientConfig { idle_timeout_ms: ARBITRARY_IDLE_TIMEOUT_MS, ..Default::default() },
+        &a_socket.ip().to_string(),
+        a_socket.port(),
+        &TEST_REMOTE_CLIENT_METRICS,
+    );
+
+    // Establish the connection and leave it in Hyper's pool.
+    client.a_get_value().await.expect("request should succeed");
+
+    assert!(
+        client_socket_has_keepalive(a_socket),
+        "SO_KEEPALIVE on the outbound socket should be set."
+    );
+}