Merge branch 'fix/followers-stream-from-the-transfer'

A request that joined another request's in-flight fetch waited with nothing
on the wire until the whole transfer finished. The leader now spools each
chunk as it passes and followers stream from it, which is the broadcast
spec §7 asked for.

Closes #2

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PqNtYNhov3fukx46KS9R7L

.gitea/workflows/deploy.yml

@@ -360,14 +360,18 @@ jobs:
             fi'
           echo "frontend and API both answering through rf.internal"
 
-      # No sudo: nginx's logs are world-readable and /var/log/nginx is
-      # traversable, so the deploy account can read them as itself. Reaching
-      # for sudo here would have meant widening the scoped whitelist for
-      # something it does not need.
+      # A diagnostic, and diagnostics must not fail a deploy that worked --
+      # which is exactly what happened when this was assumed to be readable
+      # without sudo. /var/log/nginx is drwx--x--x (traversable but not
+      # listable) and logrotate creates the files 0640 nginx:root, so the
+      # deploy account cannot read them as itself. Try sudo, fall back to
+      # saying so, and never exit non-zero either way.
       - name: nginx log
         if: always() && steps.auth.outcome == 'success'
+        continue-on-error: true
         run: |
           for log in rf.internal rustingface.com; do
             echo "--- /var/log/nginx/$log.error.log"
-            ssh "$PROXY_HOST" "tail -n 30 /var/log/nginx/$log.error.log"
+            ssh "$PROXY_HOST" "sudo -n tail -n 30 /var/log/nginx/$log.error.log 2>/dev/null \
+              || echo '(not readable by the deploy account; add tail to its sudo allowlist to see this)'"
           done

CLAUDE.md

@@ -25,9 +25,22 @@ 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.
+  recorded is invisible the moment the instance is sealed. This is also why a
+  cold range that stops short of the end is answered `503` rather than served:
+  its tail could only be released early by signalling "durable and recorded"
+  over bytes that are neither.
 
 ## Testing
 

asset/config/config.toml.tmpl

@@ -12,8 +12,27 @@
 # 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"
+# How long in-flight responses may drain on SIGTERM before the service stops
+# anyway. A blob response lasts size/bandwidth, so an unbounded drain never
+# converges: systemd's TimeoutStopSec expires and escalates to SIGABRT. Cutting
+# a download deliberately is the better failure -- the client retries.
+shutdown_grace       = "15s"
+# Where an in-flight transfer's bytes are spooled so a second request for the
+# same blob can be served from it while it runs, rather than waiting out the
+# whole transfer with nothing on the wire. This is what makes single-flight's
+# promise real. It holds no state: losing it loses nothing, because a blob is
+# not recorded until it is durable in the bucket. PrivateTmp=true gives the unit
+# its own /tmp, so this path is not shared with anything else on the host.
+spool_dir            = "/tmp/rustingface-spool"
+# How long a request that joined an in-flight fetch waits for that fetch's spool
+# to appear before giving up and returning 503. Not a wait for the transfer --
+# only for it to start.
+flight_wait_timeout  = "5s"
 
 [storage]
 endpoint               = "{{S3_ENDPOINT}}"
@@ -26,6 +45,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"
 

asset/systemd/rustingface.service

@@ -21,9 +21,12 @@ LoadCredential=s3-access-key:/etc/rustingface/s3-access-key
 LoadCredential=s3-secret-key:/etc/rustingface/s3-secret-key
 LoadCredential=hf-token:/etc/rustingface/hf-token
 
-# Long transfers must not be killed mid-flight on stop; a detached upload that
-# is cut short leaves an orphan blob, which is safe but wasteful.
-TimeoutStopSec=120
+# A backstop, not the mechanism. The service caps its own drain at
+# server.shutdown_grace and then exits cleanly, because a blob response lasts
+# size/bandwidth and waiting for one to finish does not converge -- letting this
+# expire instead means SIGABRT and a core dump. Keep it comfortably above the
+# configured grace so systemd never escalates first.
+TimeoutStopSec=60
 
 # Hardening. The process holds no durable state — the bucket is the whole of it
 # — so there is no StateDirectory to protect and ProtectSystem can stay strict

crates/rustingface-api/src/error.rs

@@ -52,6 +52,15 @@ impl IntoResponse for ApiError {
         {
             response.headers_mut().insert(X_ERROR_CODE, value);
         }
+
+        // Sent for correctness rather than in hope: `huggingface_hub` ignores
+        // it and sleeps its own exponential backoff. A client that does read
+        // it should not hammer a transfer that takes minutes.
+        if matches!(err, Error::FetchInProgress(_)) {
+            response
+                .headers_mut()
+                .insert(header::RETRY_AFTER, header::HeaderValue::from_static("5"));
+        }
         // Header values must be visible ASCII; a path or upstream detail could
         // carry anything, so a message that will not encode is simply dropped
         // rather than replacing the response with a 500.

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

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

crates/rustingface-api/tests/sovereignty.rs

@@ -253,6 +253,101 @@ async fn a_sealed_instance_refuses_loudly_for_anything_it_never_stored() {
     );
 }
 
+#[tokio::test]
+async fn a_follower_streams_from_the_transfer_instead_of_waiting_it_out() {
+    // Spec §7 says concurrent callers "subscribe to the same broadcast". They
+    // could not: the leader's live stream cannot be replayed to someone who
+    // arrived mid-body, and the multipart upload is unreadable until it
+    // completes, so a follower simply waited -- with nothing on the wire --
+    // for the whole transfer. On a multi-gigabyte blob that is minutes against
+    // a client that gives up in ten seconds.
+    //
+    // The spool is the missing buffer: the leader writes each chunk to local
+    // disk as it passes, and a follower reads that from byte zero, following
+    // it as it grows.
+    let hub = FakeHub::start().await;
+    let weights = publish_model(&hub).await;
+    // Slow enough that the leader is still going well after the follower has
+    // started reading; the whole point is what happens during that window.
+    hub.throttle(2 * 1024, Duration::from_millis(20)).await;
+
+    let bucket = bucket_dir("follower-streams");
+    let rf = instance(&bucket, Some(&hub), |_| {});
+    hub.reset_counts();
+
+    let leader = {
+        let app = rf.app.clone();
+        tokio::spawn(async move {
+            call(
+                &app,
+                "GET",
+                "/Qwen/Qwen3-32B/resolve/main/model.safetensors",
+                &[],
+            )
+            .await
+        })
+    };
+
+    // Join once the leader's fetch is genuinely under way.
+    for _ in 0..400 {
+        if hub.file_gets() >= 1 {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(5)).await;
+    }
+
+    let response = rf
+        .app
+        .clone()
+        .oneshot(
+            Request::builder()
+                .uri("/Qwen/Qwen3-32B/resolve/main/model.safetensors")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(
+        response.status(),
+        StatusCode::OK,
+        "a follower is served, not turned away"
+    );
+
+    use futures::StreamExt;
+    let mut body = response.into_body().into_data_stream();
+    let first = body
+        .next()
+        .await
+        .expect("a chunk while the leader is still transferring")
+        .unwrap();
+
+    // The assertion that matters. Chunked delivery proves nothing -- the old
+    // wait-it-out path also returned the file in chunks, just all of them
+    // after the transfer had finished. What distinguishes streaming from
+    // waiting is *when* the first byte lands: here, while the leader is still
+    // reading from upstream.
+    assert!(
+        !leader.is_finished(),
+        "the follower's first bytes arrived while the leader was still transferring"
+    );
+    assert!(!first.is_empty());
+
+    let mut served = first.to_vec();
+    while let Some(chunk) = body.next().await {
+        served.extend_from_slice(&chunk.unwrap());
+    }
+    assert_eq!(served, weights, "and it receives the whole, correct file");
+
+    let (status, _, leader_body) = leader.await.unwrap();
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(leader_body, weights);
+    assert_eq!(
+        hub.file_gets(),
+        1,
+        "and it cost no second upstream fetch: single-flight still holds"
+    );
+}
+
 #[tokio::test]
 async fn eight_concurrent_clients_produce_exactly_one_upstream_fetch() {
     let hub = FakeHub::start().await;
@@ -364,6 +459,107 @@ 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_cold_resume_to_the_end_of_the_file_is_not_deferred() {
+    // The counterpart to the bounded-range case: a range that runs to the end
+    // of the file *can* be completed from the transfer, because its holdback
+    // releases at the natural end. This is the shape a resumed `hf download`
+    // sends, so it must keep streaming rather than be turned away with a 503.
+    let hub = FakeHub::start().await;
+    let weights = publish_model(&hub).await;
+    hub.throttle(8 * 1024, Duration::from_millis(2)).await;
+
+    let bucket = bucket_dir("cold-resume-eof");
+    let rf = instance(&bucket, Some(&hub), |_| {});
+
+    let (status, headers, body) = call(
+        &rf.app,
+        "GET",
+        "/Qwen/Qwen3-32B/resolve/main/model.safetensors",
+        &[("range", "bytes=1000-")],
+    )
+    .await;
+    assert_eq!(
+        status,
+        StatusCode::PARTIAL_CONTENT,
+        "a resume to EOF is served from the tee, not deferred"
+    );
+    assert_eq!(body, weights[1000..]);
+    assert_eq!(
+        headers["content-range"],
+        format!("bytes 1000-{}/{}", weights.len() - 1, weights.len())
+    );
+}
+
 #[tokio::test]
 async fn a_ranged_request_resumes_from_the_offset_it_asks_for() {
     let hub = FakeHub::start().await;
@@ -371,8 +567,40 @@ async fn a_ranged_request_resumes_from_the_offset_it_asks_for() {
     let bucket = bucket_dir("range");
     let rf = instance(&bucket, Some(&hub), |_| {});
 
-    // Cold, with a range: the whole file must be stored before any part of it
-    // is served, or the bucket would hold a fragment.
+    // Cold, with a range that stops short of the end. The whole file must be
+    // stored before any part of it is served, or the bucket would hold a
+    // fragment -- and this range cannot be *completed* from the transfer
+    // either, because its last chunk is withheld until the digest is verified
+    // over every byte. Rather than hand the client its slice and then stall it
+    // on that tail, say so; the fetch runs on regardless.
+    let (status, headers, _) = call(
+        &rf.app,
+        "GET",
+        "/Qwen/Qwen3-32B/resolve/main/model.safetensors",
+        &[("range", "bytes=100000-100999")],
+    )
+    .await;
+    assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
+    assert_eq!(headers["retry-after"], "5");
+
+    let oid = {
+        use sha2::{Digest, Sha256};
+        hex::encode(Sha256::digest(&weights))
+    };
+    let blob = bucket.join(key::blob(&oid));
+    for _ in 0..400 {
+        if blob.exists() && std::fs::metadata(&blob).unwrap().len() == weights.len() as u64 {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(25)).await;
+    }
+    assert_eq!(
+        std::fs::metadata(&blob).unwrap().len(),
+        weights.len() as u64,
+        "a ranged cold request stores the whole file, never a fragment"
+    );
+
+    // And the retry the 503 asked for is served from the bucket, in full.
     let (status, headers, body) = call(
         &rf.app,
         "GET",
@@ -388,18 +616,6 @@ async fn a_ranged_request_resumes_from_the_offset_it_asks_for() {
     );
     assert_eq!(headers["content-length"], "1000");
 
-    let oid = {
-        use sha2::{Digest, Sha256};
-        hex::encode(Sha256::digest(&weights))
-    };
-    assert_eq!(
-        std::fs::metadata(bucket.join(key::blob(&oid)))
-            .unwrap()
-            .len(),
-        weights.len() as u64,
-        "a ranged cold request stores the whole file, never a fragment"
-    );
-
     // Warm, resuming to the end.
     let (status, _, tail) = call(
         &rf.app,

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;
@@ -21,6 +22,7 @@ use crate::digest::{self, Digest};
 use crate::flight::Outcome;
 use crate::ports::{Store, UpstreamHead};
 use crate::registry::{FetchTask, LeadResult, Registry, TEE_DEPTH};
+use crate::spool::Spool;
 
 /// Where a transfer's bytes are being written while the digest is unknown.
 enum Sink {
@@ -36,16 +38,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<Range<u64>>,
     ) -> Result<LeadResult> {
         let body = self
             .upstream()
@@ -72,6 +77,27 @@ impl Registry {
         let (client_tx, client_rx) = mpsc::channel::<Result<Bytes>>(TEE_DEPTH);
         let (outcome_tx, outcome_rx) = oneshot::channel::<Outcome>();
 
+        // Spool the bytes as they pass so a second request for the same blob
+        // can be served from this transfer rather than waiting it out. Every
+        // transfer is spooled, not just the large ones: the cost on a small
+        // file is a create and a delete, and making it conditional would mean
+        // followers behave one way for a config file and another for a shard,
+        // which is the kind of size-dependent behaviour this service has
+        // already been bitten by once.
+        let total = head.linked_size.or(head.content_length).unwrap_or(0);
+        let spool = match &self.config().server.spool_dir {
+            Some(dir) => match Spool::create(dir, total, head.clone()) {
+                Ok(spool) => Some(spool),
+                Err(err) => {
+                    // Not fatal: without a spool, followers are told to retry.
+                    // Losing the transfer over it would be worse.
+                    warn!(%err, dir = %dir.display(), "could not open a spool for this transfer; followers will be told to retry");
+                    None
+                }
+            },
+            None => None,
+        };
+
         let transfer = Transfer {
             store: Arc::clone(self.store()),
             registry_repo_type: repo_type,
@@ -82,16 +108,10 @@ impl Registry {
             sink,
             part_size: self.config().storage.multipart_part_size,
             stall_timeout: self.config().server.client_stall_timeout,
+            client_range: client_range.clone(),
+            spool: spool.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(&registry, body.stream, client_tx).await;
@@ -106,6 +126,8 @@ impl Registry {
             head,
             rx: client_rx,
             task: FetchTask::new(outcome_rx),
+            range: client_range,
+            spool,
         })
     }
 
@@ -137,6 +159,11 @@ 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<Range<u64>>,
+    /// Where the passing bytes are also written so other requests for the same
+    /// blob can be served from them while this transfer runs.
+    spool: Option<Arc<Spool>>,
 }
 
 impl Transfer {
@@ -166,19 +193,42 @@ 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()),
                 }
 
+                // Publish only after the write lands: a follower reading ahead
+                // of the bytes actually on disk would get zeroes or a short
+                // read, which is the silent corruption this service exists to
+                // prevent.
+                if let Some(spool) = &self.spool {
+                    spool.append(&chunk).await?;
+                    spool.published(offset);
+                }
+
+                // 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 +366,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<Bytes> {
+        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 {

crates/rustingface-core/src/flight.rs

@@ -19,6 +19,8 @@ use rustingface_entities::manifest::Entry;
 use rustingface_entities::repo::{RepoId, RepoType};
 use tokio::sync::watch;
 
+use crate::spool::Spool;
+
 /// Identifies one uncached blob.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct FlightKey {
@@ -90,6 +92,30 @@ impl From<&Error> for Outcome {
 #[derive(Debug)]
 pub struct Flight {
     rx: watch::Receiver<Option<Outcome>>,
+    spool: watch::Receiver<Option<Arc<Spool>>>,
+}
+
+impl Flight {
+    /// A receiver for the flight's outcome, for a reader that needs to know
+    /// when the blob became durable.
+    pub fn outcome(&self) -> watch::Receiver<Option<Outcome>> {
+        self.rx.clone()
+    }
+
+    /// The leader's spool, once it has one.
+    ///
+    /// A follower arrives before the leader has read a byte, so this waits
+    /// briefly for the spool to be published rather than concluding there is
+    /// none. `None` means the leader is not spooling — a small buffered file,
+    /// or a transfer that failed before it started — and the caller falls back
+    /// to waiting for the outcome.
+    pub async fn spool(&mut self, within: std::time::Duration) -> Option<Arc<Spool>> {
+        if let Some(spool) = self.spool.borrow_and_update().clone() {
+            return Some(spool);
+        }
+        let _ = tokio::time::timeout(within, self.spool.changed()).await;
+        self.spool.borrow().clone()
+    }
 }
 
 impl Flight {
@@ -117,9 +143,10 @@ impl Flight {
 /// and closes the channel, so a panicking leader cannot wedge later requests
 /// on a key that will never complete.
 pub struct Lead {
-    map: Arc<DashMap<FlightKey, watch::Receiver<Option<Outcome>>>>,
+    map: Arc<DashMap<FlightKey, Shared>>,
     key: FlightKey,
     tx: watch::Sender<Option<Outcome>>,
+    spool: watch::Sender<Option<Arc<Spool>>>,
 }
 
 impl Lead {
@@ -128,6 +155,14 @@ impl Lead {
         self.map.remove(&self.key);
         let _ = self.tx.send(Some(outcome));
     }
+
+    /// Offer this transfer's spool to everyone waiting on the flight.
+    ///
+    /// Called as soon as the transfer starts, so a follower that arrived
+    /// moments earlier can begin streaming from byte zero instead of waiting.
+    pub fn publish_spool(&self, spool: Arc<Spool>) {
+        let _ = self.spool.send(Some(spool));
+    }
 }
 
 impl Drop for Lead {
@@ -136,6 +171,13 @@ impl Drop for Lead {
     }
 }
 
+/// What a flight publishes to its followers.
+#[derive(Clone, Debug)]
+struct Shared {
+    outcome: watch::Receiver<Option<Outcome>>,
+    spool: watch::Receiver<Option<Arc<Spool>>>,
+}
+
 /// Either this caller leads the fetch, or it waits on one already running.
 pub enum Role {
     Leader(Lead),
@@ -145,7 +187,7 @@ pub enum Role {
 /// The in-flight registry.
 #[derive(Debug, Default)]
 pub struct Flights {
-    map: Arc<DashMap<FlightKey, watch::Receiver<Option<Outcome>>>>,
+    map: Arc<DashMap<FlightKey, Shared>>,
 }
 
 impl Flights {
@@ -162,15 +204,21 @@ impl Flights {
         use dashmap::mapref::entry::Entry as MapEntry;
         match self.map.entry(key.clone()) {
             MapEntry::Occupied(occupied) => Role::Follower(Flight {
-                rx: occupied.get().clone(),
+                rx: occupied.get().outcome.clone(),
+                spool: occupied.get().spool.clone(),
             }),
             MapEntry::Vacant(vacant) => {
                 let (tx, rx) = watch::channel(None);
-                vacant.insert(rx);
+                let (spool_tx, spool_rx) = watch::channel(None);
+                vacant.insert(Shared {
+                    outcome: rx,
+                    spool: spool_rx,
+                });
                 Role::Leader(Lead {
                     map: Arc::clone(&self.map),
                     key,
                     tx,
+                    spool: spool_tx,
                 })
             }
         }

crates/rustingface-core/src/lib.rs

@@ -13,6 +13,7 @@ pub mod flight;
 pub mod policy;
 pub mod ports;
 pub mod registry;
+pub mod spool;
 
 pub use policy::Policy;
 pub use registry::{Body, Registry, ResolvedFile, Source};

crates/rustingface-core/src/registry.rs

@@ -719,8 +719,56 @@ impl Registry {
 
         let key = FlightKey::new(repo_type, repo_id.clone(), &commit, path);
         match self.inner.flights.join(key) {
-            Role::Follower(flight) => {
+            Role::Follower(mut flight) => {
                 debug!(%repo_id, commit, path, "joining an in-flight upstream fetch");
+
+                // Spec §7: followers subscribe to the same broadcast rather
+                // than issuing their own request. The spool is what makes that
+                // possible -- the leader's live stream cannot be replayed, but
+                // its bytes on disk can. Without one there is nothing to read
+                // and the only honest answer is to come back later.
+                let grace = self.inner.config.server.flight_wait_timeout;
+                if let Some(spool) = flight.spool(grace).await {
+                    let total = spool.total();
+                    let asked = match &range {
+                        // Same rule as a cold ranged lead: a range that stops
+                        // short of the end cannot be completed before the whole
+                        // file is verified.
+                        Some(asked) if asked.end != total => {
+                            return Err(Error::FetchInProgress(format!(
+                                "{repo_id}@{commit}/{path} is being fetched; a range that stops \
+                                 short of the end can only be served once it is stored. Retry \
+                                 shortly."
+                            )));
+                        }
+                        Some(asked) => asked.clone(),
+                        None => 0..total,
+                    };
+                    let head = spool.head().clone();
+                    let len = asked.end - asked.start;
+                    let ranged = range.clone();
+                    let stream = crate::spool::read_while_written(
+                        Arc::clone(&spool),
+                        asked,
+                        flight.outcome(),
+                    );
+                    return Ok((
+                        ResolvedFile {
+                            commit,
+                            size: total,
+                            etag: head.etag,
+                            linked_etag: head.linked_etag,
+                            linked_size: head.linked_size,
+                            body: Body::Stream {
+                                stream,
+                                len,
+                                range: ranged,
+                            },
+                        },
+                        Source::Upstream,
+                    ));
+                }
+
                 let entry = flight.wait().await?;
                 let file = self
                     .serve_stored(&entry, &commit, range)
@@ -734,7 +782,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 +801,69 @@ impl Registry {
                             })?;
                         Ok((file, Source::Bucket))
                     }
-                    Ok(LeadResult::Streaming { head, rx, task }) => {
+                    Ok(LeadResult::Streaming {
+                        head,
+                        rx,
+                        task,
+                        range,
+                        spool,
+                    }) => {
+                        // Offer the spool before handing the flight over, so a
+                        // follower already waiting can start reading at once.
+                        if let Some(spool) = spool {
+                            lead.publish_spool(spool);
+                        }
                         // 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 range that stops short of the end cannot be
+                        // completed from this transfer. Its bytes arrive
+                        // quickly, but the last of them is withheld until the
+                        // manifest write lands -- and that waits on the whole
+                        // file, because the digest is only verified once every
+                        // byte has passed. Releasing the tail early would
+                        // signal "durable and recorded" over bytes that are
+                        // neither, which is the one thing the holdback exists
+                        // to prevent.
+                        //
+                        // So the client would receive its slice and then stall
+                        // a few hundred bytes from the end for the length of a
+                        // multi-gigabyte transfer. Say so instead. The fetch is
+                        // already running and detached, so a retry is served
+                        // from the bucket.
+                        //
+                        // A range that extends to the end of the file -- what a
+                        // resumed `hf download` sends -- is not affected: its
+                        // holdback releases at the natural end of the transfer.
+                        if let Some(asked) = &range
+                            && asked.end != size
+                        {
+                            drop(rx);
+                            warn!(
+                                %repo_id, commit, path, ?asked, size,
+                                "a bounded range on a file the bucket does not hold cannot be \
+                                 completed until the whole file is verified; answering 503 rather \
+                                 than stalling the client on its last chunk"
+                            );
+                            return Err(Error::FetchInProgress(format!(
+                                "{repo_id}@{commit}/{path} is being fetched; a range that stops \
+                                 short of the end can only be served once it is stored. Retry \
+                                 shortly."
+                            )));
+                        }
+
+                        // A range to the end of the file 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 +873,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 +991,12 @@ pub(crate) enum LeadResult {
     Streaming {
         head: UpstreamHead,
         rx: mpsc::Receiver<Result<Bytes>>,
+        /// The slice of the file the tee is narrowed to, if any.
+        range: Option<Range<u64>>,
+        /// Where the transfer's bytes are also landing, for followers to read.
+        spool: Option<Arc<crate::spool::Spool>>,
         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.

crates/rustingface-core/src/spool.rs

@@ -0,0 +1,244 @@
+//! The spool: an in-flight transfer's bytes, readable before it finishes.
+//!
+//! Single-flight (spec §7) says concurrent requests for the same uncached blob
+//! must produce exactly one upstream fetch, and that "subsequent callers
+//! subscribe to the same broadcast". Subscribing is the part that was missing:
+//! a follower cannot be handed the leader's live stream, because the bytes
+//! that already passed are gone from it and the multipart upload is not
+//! readable until it completes. So it waited instead — with nothing on the
+//! wire — for the whole transfer, which on a multi-gigabyte blob is minutes
+//! against a client that gives up in ten seconds.
+//!
+//! The spool is the missing buffer. The leader writes every chunk to a local
+//! file as it passes and publishes how much is safe to read; a follower opens
+//! that file and streams it from the start, waiting for growth when it catches
+//! up. One upstream fetch, every caller served, nobody in silence.
+//!
+//! It holds no state. Losing it loses nothing: the blob is not recorded until
+//! it is durable in the bucket, so a crash mid-transfer leaves the file
+//! unreferenced and the next request refetches. It is the same role the
+//! in-memory buffer already plays for small files, on disk so it can hold a
+//! large one.
+
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+use tokio::sync::watch;
+
+use crate::ports::UpstreamHead;
+
+/// A transfer's bytes on local disk, growing as they arrive.
+#[derive(Debug)]
+pub struct Spool {
+    path: PathBuf,
+    total: u64,
+    head: UpstreamHead,
+    /// Bytes durably written to `path` and safe for a follower to read.
+    written: watch::Sender<u64>,
+}
+
+impl Spool {
+    /// Create a spool for a transfer of `total` bytes.
+    pub fn create(dir: &Path, total: u64, head: UpstreamHead) -> std::io::Result<Arc<Self>> {
+        std::fs::create_dir_all(dir)?;
+        let path = dir.join(format!("spool-{}", uuid()));
+        // Created here so a follower can open it the moment it is published.
+        std::fs::File::create(&path)?;
+        Ok(Arc::new(Self {
+            path,
+            total,
+            head,
+            written: watch::channel(0).0,
+        }))
+    }
+
+    pub fn path(&self) -> &Path {
+        &self.path
+    }
+
+    pub fn total(&self) -> u64 {
+        self.total
+    }
+
+    pub fn head(&self) -> &UpstreamHead {
+        &self.head
+    }
+
+    /// Watch how much of the file is readable.
+    pub fn progress(&self) -> watch::Receiver<u64> {
+        self.written.subscribe()
+    }
+
+    /// Append a chunk. The file is opened per call rather than held, so the
+    /// leader's handle cannot outlive a failed transfer and strand the file.
+    pub async fn append(&self, chunk: &[u8]) -> rustingface_entities::error::Result<()> {
+        use tokio::io::AsyncWriteExt;
+        let mut file = tokio::fs::OpenOptions::new()
+            .append(true)
+            .open(&self.path)
+            .await
+            .map_err(|e| spool_error(&self.path, e))?;
+        file.write_all(chunk)
+            .await
+            .map_err(|e| spool_error(&self.path, e))?;
+        Ok(())
+    }
+
+    /// Announce that `n` bytes are on disk. Call only after the write lands.
+    pub fn published(&self, n: u64) {
+        let _ = self.written.send(n);
+    }
+}
+
+impl Drop for Spool {
+    fn drop(&mut self) {
+        // Every reader holds an Arc, so this runs once the last one is done.
+        let _ = std::fs::remove_file(&self.path);
+    }
+}
+
+/// A spool failure is a storage failure: the bytes did not land where the
+/// next reader expects them.
+fn spool_error(path: &Path, err: std::io::Error) -> rustingface_entities::error::Error {
+    rustingface_entities::error::Error::Storage(format!("spool {}: {err}", path.display()))
+}
+
+/// A unique-enough name; the directory is ours alone.
+fn uuid() -> String {
+    use std::sync::atomic::{AtomicU64, Ordering};
+    use std::time::{SystemTime, UNIX_EPOCH};
+    static SEQ: AtomicU64 = AtomicU64::new(0);
+    let nanos = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_nanos())
+        .unwrap_or(0);
+    format!("{nanos:x}-{:x}", SEQ.fetch_add(1, Ordering::Relaxed))
+}
+
+/// Stream a spool's bytes from `range.start`, following it as it grows.
+///
+/// The final chunk is withheld until the flight reports the blob stored, for
+/// the same reason the tee runs one chunk behind: a response that completes
+/// before the manifest write would tell the client its bytes are recorded when
+/// they are not. A failed flight ends the stream with that failure rather than
+/// a short read.
+pub fn read_while_written(
+    spool: Arc<Spool>,
+    range: std::ops::Range<u64>,
+    outcome: watch::Receiver<Option<crate::flight::Outcome>>,
+) -> futures::stream::BoxStream<'static, rustingface_entities::error::Result<bytes::Bytes>> {
+    use futures::StreamExt;
+    use tokio::io::{AsyncReadExt, AsyncSeekExt};
+
+    const CHUNK: usize = 1024 * 1024;
+
+    let progress = spool.progress();
+    let stream = futures::stream::unfold(
+        (spool, range.start, range.end, None::<bytes::Bytes>, false),
+        move |(spool, mut pos, end, held, mut done)| {
+            let mut progress = progress.clone();
+            let mut outcome = outcome.clone();
+            async move {
+                loop {
+                    // Everything asked for has been read; release the withheld
+                    // tail once the blob is durable and referenced.
+                    if pos >= end {
+                        let tail = held?;
+                        loop {
+                            // Bound the borrow before awaiting: holding a
+                            // watch guard across an await makes the stream
+                            // non-Send.
+                            let current = outcome.borrow_and_update().clone();
+                            match current {
+                                Some(crate::flight::Outcome::Stored(_)) => break,
+                                Some(failed) => {
+                                    let err = failed.into_result().unwrap_err();
+                                    return Some((Err(err), (spool, pos, end, None, true)));
+                                }
+                                None => {
+                                    if outcome.changed().await.is_err() {
+                                        return Some((
+                                            Err(rustingface_entities::error::Error::Storage(
+                                                "the transfer this response was reading from ended without reporting an outcome".into(),
+                                            )),
+                                            (spool, pos, end, None, true),
+                                        ));
+                                    }
+                                }
+                            }
+                        }
+                        return Some((Ok(tail), (spool, pos, end, None, true)));
+                    }
+
+                    let available = *progress.borrow_and_update();
+                    if available > pos {
+                        let want = (available.min(end) - pos).min(CHUNK as u64) as usize;
+                        let mut buf = vec![0u8; want];
+                        let read = async {
+                            let mut file = tokio::fs::File::open(spool.path()).await?;
+                            file.seek(std::io::SeekFrom::Start(pos)).await?;
+                            file.read_exact(&mut buf).await?;
+                            Ok::<_, std::io::Error>(())
+                        }
+                        .await;
+                        if let Err(err) = read {
+                            return Some((
+                                Err(spool_error(spool.path(), err)),
+                                (spool, pos, end, None, true),
+                            ));
+                        }
+                        pos += want as u64;
+                        let chunk = bytes::Bytes::from(buf);
+                        // One chunk behind, exactly like the tee.
+                        let previous = held;
+                        return match previous {
+                            Some(previous) => {
+                                Some((Ok(previous), (spool, pos, end, Some(chunk), done)))
+                            }
+                            None => {
+                                // Nothing to emit yet; go round again holding
+                                // this one.
+                                let state = (spool, pos, end, Some(chunk), done);
+                                Some((Ok(bytes::Bytes::new()), state))
+                            }
+                        };
+                    }
+
+                    // Caught up with the writer. Either more is coming, or the
+                    // transfer has ended and this is all there will be.
+                    if done {
+                        return None;
+                    }
+                    tokio::select! {
+                        changed = progress.changed() => {
+                            if changed.is_err() {
+                                done = true;
+                            }
+                        }
+                        changed = outcome.changed() => {
+                            if changed.is_err() {
+                                done = true;
+                            } else {
+                                let current = outcome.borrow().clone();
+                                if let Some(failed @ crate::flight::Outcome::Failed { .. }) = current
+                                {
+                                    let err = failed.into_result().unwrap_err();
+                                    return Some((Err(err), (spool, pos, end, None, true)));
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        },
+    );
+
+    // The empty frames emitted while priming the holdback are noise on the
+    // wire; drop them rather than making the client parse zero-length chunks.
+    stream
+        .filter(|item| {
+            let keep = !matches!(item, Ok(bytes) if bytes.is_empty());
+            async move { keep }
+        })
+        .boxed()
+}

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<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"));
+    }
+}

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,13 +47,38 @@ 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")]
     pub client_stall_timeout: Duration,
+    /// How long a request that joined another request's in-flight fetch waits
+    /// for that fetch's spool to appear before giving up on it.
+    ///
+    /// A follower arrives before the leader has read a byte, so this is not a
+    /// wait for the *transfer* but for it to start. It must stay well under a
+    /// client's read timeout -- `huggingface_hub` allows 10s.
+    #[serde(with = "humantime_serde")]
+    pub flight_wait_timeout: Duration,
+    /// Where an in-flight transfer's bytes are spooled so that other requests
+    /// for the same blob can be served from it while it runs.
+    ///
+    /// This is what makes single-flight's promise real: without it a second
+    /// request for the same uncached blob has nothing to read and can only
+    /// wait for the whole transfer. `None` disables spooling, at the cost of
+    /// returning those requests a 503 instead. Holds no state — losing it
+    /// loses nothing, because a blob is not recorded until it is durable in
+    /// the bucket.
+    pub spool_dir: Option<PathBuf>,
+    /// How long to let in-flight requests drain on SIGTERM before stopping
+    /// anyway.
+    ///
+    /// A blob response is an in-flight request that lasts `size / bandwidth`,
+    /// so an unbounded drain does not converge on a restart: systemd's
+    /// `TimeoutStopSec` expires and escalates to SIGABRT, which is a core dump
+    /// rather than a shutdown. Cutting a download deliberately is the better
+    /// failure -- clients retry, and nothing in the bucket depends on it.
+    #[serde(with = "humantime_serde")]
+    pub shutdown_grace: Duration,
 }
 
 impl Default for Server {
@@ -54,8 +86,10 @@ 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),
+            flight_wait_timeout: Duration::from_secs(5),
+            spool_dir: Some(PathBuf::from("/tmp/rustingface-spool")),
+            shutdown_grace: Duration::from_secs(15),
         }
     }
 }
@@ -84,6 +118,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 +147,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 +398,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 +406,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 +434,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());
     }
 

crates/rustingface-entities/src/error.rs

@@ -66,6 +66,15 @@ pub enum Error {
     #[error("disabled repo: {0}")]
     DisabledRepo(String),
 
+    /// The blob is being fetched and this request cannot be served from that
+    /// transfer promptly, so the client is asked to come back.
+    ///
+    /// Holding the connection open instead would be a lie: the client would
+    /// receive nothing, or receive its bytes and then stall on a tail that
+    /// cannot be released until the whole file is verified and recorded.
+    #[error("fetch in progress: {0}")]
+    FetchInProgress(String),
+
     /// Upstream authentication failed (bad or missing `upstream.token_file`).
     #[error("upstream authentication failed: {0}")]
     UpstreamUnauthorized(String),
@@ -132,6 +141,7 @@ impl Error {
             Self::GatedRepo(_) | Self::DisabledRepo(_) | Self::PolicyDenied(_) => 403,
             Self::UpstreamUnauthorized(_) | Self::Unauthorized => 401,
             Self::BadRequest(_) => 400,
+            Self::FetchInProgress(_) => 503,
             Self::RangeNotSatisfiable(_) => 416,
             Self::DigestMismatch { .. } | Self::Storage(_) | Self::Upstream(_) => 502,
             Self::Config(_) | Self::Internal(_) => 500,
@@ -150,6 +160,7 @@ impl Error {
             Self::Unauthorized => "unauthorized",
             Self::PolicyDenied(_) => "policy_denied",
             Self::BadRequest(_) => "bad_request",
+            Self::FetchInProgress(_) => "fetch_in_progress",
             Self::RangeNotSatisfiable(_) => "range_not_satisfiable",
             Self::DigestMismatch { .. } => "digest_mismatch",
             Self::Storage(_) => "storage",

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)

crates/rustingface/src/cmd/serve.rs

@@ -70,7 +70,7 @@ pub async fn run(config: Config) -> Result<()> {
     let app = router(AppState::new(registry, tokens.into(), metrics));
 
     axum::serve(listener, app)
-        .with_graceful_shutdown(shutdown())
+        .with_graceful_shutdown(shutdown(config.server.shutdown_grace))
         .await
         .context("serving")
 }
@@ -87,12 +87,22 @@ fn token_file_has_content(config: &Config) -> bool {
     })
 }
 
-/// Stop accepting on `SIGTERM`/`SIGINT` and let in-flight requests drain.
+/// Stop accepting on `SIGTERM`/`SIGINT` and let in-flight requests drain --
+/// for a bounded time.
 ///
-/// Detached transfers are not requests and are not waited on here: the process
-/// exiting mid-upload leaves an orphan blob, which is safe and reclaimable,
-/// while a manifest is only ever written after an upload completes.
-async fn shutdown() {
+/// The bound is the point. A blob response *is* an in-flight request and it
+/// lasts `size / bandwidth`, so waiting for one to finish does not converge:
+/// a 3GB shard at 8MB/s holds the drain open for seven minutes, systemd's
+/// `TimeoutStopSec` expires long before that, and the service is killed with
+/// SIGABRT and a core dump. Cutting the download deliberately after
+/// `server.shutdown_grace` is strictly better -- the client retries, which is
+/// a normal event, and the restart is prompt.
+///
+/// Detached transfers are not requests and are not waited on here either: the
+/// process exiting mid-upload leaves an orphan blob, which is safe and
+/// reclaimable, while a manifest is only ever written after an upload
+/// completes.
+async fn shutdown(grace: std::time::Duration) {
     let mut term = match signal(SignalKind::terminate()) {
         Ok(signal) => signal,
         Err(err) => {
@@ -101,11 +111,26 @@ async fn shutdown() {
         }
     };
     tokio::select! {
-        _ = term.recv() => info!("SIGTERM received; draining"),
+        _ = term.recv() => info!(?grace, "SIGTERM received; draining"),
         result = tokio::signal::ctrl_c() => {
             if result.is_ok() {
-                info!("interrupt received; draining");
+                info!(?grace, "interrupt received; draining");
             }
         }
     }
+
+    // Returning here only stops the listener accepting; axum then waits on
+    // every in-flight response, blob streams included. This watchdog is what
+    // makes that wait bounded.
+    tokio::spawn(async move {
+        tokio::time::sleep(grace).await;
+        tracing::warn!(
+            ?grace,
+            "in-flight responses did not finish within server.shutdown_grace; \
+             stopping anyway and cutting them. Clients retry; stored blobs are \
+             unaffected, because a manifest entry is only written once its blob \
+             is durable."
+        );
+        std::process::exit(0);
+    });
 }

doc/spec.md

@@ -225,7 +225,10 @@ 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
+shutdown_grace = "15s"           # bounded drain on SIGTERM; see below
+spool_dir = "/tmp/rustingface-spool"   # in-flight bytes, readable by followers
+flight_wait_timeout = "5s"       # how long a follower waits for a spool
 
 [storage]
 endpoint               = "https://s3.internal.example:9000"
@@ -235,6 +238,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 +265,58 @@ 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.
+
+`server.shutdown_grace` is the one bound that is not idle-based, and
+deliberately so. It caps how long in-flight responses may drain on SIGTERM
+before the process stops regardless. A blob response is an in-flight request
+lasting `size / bandwidth`, so an unbounded drain does not converge: systemd's
+`TimeoutStopSec` expires first and escalates to `SIGABRT`. Cutting a download
+deliberately is the better failure — the client retries, and nothing in the
+bucket depends on the response, because a manifest entry is only ever written
+once its blob is durable.
+
+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.
+
+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.
+
+That holds only for a range running to the end of the file. A range that stops
+short of it cannot be *completed* from the transfer at all: its last chunk is
+withheld until the manifest write lands, and that waits on every remaining byte,
+because the digest is verified only once the whole file has passed. Such a
+request is answered `503` with `Retry-After` while the fetch runs on, rather
+than being handed its bytes and then stalled a few hundred bytes from the end.
+
 ### `policy.ref_resolution`
 
 - `freeze` (default) — the first resolution of a mutable ref is recorded and
@@ -355,6 +412,22 @@ subsequent callers subscribe to the same broadcast rather than issuing their own
 request. Without this, a fleet-wide rollout of a new model produces N concurrent
 40GB pulls.
 
+Subscribing is what the spool is for. A follower cannot be handed the leader's
+live stream — the bytes that already passed are gone from it, and the multipart
+upload is not readable until it completes — so without one it can only wait for
+the entire transfer, with nothing on the wire, which a client's read timeout
+ends long before. The leader therefore writes every chunk to `server.spool_dir`
+as it passes and publishes how much is safe to read; a follower opens that file
+and streams it from the start, following it as it grows. One upstream fetch,
+every caller served, nobody in silence.
+
+The spool is not state. Losing it loses nothing, because a blob is not recorded
+until it is durable in the bucket, so a crash mid-transfer leaves an
+unreferenced blob and the next request refetches. It is the role the in-memory
+buffer already plays for small files, on disk so it can hold a large one. With
+`spool_dir` unset there is nothing for a follower to read and it is answered
+`503` instead.
+
 ### Streaming tee
 
 The upstream response body is consumed once and written to two sinks

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}