diff --git a/CLAUDE.md b/CLAUDE.md index 15785da..b651456 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -37,7 +37,10 @@ The five guarantees in `doc/spec.md` §1 are contractual. In particular: 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 diff --git a/crates/rustingface-api/src/error.rs b/crates/rustingface-api/src/error.rs index 53fdada..f03a16b 100644 --- a/crates/rustingface-api/src/error.rs +++ b/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. diff --git a/crates/rustingface-api/tests/sovereignty.rs b/crates/rustingface-api/tests/sovereignty.rs index 2c0752f..3746215 100644 --- a/crates/rustingface-api/tests/sovereignty.rs +++ b/crates/rustingface-api/tests/sovereignty.rs @@ -433,6 +433,38 @@ async fn a_cold_ranged_request_streams_while_the_file_is_still_being_fetched() { ); } +#[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; @@ -440,8 +472,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", @@ -457,18 +521,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, diff --git a/crates/rustingface-core/src/registry.rs b/crates/rustingface-core/src/registry.rs index 2b38d7d..73c5f60 100644 --- a/crates/rustingface-core/src/registry.rs +++ b/crates/rustingface-core/src/registry.rs @@ -764,10 +764,48 @@ impl Registry { // 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. + + // 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, diff --git a/crates/rustingface-entities/src/error.rs b/crates/rustingface-entities/src/error.rs index a32d810..7a38a34 100644 --- a/crates/rustingface-entities/src/error.rs +++ b/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", diff --git a/doc/spec.md b/doc/spec.md index 225c89d..4184b00 100644 --- a/doc/spec.md +++ b/doc/spec.md @@ -308,6 +308,13 @@ 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