Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9b35dfdc09
|
@@ -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
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user