From cfa53e14bd7582b060528fd5a8eb6873bed62815 Mon Sep 17 00:00:00 2001 From: rob thijssen Date: Wed, 2 Sep 2026 13:03:34 +0300 Subject: [PATCH 1/2] 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} From 5f91032e6fe92da00d5e41289c575181e47387ba Mon Sep 17 00:00:00 2001 From: rob thijssen Date: Wed, 2 Sep 2026 13:03:49 +0300 Subject: [PATCH 2/2] fix(core): a ranged fetch of a cold file left the client in silence A ranged GET on a file the bucket does not hold used to fetch and store the whole thing before sending a single byte. Asking for the first megabyte of a 3.3GB shard therefore held the connection open, silent, for the length of the entire transfer: 93 seconds measured against the live service at 28MB/s, and 400-800s at the 8MB/s a GPU host gets. No client waits that long for a first byte. What makes it unrecoverable rather than merely slow is that `huggingface_hub` resumes with a Range header. So the first failure leaves an `.incomplete` file, and *every retry from then on* takes this path and is answered with silence. The partial file never grows, its mtime goes stale, and the retry loop cannot converge -- while the server logs nothing but `stored blob and recorded manifest entry`, because from its side the fetch succeeded. The whole file must still be fetched and stored, since a fragment must never reach the bucket. What must not happen is the client waiting on that in silence. So narrow the tee instead of blocking on it: the client is sent its slice as those bytes pass, one chunk behind as before, with the last piece still withheld until the manifest write lands. For the open-ended `bytes=N-` a resume actually asks for, that is a continuous stream to the end of the file. `blocking` and `LeadResult::Completed` go with it; nothing else used them. The test asserting a ranged cold request stores the whole file and never a fragment is unchanged and still passes. Refs #1 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01PqNtYNhov3fukx46KS9R7L --- crates/rustingface-api/tests/sovereignty.rs | 69 +++++++++++++++++++++ crates/rustingface-core/src/fetch.rs | 58 +++++++++++++---- crates/rustingface-core/src/registry.rs | 38 ++++++------ doc/spec.md | 7 +++ 4 files changed, 140 insertions(+), 32 deletions(-) diff --git a/crates/rustingface-api/tests/sovereignty.rs b/crates/rustingface-api/tests/sovereignty.rs index cf355cd..2c0752f 100644 --- a/crates/rustingface-api/tests/sovereignty.rs +++ b/crates/rustingface-api/tests/sovereignty.rs @@ -364,6 +364,75 @@ async fn a_client_that_disconnects_mid_transfer_still_leaves_the_file_stored() { assert_eq!(served, weights); } +#[tokio::test] +async fn a_cold_ranged_request_streams_while_the_file_is_still_being_fetched() { + // The failure this pins down: a ranged GET on a file the bucket does not + // hold used to fetch and store the whole thing before sending a single + // byte. A client asking for a 3GB shard therefore saw an open connection + // and total silence for the length of the transfer, which its read timeout + // ends long before -- and because `hf download` resumes with a Range + // header, *every retry after the first* took that path. The retry loop + // could never converge, and the server logged nothing but success. + // + // The whole file must still be stored (never a fragment); what must not + // happen is the client waiting on that in silence. + let hub = FakeHub::start().await; + let weights = publish_model(&hub).await; + hub.throttle(8 * 1024, Duration::from_millis(5)).await; + + let bucket = bucket_dir("cold-range-streams"); + let rf = instance(&bucket, Some(&hub), |_| {}); + + // Resume from an offset, open-ended, which is the shape huggingface_hub + // actually sends. + let response = rf + .app + .clone() + .oneshot( + Request::builder() + .uri("/Qwen/Qwen3-32B/resolve/main/model.safetensors") + .header("range", "bytes=1000-") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::PARTIAL_CONTENT); + assert_eq!( + response.headers()["content-range"], + format!("bytes 1000-{}/{}", weights.len() - 1, weights.len()) + ); + + use futures::StreamExt; + let mut body = response.into_body().into_data_stream(); + let first = body + .next() + .await + .expect("a chunk before the fetch ends") + .unwrap(); + assert!( + first.len() < weights.len() - 1000, + "the first chunk arrived while the transfer was still running, not after it" + ); + + let mut served = first.to_vec(); + while let Some(chunk) = body.next().await { + served.extend_from_slice(&chunk.unwrap()); + } + assert_eq!(served, weights[1000..], "and the range itself is intact"); + + // The bucket still holds the whole file, not the slice that was served. + let oid = { + use sha2::{Digest, Sha256}; + hex::encode(Sha256::digest(&weights)) + }; + assert_eq!( + std::fs::read(bucket.join(key::blob(&oid))).unwrap(), + weights, + "a ranged cold request stores the whole file, never a fragment" + ); +} + #[tokio::test] async fn a_ranged_request_resumes_from_the_offset_it_asks_for() { let hub = FakeHub::start().await; diff --git a/crates/rustingface-core/src/fetch.rs b/crates/rustingface-core/src/fetch.rs index b3a7086..556b28b 100644 --- a/crates/rustingface-core/src/fetch.rs +++ b/crates/rustingface-core/src/fetch.rs @@ -6,6 +6,7 @@ //! Ctrl-C at 30GB does not discard 30GB: the retention guarantee is better //! served by finishing (spec §7). +use std::ops::Range; use std::sync::Arc; use bytes::Bytes; @@ -36,16 +37,19 @@ enum Sink { impl Registry { /// Lead an upstream fetch for one file. /// - /// `blocking` forces the transfer to run to completion before returning, - /// which is what a ranged request on a cold file needs: a partial blob - /// must never reach the bucket. + /// `client_range` narrows what the *client* is sent; the whole file is + /// fetched and stored either way, because a partial blob must never reach + /// the bucket. Narrowing the tee rather than waiting for the store is what + /// keeps a ranged request on a cold file from sitting in silence for the + /// length of a multi-gigabyte transfer — which is what a resumed + /// `hf download` asks for on every retry after its first failure. pub(crate) async fn lead_fetch( &self, repo_type: RepoType, repo_id: &RepoId, commit: &str, path: &str, - blocking: bool, + client_range: Option>, ) -> Result { let body = self .upstream() @@ -82,16 +86,9 @@ impl Registry { sink, part_size: self.config().storage.multipart_part_size, stall_timeout: self.config().server.client_stall_timeout, + client_range: client_range.clone(), }; - if blocking { - // Nobody is reading the tee, so drop the client half up front and - // let the transfer run at full speed. - drop(client_rx); - let entry = transfer.run(self, body.stream, client_tx).await?; - return Ok(LeadResult::Completed(entry)); - } - let registry = self.clone(); tokio::spawn(async move { let result = transfer.run(®istry, body.stream, client_tx).await; @@ -106,6 +103,7 @@ impl Registry { head, rx: client_rx, task: FetchTask::new(outcome_rx), + range: client_range, }) } @@ -137,6 +135,8 @@ struct Transfer { sink: Sink, part_size: u64, stall_timeout: std::time::Duration, + /// Which bytes of the file the client asked for. `None` is the whole file. + client_range: Option>, } impl Transfer { @@ -166,19 +166,33 @@ impl Transfer { } }; + // Where the next chunk starts in the file, so a narrowed tee can work + // out which slice of it the client asked for. + let mut offset: u64 = 0; + let transfer = async { while let Some(chunk) = stream.next().await { let chunk = chunk?; hasher.update(&chunk); + let start = offset; + offset += chunk.len() as u64; + match &mut writer { Some(writer) => writer.write(chunk.clone()).await?, None => buffered.push(chunk.clone()), } + // The client's share of this chunk. The whole chunk when it + // asked for the whole file; the overlap with its range + // otherwise; nothing at all for a chunk outside it. + let Some(share) = self.client_share(&chunk, start, offset) else { + continue; + }; + // Tee to the client one chunk behind, dropping the client -- // not the transfer -- if it has gone away or stopped reading. - if let Some(previous) = holdback.replace(chunk) + if let Some(previous) = holdback.replace(share) && let Some(tx) = &client { match tokio::time::timeout(self.stall_timeout, tx.send(Ok(previous))).await { @@ -316,6 +330,24 @@ impl Transfer { } } +impl Transfer { + /// The part of `chunk` — which spans `[start, end)` of the file — that the + /// client asked for, or `None` when none of it is. + /// + /// Slicing is cheap: `Bytes` shares the buffer rather than copying it. + fn client_share(&self, chunk: &Bytes, start: u64, end: u64) -> Option { + let Some(range) = &self.client_range else { + return Some(chunk.clone()); + }; + let from = start.max(range.start); + let to = end.min(range.end); + if from >= to { + return None; + } + Some(chunk.slice((from - start) as usize..(to - start) as usize)) + } +} + /// Build the manifest entry for a completed transfer, carrying upstream's /// values through unchanged. fn entry_from(path: &str, size: u64, hex: &str, head: &UpstreamHead) -> Entry { diff --git a/crates/rustingface-core/src/registry.rs b/crates/rustingface-core/src/registry.rs index fa3f1c6..2b38d7d 100644 --- a/crates/rustingface-core/src/registry.rs +++ b/crates/rustingface-core/src/registry.rs @@ -734,7 +734,7 @@ impl Registry { } Role::Leader(lead) => { let outcome = self - .lead_fetch(repo_type, repo_id, &commit, path, range.is_some()) + .lead_fetch(repo_type, repo_id, &commit, path, range.clone()) .await; match outcome { Err(err) => { @@ -753,12 +753,25 @@ impl Registry { })?; Ok((file, Source::Bucket)) } - Ok(LeadResult::Streaming { head, rx, task }) => { + Ok(LeadResult::Streaming { + head, + rx, + task, + range, + }) => { // The upload task owns the flight from here: it // publishes the outcome whether or not this client // stays connected. task.attach(lead); let size = head.linked_size.or(head.content_length).unwrap_or(0); + // A ranged request is answered from the same tee, + // narrowed. The whole file is still fetched and stored + // -- a fragment must never reach the bucket -- but the + // client is not made to wait for that in silence. + let len = match &range { + Some(range) => range.end - range.start, + None => size, + }; Ok(( ResolvedFile { commit, @@ -768,25 +781,13 @@ impl Registry { linked_size: head.linked_size, body: Body::Stream { stream: Box::pin(tokio_stream_of(rx)), - len: size, - range: None, + len, + range, }, }, Source::Upstream, )) } - Ok(LeadResult::Completed(entry)) => { - lead.finish(Outcome::Stored(entry.clone())); - let file = self - .serve_stored(&entry, &commit, range) - .await? - .ok_or_else(|| { - Error::Storage(format!( - "the blob for {repo_id}@{commit}/{path} vanished after being stored" - )) - })?; - Ok((file, Source::Upstream)) - } } } } @@ -898,11 +899,10 @@ pub(crate) enum LeadResult { Streaming { head: UpstreamHead, rx: mpsc::Receiver>, + /// The slice of the file the tee is narrowed to, if any. + range: Option>, task: FetchTask, }, - /// The transfer ran to completion before the response was built, because - /// the client asked for a range and a partial blob may not be stored. - Completed(Entry), } /// Handle to a detached transfer, which owns the flight once attached. diff --git a/doc/spec.md b/doc/spec.md index 04aa970..dd6b77f 100644 --- a/doc/spec.md +++ b/doc/spec.md @@ -291,6 +291,13 @@ 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. +A ranged request on a file the bucket does not hold is subject to the same +principle from the other side. The whole file is still fetched and stored, since +a fragment must never be stored, but the client is sent its slice as those bytes +pass rather than being made to wait in silence for the transfer to end. A +resumed `hf download` asks for exactly that shape on every retry, and silence +there is indistinguishable to it from a dead server. + ### `policy.ref_resolution` - `freeze` (default) — the first resolution of a mutable ref is recorded and