fix: handle libc errors

Author: 0x416e746f6e

Cargo.lock

@@ -59,7 +59,7 @@ sysctl = "0.7.1"
 thiserror = "2.0.16"
 time = { version = "0.3.43", features = ["formatting"] }
 tokio = { version = "1.47.1", features = ["macros", "rt", "rt-multi-thread", "signal"] }
-tokio-tungstenite = "0.27.0"
+tokio-tungstenite = {version = "0.27.0", features = ["rustls-tls-native-roots"] }
 tokio-util = "0.7.16"
 tracing = { version = "0.1.41", features = ["valuable"] }
 tracing-subscriber = { version = "0.3.20", features = ["env-filter", "json", "valuable"] }

crates/rproxy/Cargo.toml

@@ -84,6 +84,18 @@ pub(crate) struct ConfigAuthrpc {
     )]
     pub(crate) idle_connection_timeout: Duration,
 
+    /// interval between tcp keepalive packets on authrpc connections
+    #[arg(
+        default_value = "5s",
+        env = "RPROXY_AUTHRPC_KEEPALIVE_INTERVAL",
+        help_heading = "authrpc",
+        long("authrpc-keepalive-interval"),
+        name("authrpc_keepalive_interval"),
+        value_name = "duration",
+        value_parser = humantime::parse_duration
+    )]
+    pub(crate) keepalive_interval: Duration,
+
     /// host:port for authrpc proxy
     #[arg(
         default_value = "0.0.0.0:8651",
@@ -353,6 +365,11 @@ impl ConfigProxyHttp for ConfigAuthrpc {
         self.idle_connection_timeout
     }
 
+    #[inline]
+    fn keepalive_interval(&self) -> Duration {
+        self.keepalive_interval
+    }
+
     #[inline]
     fn listen_address(&self) -> SocketAddr {
         self.listen_address.parse::<SocketAddr>().expect(ALREADY_VALIDATED)

crates/rproxy/src/server/proxy/config/authrpc.rs

@@ -10,6 +10,19 @@ use crate::{config::ALREADY_VALIDATED, server::proxy::ws::config::ConfigProxyWs}
 
 #[derive(Args, Clone, Debug)]
 pub(crate) struct ConfigFlashblocks {
+    /// timeout to establish backend connections of to receive pong
+    /// websocket response
+    #[arg(
+        default_value = "30s",
+        env = "RPROXY_FLASHBLOCKS_BACKEND_TIMEOUT",
+        help_heading = "flashblocks",
+        long("flashblocks-backend-timeout"),
+        name("flashblocks_backend_timeout"),
+        value_name = "duration",
+        value_parser = humantime::parse_duration
+    )]
+    pub(crate) backend_timeout: Duration,
+
     /// url of flashblocks backend
     #[arg(
         default_value = "ws://127.0.0.1:11111",
@@ -29,19 +42,17 @@ pub(crate) struct ConfigFlashblocks {
         name("flashblocks_enabled")
     )]
     pub(crate) enabled: bool,
-
-    /// timeout to establish backend connections of to receive pong
-    /// websocket response
+    /// interval between tcp keepalive packets on flashblocks connections
     #[arg(
-        default_value = "30s",
-        env = "RPROXY_FLASHBLOCKS_BACKEND_TIMEOUT",
+        default_value = "5s",
+        env = "RPROXY_FLASHBLOCKS_KEEPALIVE_INTERVAL",
         help_heading = "flashblocks",
-        long("flashblocks-backend-timeout"),
-        name("flashblocks_backend_timeout"),
+        long("flashblocks-keepalive-interval"),
+        name("flashblocks_keepalive_interval"),
         value_name = "duration",
         value_parser = humantime::parse_duration
     )]
-    pub(crate) backend_timeout: Duration,
+    pub(crate) keepalive_interval: Duration,
 
     /// host:port for flashblocks proxy
     #[arg(
@@ -212,6 +223,11 @@ impl ConfigProxyWs for ConfigFlashblocks {
         self.backend_url.parse::<tungstenite::http::Uri>().expect(ALREADY_VALIDATED)
     }
 
+    #[inline]
+    fn keep_alive_interval(&self) -> Duration {
+        self.keepalive_interval
+    }
+
     #[inline]
     fn listen_address(&self) -> SocketAddr {
         self.listen_address.parse::<SocketAddr>().expect(ALREADY_VALIDATED)

crates/rproxy/src/server/proxy/config/flashblocks.rs

@@ -73,6 +73,18 @@ pub(crate) struct ConfigRpc {
     )]
     pub(crate) idle_connection_timeout: Duration,
 
+    /// interval between tcp keepalive packets on rpc connections
+    #[arg(
+        default_value = "5s",
+        env = "RPROXY_RPC_KEEPALIVE_INTERVAL",
+        help_heading = "rpc",
+        long("rpc-keepalive-interval"),
+        name("rpc_keepalive_interval"),
+        value_name = "duration",
+        value_parser = humantime::parse_duration
+    )]
+    pub(crate) keepalive_interval: Duration,
+
     /// host:port for rpc proxy
     #[arg(
         default_value = "0.0.0.0:8645",
@@ -349,6 +361,11 @@ impl ConfigProxyHttp for ConfigRpc {
         self.idle_connection_timeout
     }
 
+    #[inline]
+    fn keepalive_interval(&self) -> Duration {
+        self.keepalive_interval
+    }
+
     #[inline]
     fn listen_address(&self) -> SocketAddr {
         self.listen_address.parse::<SocketAddr>().expect(ALREADY_VALIDATED)

crates/rproxy/src/server/proxy/config/rpc.rs

@@ -1,6 +1,5 @@
 use std::{
     any::Any,
-    os::fd::{AsFd, AsRawFd},
     sync::{
         Arc,
         LazyLock,
@@ -14,7 +13,10 @@ use sysctl::Sysctl;
 use tracing::{debug, warn};
 use uuid::Uuid;
 
-use crate::server::metrics::{LabelsProxy, Metrics};
+use crate::{
+    server::metrics::{LabelsProxy, Metrics},
+    utils::setup_keepalive,
+};
 
 pub(crate) static TCP_KEEPALIVE_ATTEMPTS: LazyLock<libc::c_int> = LazyLock::new(|| {
     #[cfg(target_os = "linux")]
@@ -81,12 +83,8 @@ impl ConnectionGuard {
         proxy: &'static str,
         metrics: Arc<Metrics>,
         client_connections_count: Arc<AtomicI64>,
-        keep_alive_timeout: Duration,
+        keep_alive_interval: Duration,
     ) -> impl Fn(&dyn Any, &mut Extensions) {
-        let keep_alive_attempts: libc::c_int = *TCP_KEEPALIVE_ATTEMPTS;
-        let keep_alive_interval: libc::c_int =
-            keep_alive_timeout.div_f64(f64::from(keep_alive_attempts)).as_secs_f64().ceil() as i32;
-
         move |connection, extensions| {
             {
                 let val = client_connections_count.fetch_add(1, Ordering::Relaxed) + 1;
@@ -103,48 +101,21 @@ impl ConnectionGuard {
                 Some(stream)
             } else {
                 warn!(
+                    proxy = proxy,
                     connection_type = std::any::type_name_of_val(connection),
                     "Unexpected connection type",
                 );
                 None
             };
 
-            if let Some(stream) = stream {
-                #[cfg(target_os = "linux")]
-                unsafe {
-                    libc::setsockopt(
-                        stream.as_fd().as_raw_fd(),
-                        libc::IPPROTO_TCP,
-                        libc::TCP_KEEPIDLE,
-                        &keep_alive_interval as *const _ as *const libc::c_void,
-                        size_of_val(&keep_alive_interval) as libc::socklen_t,
-                    );
-                    libc::setsockopt(
-                        stream.as_fd().as_raw_fd(),
-                        libc::IPPROTO_TCP,
-                        libc::TCP_KEEPINTVL,
-                        &keep_alive_interval as *const _ as *const libc::c_void,
-                        size_of_val(&keep_alive_interval) as libc::socklen_t,
-                    );
-                    libc::setsockopt(
-                        stream.as_fd().as_raw_fd(),
-                        libc::IPPROTO_TCP,
-                        libc::TCP_KEEPCNT,
-                        &keep_alive_attempts as *const _ as *const libc::c_void,
-                        size_of_val(&keep_alive_attempts) as libc::socklen_t,
-                    );
-                }
-
-                #[cfg(target_os = "macos")]
-                unsafe {
-                    libc::setsockopt(
-                        stream.as_fd().as_raw_fd(),
-                        libc::IPPROTO_TCP,
-                        libc::TCP_KEEPALIVE,
-                        &keep_alive_interval as *const _ as *const _,
-                        std::mem::size_of_val(&keep_alive_interval) as libc::socklen_t,
-                    );
-                }
+            if let Some(stream) = stream &&
+                let Err(err) = setup_keepalive(stream, keep_alive_interval)
+            {
+                warn!(
+                    proxy = proxy,
+                    error = ?err,
+                    "Failed to set keepalive interval",
+                );
             }
 
             if let Some(stream) = stream {

crates/rproxy/src/server/proxy/connection_guard.rs

@@ -9,6 +9,7 @@ pub(crate) trait ConfigProxyHttp: Clone + Send + Unpin + 'static {
     fn backend_timeout(&self) -> Duration;
     fn backend_url(&self) -> Url;
     fn idle_connection_timeout(&self) -> Duration;
+    fn keepalive_interval(&self) -> Duration;
     fn listen_address(&self) -> SocketAddr;
     fn log_mirrored_requests(&self) -> bool;
     fn log_mirrored_responses(&self) -> bool;