feat(storage): Make TCP user timeout configurable

objectstore-service/src/backend/common.rs

@@ -604,21 +604,33 @@ pub enum TieredUpdate {
 ///
 /// Automatic decompression is disabled because backends store pre-compressed
 /// payloads and manage `Content-Encoding` themselves.
-pub(super) fn reqwest_client() -> reqwest::Client {
-    reqwest::Client::builder()
+pub(super) fn reqwest_client(tcp_user_timeout: Duration) -> reqwest::Client {
+    let builder = reqwest::Client::builder()
         .user_agent(USER_AGENT)
         .hickory_dns(true)
         .http1_only()
         .no_zstd()
         .no_brotli()
         .no_gzip()
-        .no_deflate()
+        .no_deflate();
+
+    #[cfg(target_os = "linux")]
+    let builder = builder.tcp_user_timeout(tcp_user_timeout);
+    #[cfg(not(target_os = "linux"))]
+    let _ = tcp_user_timeout;
+
+    builder
         .build()
         // INVARIANT: Building fails only if the TLS backend cannot be initialized, which
         // is checked at startup when the rustls crypto provider is installed.
         .expect("failed to build backend HTTP client")
 }
 
+/// Matches reqwest's default TCP user timeout on supported platforms.
+pub(super) fn default_tcp_user_timeout() -> Duration {
+    Duration::from_secs(30)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

objectstore-service/src/backend/gcs.rs

@@ -5,7 +5,7 @@ use std::collections::BTreeMap;
 use std::future::Future;
 use std::num::NonZeroU64;
 use std::sync::Arc;
-use std::time::SystemTime;
+use std::time::{Duration, SystemTime};
 use std::{fmt, io};
 
 use futures_util::{StreamExt, TryStreamExt};
@@ -80,6 +80,22 @@ pub struct GcsConfig {
     /// - `OS__STORAGE__BUCKET=my-gcs-bucket`
     pub bucket: String,
 
+    /// Maximum time TCP data may remain unacknowledged or blocked before failing the request.
+    ///
+    /// This detects stalled transports. It does not limit the total upload duration or the time
+    /// waiting for a response after the upload is acknowledged. Applied on Linux and ignored on
+    /// other platforms. `0s` uses the operating system's default.
+    ///
+    /// # Default
+    ///
+    /// `30s`
+    ///
+    /// # Environment Variables
+    ///
+    /// - `OS__STORAGE__TCP_USER_TIMEOUT=30s`
+    #[serde(default = "common::default_tcp_user_timeout", with = "humantime_serde")]
+    pub tcp_user_timeout: Duration,
+
     /// Reports what this backend stores, for per-usecase cost attribution.
     ///
     /// # Default
@@ -515,6 +531,7 @@ impl GcsBackend {
         let GcsConfig {
             endpoint,
             bucket,
+            tcp_user_timeout,
             cogs,
         } = config;
         let change_stream = streams.build(cogs.as_ref());
@@ -528,7 +545,7 @@ impl GcsBackend {
         let endpoint_str = endpoint.as_deref().unwrap_or(DEFAULT_ENDPOINT);
 
         Ok(Self {
-            client: common::reqwest_client(),
+            client: common::reqwest_client(tcp_user_timeout),
             endpoint: endpoint_str
                 .parse()
                 .map_err(|e| anyhow::Error::new(e).context("invalid GCS endpoint URL"))?,
@@ -1848,6 +1865,7 @@ mod tests {
         GcsConfig {
             endpoint: Some("http://localhost:8087".into()),
             bucket: "test-bucket".into(),
+            tcp_user_timeout: Duration::from_secs(30),
             cogs: None,
         }
     }
@@ -1932,6 +1950,7 @@ mod tests {
             GcsConfig {
                 endpoint: Some(endpoint),
                 bucket: "bucket".into(),
+                tcp_user_timeout: Duration::from_secs(30),
                 cogs: None,
             },
             &ChangeStreamFactory::default(),

objectstore-service/src/backend/s3_compatible.rs

@@ -4,6 +4,7 @@ use std::convert::Infallible;
 use std::error::Error as StdError;
 use std::sync::Arc;
 use std::sync::atomic::Ordering;
+use std::time::Duration;
 use std::{fmt, io};
 
 use futures_util::{StreamExt, TryStreamExt};
@@ -62,6 +63,22 @@ pub struct S3CompatibleConfig {
     /// - `OS__STORAGE__BUCKET=my-bucket`
     pub bucket: String,
 
+    /// Maximum time TCP data may remain unacknowledged or blocked before failing the request.
+    ///
+    /// This detects stalled transports. It does not limit the total upload duration or the time
+    /// waiting for a response after the upload is acknowledged. Applied on Linux and ignored on
+    /// other platforms. `0s` uses the operating system's default.
+    ///
+    /// # Default
+    ///
+    /// `30s`
+    ///
+    /// # Environment Variables
+    ///
+    /// - `OS__STORAGE__TCP_USER_TIMEOUT=30s`
+    #[serde(default = "common::default_tcp_user_timeout", with = "humantime_serde")]
+    pub tcp_user_timeout: Duration,
+
     /// Reports what this backend stores, for per-usecase cost attribution.
     ///
     /// # Default
@@ -151,10 +168,11 @@ impl<T> S3CompatibleBackend<T> {
         let S3CompatibleConfig {
             endpoint,
             bucket,
+            tcp_user_timeout,
             cogs,
         } = config;
         Self {
-            client: common::reqwest_client(),
+            client: common::reqwest_client(tcp_user_timeout),
             endpoint,
             bucket,
             token_provider,
@@ -560,6 +578,7 @@ mod tests {
             S3CompatibleConfig {
                 endpoint: "http://localhost:8089".into(),
                 bucket: "test-bucket".into(),
+                tcp_user_timeout: Duration::from_secs(30),
                 cogs: None,
             },
             &ChangeStreamFactory::default(),
@@ -637,6 +656,7 @@ mod tests {
                 S3CompatibleConfig {
                     endpoint,
                     bucket: "bucket".into(),
+                    tcp_user_timeout: Duration::from_secs(30),
                     cogs: None,
                 },
                 &ChangeStreamFactory::default(),

objectstore-service/src/backend/tiered.rs

@@ -1401,6 +1401,7 @@ mod tests {
             GcsConfig {
                 endpoint: Some("http://localhost:8087".into()),
                 bucket: "test-bucket".into(),
+                tcp_user_timeout: Duration::from_secs(30),
                 cogs: None,
             },
             &streams,

objectstore-service/src/service.rs

@@ -701,6 +701,7 @@ mod tests {
         let config = GcsConfig {
             endpoint: Some("http://localhost:8087".into()),
             bucket: "test-bucket".into(), // aligned with the env var in devservices and CI
+            tcp_user_timeout: Duration::from_secs(30),
             cogs: None,
         };
 
@@ -744,6 +745,7 @@ mod tests {
         let gcs_config = GcsConfig {
             endpoint: Some("http://localhost:8087".into()),
             bucket: "test-bucket".into(),
+            tcp_user_timeout: Duration::from_secs(30),
             cogs: None,
         };