diff --git a/objectstore-service/src/backend/common.rs b/objectstore-service/src/backend/common.rs index 560c095a0..6340c5679 100644 --- a/objectstore-service/src/backend/common.rs +++ b/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::*; diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index acb549cb9..ef4c83d65 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/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(), diff --git a/objectstore-service/src/backend/s3_compatible.rs b/objectstore-service/src/backend/s3_compatible.rs index bdff4c358..154cc81a0 100644 --- a/objectstore-service/src/backend/s3_compatible.rs +++ b/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 S3CompatibleBackend { 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(), diff --git a/objectstore-service/src/backend/tiered.rs b/objectstore-service/src/backend/tiered.rs index 97bee0636..1a9fe3c14 100644 --- a/objectstore-service/src/backend/tiered.rs +++ b/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, diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index adb029639..9160e5ef9 100644 --- a/objectstore-service/src/service.rs +++ b/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, };