grenade/rustingface
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(data): a hidden 30s body timeout capped every proxied blob

Browse Source 
`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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PqNtYNhov3fukx46KS9R7L
This commit is contained in:
rob thijssen
2026-09-02 13:03:34 +03:00
parent 64c1e0e3ad
commit cfa53e14bd
9 changed files with 217 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
CLAUDE.md
View File

@@ -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.

11
asset/config/config.toml.tmpl
View File

@@ -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"

60
crates/rustingface-api/src/handlers/blob.rs
View File

@@ -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<Bytes>>,
expected: u64,
source: Source,
what: String,
) -> impl Stream<Item = std::result::Result<Bytes, std::io::Error>> {
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

14
crates/rustingface-api/src/telemetry.rs
View File

@@ -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);

65
crates/rustingface-data/src/store.rs
View File

@@ -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<Self> {
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"));
}
}

29
crates/rustingface-entities/src/config.rs
View File

@@ -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<String>,
/// 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<PathBuf>,
pub secret_access_key_file: Option<PathBuf>,
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());
}

10
crates/rustingface/src/cmd/doctor.rs
View File

@@ -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)

33
doc/spec.md
View File

@@ -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

1
test/conformance/conformance.py
View File

@@ -134,7 +134,6 @@ class Instance:
f"""
[server]
listen = "127.0.0.1:{self.port}"
request_timeout = "600s"
[storage]
{storage_lines}