From cfa53e14bd7582b060528fd5a8eb6873bed62815 Mon Sep 17 00:00:00 2001 From: rob thijssen Date: Wed, 2 Sep 2026 13:03:34 +0300 Subject: [PATCH] fix(data): a hidden 30s body timeout capped every proxied blob `object_store` puts a 30-second `reqwest` timeout on the whole request, body included. On a proxied blob that is not a timeout at all: the bucket is read at the pace the client drains it, so it is a maximum servable file size. The reason it never looked like one is that object_store hides its own consequences. It catches every response-body error and silently retries with a resumed Range, so a large read does not fail at 30s -- it reconnects every 30s and keeps going, until RetryConfig's 180s retry_timeout is spent. Then the body simply ends, with a 200 already on the wire and nothing logged anywhere. The ceiling is 180s x bandwidth: at the 8MB/s a GPU host gets to caveman, no file over ~1.4GB can ever be served, and no client retry converges because the next attempt is no faster. Measured against the live bucket, 500MiB at 1MB/s: cut at 184.1s, 193,340,351 of 524,288,000 bytes. With the timeout disabled, 499.6s and complete -- the same wall time curl takes reading the same object straight from MinIO. Disable it, and guard the stream the way a stream should be guarded: `storage.connect_timeout` and an idle `storage.read_timeout` that bounds silence rather than progress. A test asserts the total cap stays off. Issue #1 named `server.request_timeout` as the cause. It was not -- that key was parsed and then read by nothing, which is its own problem and is why the hunt started 300 seconds away from a 180-second bug. Remove it. Also log when a blob response ends before its Content-Length. Once the headers are out the bytes leave through hyper, so a cut mid-body was invisible on this side: the service logged a clean 200 while the client saw a short read and retried forever. Refs #1 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01PqNtYNhov3fukx46KS9R7L --- CLAUDE.md | 10 ++++ asset/config/config.toml.tmpl | 11 +++- crates/rustingface-api/src/handlers/blob.rs | 60 +++++++++++++++++-- crates/rustingface-api/src/telemetry.rs | 14 +++++ crates/rustingface-data/src/store.rs | 65 ++++++++++++++++++++- crates/rustingface-entities/src/config.rs | 29 +++++++-- crates/rustingface/src/cmd/doctor.rs | 10 ++++ doc/spec.md | 33 ++++++++++- test/conformance/conformance.py | 1 - 9 files changed, 217 insertions(+), 16 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index a04b941..15785da 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -25,6 +25,16 @@ The five guarantees in `doc/spec.md` §1 are contractual. In particular: ordering in `rustingface-core/src/fetch.rs` is deliberate: a crash may leak an orphan blob, which `gc` reclaims, but a manifest entry pointing at incomplete bytes is a silent correctness failure. +- **Never put a total-duration cap on a blob transfer.** A blob response lasts + `size / bandwidth`, so a ceiling on the whole request is a maximum servable + file size wearing a timeout's clothes, and no client retry can converge + against it. Every guard on this path is an *idle* timeout — + `server.client_stall_timeout`, `storage.read_timeout`, `upstream.read_timeout`. + `object_store` defaults to a 30s total request timeout *and* silently + retries the body errors it causes until its 180s `retry_timeout` runs out, + which caps a proxied blob at `180s x bandwidth` and logs nothing at all; + `rustingface-data`'s S3 client disables that timeout explicitly, and there is + a test asserting it stays disabled. - **Never complete a client response before the manifest write lands.** The tee runs one chunk behind for this reason. A file that was served but not recorded is invisible the moment the instance is sealed. diff --git a/asset/config/config.toml.tmpl b/asset/config/config.toml.tmpl index 62ec7c4..90baaaf 100644 --- a/asset/config/config.toml.tmpl +++ b/asset/config/config.toml.tmpl @@ -12,7 +12,10 @@ # different host from the service (bob). firewalld bounds who may reach it. listen = "{{LISTEN_ADDR}}" external_url = "https://rf.internal" -request_timeout = "300s" +# There is no total-duration cap on a request, deliberately: a blob response +# lasts size/bandwidth, so one would be a maximum file size in disguise. Every +# timeout here is an idle timeout instead. This one bounds how long a client may +# stop reading before the tee detaches it and finishes the transfer anyway. client_stall_timeout = "60s" [storage] @@ -26,6 +29,12 @@ secret_access_key_file = "/etc/rustingface/s3-secret-key" # which takes rustingface out of the data path but exposes the object store's # address to every client. blob_delivery = "proxy" +connect_timeout = "10s" +# Idle only: a proxied blob is read from the bucket at the pace the client +# drains it, so this bounds silence from the object store, not the length of the +# read. Anything measuring the whole request would cap the size of file this +# service can serve. +read_timeout = "60s" presign_ttl = "15m" multipart_part_size = "16MiB" diff --git a/crates/rustingface-api/src/handlers/blob.rs b/crates/rustingface-api/src/handlers/blob.rs index 32e5b15..d140e38 100644 --- a/crates/rustingface-api/src/handlers/blob.rs +++ b/crates/rustingface-api/src/handlers/blob.rs @@ -9,11 +9,14 @@ use std::time::Instant; use axum::body::Body as AxumBody; use axum::extract::State; use axum::response::{IntoResponse, Response}; -use futures::StreamExt; +use bytes::Bytes; +use futures::stream::BoxStream; +use futures::{Stream, StreamExt}; use http::{HeaderMap, HeaderValue, StatusCode, header}; use rustingface_core::{Body, ResolvedFile, Source}; use rustingface_entities::error::{Error, Result}; use rustingface_entities::repo::RepoType; +use tracing::warn; use crate::error::ApiError; use crate::path::{ResolvePath, parse_resolve}; @@ -126,9 +129,12 @@ pub async fn get( }; telemetry::served(source, len); telemetry::request("resolve_get", status.as_u16()); - let body = AxumBody::from_stream( - stream.map(|chunk| chunk.map_err(|err| std::io::Error::other(err.to_string()))), - ); + let body = AxumBody::from_stream(observed( + stream, + len, + source, + format!("{repo_id}/{filename}"), + )); Ok((status, headers, body).into_response()) } Body::None => Err(ApiError(Error::Internal( @@ -137,6 +143,52 @@ pub async fn get( } } +/// Wrap a body stream so a response that ends before its `Content-Length` says +/// so on this side of the wire. +/// +/// Once the headers are out the bytes leave through hyper, and a handler that +/// has already returned cannot report anything. Without this, a transfer cut +/// mid-body — a storage read that died, a client that vanished — is logged as a +/// successful 200 while the client sees a short read and retries forever. That +/// asymmetry is what makes such a failure read as "the client is flaky". +fn observed( + stream: BoxStream<'static, Result>, + expected: u64, + source: Source, + what: String, +) -> impl Stream> { + futures::stream::unfold( + (stream, 0u64, what), + move |(mut stream, sent, what)| async move { + match stream.next().await { + Some(Ok(chunk)) => { + let sent = sent + chunk.len() as u64; + Some((Ok(chunk), (stream, sent, what))) + } + Some(Err(err)) => { + warn!( + path = %what, sent, expected, %err, + "blob response failed mid-body; the client sees a short read, not an error" + ); + telemetry::truncated(source, sent); + let err = std::io::Error::other(err.to_string()); + Some((Err(err), (stream, sent, what))) + } + None => { + if sent < expected { + warn!( + path = %what, sent, expected, + "blob response ended early with no error; the client sees a short read" + ); + telemetry::truncated(source, sent); + } + None + } + } + }, + ) +} + /// Build the header set both `HEAD` and `GET` must carry. /// /// `Content-Encoding` is deliberately absent and must stay that way: the diff --git a/crates/rustingface-api/src/telemetry.rs b/crates/rustingface-api/src/telemetry.rs index 9cb3892..237f56e 100644 --- a/crates/rustingface-api/src/telemetry.rs +++ b/crates/rustingface-api/src/telemetry.rs @@ -34,6 +34,20 @@ pub fn served(source: Source, bytes: u64) { counter!("rustingface_bytes_served_total", "source" => label).increment(bytes); } +/// Record a blob response that ended before its `Content-Length`. +/// +/// The bytes leave through hyper rather than through a handler that can return +/// a status, so a cut mid-body is otherwise invisible on this side: the client +/// retries forever and the service logs a clean 200. +pub fn truncated(source: Source, bytes: u64) { + let label = match source { + Source::Bucket => "bucket", + Source::Upstream => "upstream", + }; + counter!("rustingface_blob_truncations_total", "source" => label).increment(1); + counter!("rustingface_bytes_truncated_total", "source" => label).increment(bytes); +} + /// Record the outcome of an upstream fetch. pub fn upstream_fetch(result: &'static str, seconds: f64) { counter!("rustingface_upstream_fetches_total", "result" => result).increment(1); diff --git a/crates/rustingface-data/src/store.rs b/crates/rustingface-data/src/store.rs index 8de90ee..eaf33ee 100644 --- a/crates/rustingface-data/src/store.rs +++ b/crates/rustingface-data/src/store.rs @@ -17,8 +17,8 @@ use object_store::local::LocalFileSystem; use object_store::path::Path; use object_store::signer::Signer; use object_store::{ - GetOptions, GetRange, ObjectStore, ObjectStoreExt, PutMode, PutOptions, PutPayload, - UpdateVersion, WriteMultipart, + ClientOptions, GetOptions, GetRange, ObjectStore, ObjectStoreExt, PutMode, PutOptions, + PutPayload, UpdateVersion, WriteMultipart, }; use rustingface_core::ports::{BlobRead, BlobWriter, Store, StoredObject, Version}; use rustingface_entities::config::Storage as StorageConfig; @@ -63,7 +63,8 @@ impl ObjectStoreAdapter { let mut builder = AmazonS3Builder::new() .with_bucket_name(&cfg.bucket) .with_region(&cfg.region) - .with_virtual_hosted_style_request(!cfg.path_style); + .with_virtual_hosted_style_request(!cfg.path_style) + .with_client_options(Self::client_options(cfg)); if let Some(endpoint) = &cfg.endpoint { // MinIO on the mesh is reached over plain HTTP; the mesh itself is @@ -93,6 +94,36 @@ impl ObjectStoreAdapter { }) } + /// HTTP client settings for the S3 backend. + /// + /// `object_store` defaults to a 30-second timeout on the *whole* request, + /// body included, and that default must stay disabled here. A proxied blob + /// is read from the bucket at the pace the client drains it, so a cap on + /// the whole request is not a timeout at all -- it is a maximum servable + /// file size. + /// + /// The way it fails is worth knowing, because none of it is visible from + /// outside. The 30s deadline kills the response body mid-stream; + /// `object_store` catches every body error and silently retries, resuming + /// with a `Range` from where it stopped (`client/get.rs`, "Retry all + /// response body errors"). So a large read does not fail at 30s -- it + /// quietly reconnects every 30s and keeps going, until `RetryConfig`'s + /// 180s `retry_timeout` is exhausted. Then the body simply ends, with a + /// 200 already on the wire and nothing logged. The ceiling is therefore + /// `180s x bandwidth`: at 8MB/s, no file over ~1.4GB can ever be served, + /// and no client retry can converge because the next attempt is no faster. + /// + /// The guard that belongs on a stream is an idle one, which bounds silence + /// rather than progress and resets on every chunk. `read_timeout` covers + /// the wait for response headers too, so a hung backend still fails a + /// small metadata request promptly. + fn client_options(cfg: &StorageConfig) -> ClientOptions { + ClientOptions::new() + .with_timeout_disabled() + .with_connect_timeout(cfg.connect_timeout) + .with_read_timeout(cfg.read_timeout) + } + /// A filesystem-backed adapter, for tests. pub fn local(path: &FsPath) -> Result { std::fs::create_dir_all(path) @@ -363,3 +394,31 @@ impl BlobWriter for MultipartBlobWriter { .map_err(|e| storage(&format!("aborting upload of {key}"), e)) } } + +#[cfg(test)] +mod tests { + use super::*; + use object_store::ClientConfigKey; + use object_store::aws::AmazonS3ConfigKey; + + /// A total-duration timeout on an S3 request is a maximum servable file + /// size, because a proxied blob is read at the pace the client drains it. + /// `object_store` ships one on by default (30s); it must be off here, and + /// the guards that remain must be idle ones. + #[test] + fn the_s3_client_has_no_total_request_timeout() { + let cfg = StorageConfig { + endpoint: Some("http://localhost:9000".into()), + connect_timeout: Duration::from_secs(10), + read_timeout: Duration::from_secs(60), + ..Default::default() + }; + let builder = + AmazonS3Builder::new().with_client_options(ObjectStoreAdapter::client_options(&cfg)); + let get = |key| builder.get_config_value(&AmazonS3ConfigKey::Client(key)); + + assert_eq!(get(ClientConfigKey::Timeout), None); + assert_eq!(get(ClientConfigKey::ConnectTimeout).as_deref(), Some("10s")); + assert_eq!(get(ClientConfigKey::ReadTimeout).as_deref(), Some("1m")); + } +} diff --git a/crates/rustingface-entities/src/config.rs b/crates/rustingface-entities/src/config.rs index f350080..487be02 100644 --- a/crates/rustingface-entities/src/config.rs +++ b/crates/rustingface-entities/src/config.rs @@ -32,6 +32,13 @@ pub struct Config { } /// HTTP listener settings. +/// +/// There is deliberately no total-duration cap on a request. A blob response +/// lasts `size / bandwidth`, so a ceiling on the whole request is a ceiling on +/// file size in disguise: the same client asking for the same 3GB shard +/// succeeds on a fast link and fails on every retry on a slow one. Every guard +/// here and in [`Storage`] and [`Upstream`] is therefore an *idle* timeout, +/// which bounds silence rather than progress. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(default, deny_unknown_fields)] pub struct Server { @@ -40,9 +47,6 @@ pub struct Server { pub listen: String, /// How this service is reached from outside, used to build absolute URLs. pub external_url: Option, - /// Ceiling on a single request, including a cold multi-gigabyte fetch. - #[serde(with = "humantime_serde")] - pub request_timeout: Duration, /// How long a stalled client may hold up the tee before the transfer is /// detached and finished without it. #[serde(with = "humantime_serde")] @@ -54,7 +58,6 @@ impl Default for Server { Self { listen: "127.0.0.1:20482".into(), external_url: None, - request_timeout: Duration::from_secs(300), client_stall_timeout: Duration::from_secs(60), } } @@ -84,6 +87,15 @@ pub struct Storage { pub access_key_id_file: Option, pub secret_access_key_file: Option, pub blob_delivery: BlobDelivery, + /// How long to wait for a TCP connection to the object store. + #[serde(with = "humantime_serde")] + pub connect_timeout: Duration, + /// How long the object store may go silent mid-response before the read is + /// abandoned. Idle only, and it resets on every chunk: a proxied blob is + /// read at the client's pace, so anything measuring the whole request + /// instead would cap the size of file this service can serve. + #[serde(with = "humantime_serde")] + pub read_timeout: Duration, #[serde(with = "humantime_serde")] pub presign_ttl: Duration, #[serde(deserialize_with = "de_byte_size")] @@ -104,6 +116,8 @@ impl Default for Storage { access_key_id_file: None, secret_access_key_file: None, blob_delivery: BlobDelivery::Proxy, + connect_timeout: Duration::from_secs(10), + read_timeout: Duration::from_secs(60), presign_ttl: Duration::from_secs(15 * 60), multipart_part_size: 16 * 1024 * 1024, manifest_write_retries: 5, @@ -353,7 +367,7 @@ mod tests { [server] listen = "127.0.0.1:20482" external_url = "https://rf.internal" - request_timeout = "300s" + client_stall_timeout = "60s" [storage] endpoint = "http://caveman.kosherinata.internal:9000" @@ -361,6 +375,8 @@ mod tests { region = "us-east-1" path_style = true blob_delivery = "proxy" + connect_timeout = "10s" + read_timeout = "60s" presign_ttl = "15m" multipart_part_size = "16MiB" @@ -387,7 +403,8 @@ mod tests { cfg.validate().unwrap(); assert_eq!(cfg.storage.multipart_part_size, 16 * 1024 * 1024); assert_eq!(cfg.policy.ref_resolution, RefResolution::Freeze); - assert_eq!(cfg.server.request_timeout, Duration::from_secs(300)); + assert_eq!(cfg.server.client_stall_timeout, Duration::from_secs(60)); + assert_eq!(cfg.storage.read_timeout, Duration::from_secs(60)); assert!(!cfg.sealed()); } diff --git a/crates/rustingface/src/cmd/doctor.rs b/crates/rustingface/src/cmd/doctor.rs index 0bd68dc..1dc4403 100644 --- a/crates/rustingface/src/cmd/doctor.rs +++ b/crates/rustingface/src/cmd/doctor.rs @@ -51,6 +51,16 @@ pub async fn run(config: rustingface_entities::config::Config) -> Result<()> { } ); println!(" blob delivery {:?}", config.storage.blob_delivery); + // Named here because the failure they prevent looks like a client problem: + // a total-duration cap on a blob read is a maximum servable file size, and + // an operator chasing one needs to see that these are idle guards. + println!( + " idle guards storage {:?}, upstream {:?}, client {:?} \ + (no total-duration cap)", + config.storage.read_timeout, + config.upstream.read_timeout, + config.server.client_stall_timeout, + ); println!( " part size {}", human_bytes(config.storage.multipart_part_size) diff --git a/doc/spec.md b/doc/spec.md index d727c5f..04aa970 100644 --- a/doc/spec.md +++ b/doc/spec.md @@ -225,7 +225,7 @@ config. [server] listen = "127.0.0.1:8080" external_url = "https://models.internal.example" # used to build redirect URLs -request_timeout = "300s" +client_stall_timeout = "60s" # idle guard on the tee; see below [storage] endpoint = "https://s3.internal.example:9000" @@ -235,6 +235,8 @@ path_style = true access_key_id_file = "/etc/rustingface/s3-access-key" secret_access_key_file = "/etc/rustingface/s3-secret-key" blob_delivery = "proxy" # proxy | redirect +connect_timeout = "10s" +read_timeout = "60s" # idle guard; see below presign_ttl = "15m" # redirect mode only multipart_part_size = "16MiB" @@ -260,6 +262,35 @@ metrics = true # /metrics, Prometheus text format log_format = "json" # json | text ``` +### Timeouts + +Every timeout in this configuration is an **idle** timeout: it bounds silence, +never progress. There is deliberately no ceiling on how long a request may +take. + +A blob response lasts `size / bandwidth`. A total-duration cap on it is +therefore not a timeout but a maximum servable file size, and one expressed in +the wrong units: the same client asking for the same file succeeds over a fast +link and fails on every retry over a slow one, with no diagnostic difference +between the two. A retry loop cannot converge against it, because nothing about +the next attempt is any faster. + +- `server.client_stall_timeout` — how long a client may stop reading before the + tee detaches it and finishes the transfer without it. +- `storage.read_timeout` — how long the object store may go silent mid-response. + It resets on every chunk, and also bounds the wait for response headers, so a + hung backend still fails a small metadata request promptly. +- `upstream.read_timeout` — the same guard on the Hub side. + +This applies to the object-storage client in particular. `object_store` +defaults to a 30-second cap on the whole request, body included, and then hides +its own consequences: it catches the resulting body errors and silently retries +with a resumed range, so a large read reconnects every 30 seconds until the +180-second `retry_timeout` is exhausted, at which point the body ends with a 200 +already on the wire and nothing logged. The adapter must disable that cap +explicitly, or every proxied blob over `180s x bandwidth` fails deterministically +and forever -- at 8MB/s, anything over about 1.4GB. + ### `policy.ref_resolution` - `freeze` (default) — the first resolution of a mutable ref is recorded and diff --git a/test/conformance/conformance.py b/test/conformance/conformance.py index 0c53ba6..7eb62df 100755 --- a/test/conformance/conformance.py +++ b/test/conformance/conformance.py @@ -134,7 +134,6 @@ class Instance: f""" [server] listen = "127.0.0.1:{self.port}" -request_timeout = "600s" [storage] {storage_lines}