fix(data): record the Hub's headers, not the CDN's, across a redirect

The Hub answers a resolve request with a redirect: 307 to its own
resolve-cache for an ordinary file, 302 to a CDN for an LFS-backed one.
X-Repo-Commit, X-Linked-Etag and X-Linked-Size ride on that redirect
response; the CDN it points at carries none of them and sets an ETag of
its own.

reqwest was following the redirect itself, so head_of() only ever saw
the final hop. Every recorded etag was the CDN's, X-Linked-Etag and
X-Linked-Size were lost, and no file was ever recognised as LFS -- which
also meant expected_sha256() never fired, so nothing was verified
against upstream's oid and every transfer took the staging path instead
of writing straight to its content-addressed key.

Follow redirects by hand instead, absorbing replayable metadata only
while still on the Hub's origin and taking the content length from
whichever hop actually serves the bytes. The operator's upstream token
is dropped when a redirect crosses origins, since a presigned CDN URL
needs no authorization of ours and forwarding one leaks a credential.

Also derives the lfs flag from X-Linked-Size rather than the presence of
X-Linked-Etag: the Hub sends a linked etag for every file (a git blob
SHA-1 for ordinary ones), so the etag alone marked everything as LFS.

Found by running the conformance suite against the MinIO on caveman.

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

crates/rustingface-data/src/upstream.rs

@@ -6,8 +6,10 @@
 
 use async_trait::async_trait;
 use futures::StreamExt;
-use reqwest::header::{ACCEPT_ENCODING, AUTHORIZATION, ETAG, HeaderMap, HeaderValue, USER_AGENT};
-use reqwest::{Client, Response, StatusCode};
+use reqwest::header::{
+    ACCEPT_ENCODING, AUTHORIZATION, ETAG, HeaderMap, HeaderValue, LOCATION, USER_AGENT,
+};
+use reqwest::{Client, Method, Response, StatusCode};
 use rustingface_core::ports::{Upstream, UpstreamBody, UpstreamHead};
 use rustingface_entities::error::{Error, Result};
 use rustingface_entities::repo::{RepoId, RepoType};
@@ -57,6 +59,15 @@ impl HubClient {
 
         let client = Client::builder()
             .default_headers(headers)
+            // Redirects are followed by hand, not by reqwest. The Hub answers
+            // a resolve request for an LFS-backed file with a 302 whose
+            // headers carry X-Repo-Commit, X-Linked-Etag and X-Linked-Size --
+            // and the CDN it points at carries none of them. Letting the
+            // client library follow the redirect silently replaces the values
+            // rustingface exists to record with the CDN's own, so every
+            // recorded etag would be wrong and no file would ever look like
+            // LFS.
+            .redirect(reqwest::redirect::Policy::none())
             .connect_timeout(cfg.connect_timeout)
             .read_timeout(cfg.read_timeout)
             .build()
@@ -140,39 +151,151 @@ impl HubClient {
 
     /// Fetch and parse a JSON document from the Hub API.
     async fn get_json(&self, url: &str, what: &str) -> Result<Value> {
-        let response = self
-            .authorize(self.client.get(url))
-            .send()
-            .await
-            .map_err(|e| Error::Upstream(format!("{what}: {e}")))?;
-        let response = self.check(response, what).await?;
+        let (_, response) = self.send(Method::GET, url, what).await?;
         response
             .json()
             .await
             .map_err(|e| Error::Upstream(format!("{what}: response was not JSON: {e}")))
     }
+
+    /// Send a request, following redirects while keeping the Hub's metadata.
+    ///
+    /// The returned [`UpstreamHead`] takes each replayable field from the
+    /// first response that carried it -- the Hub's own -- and the content
+    /// length from the final one, which is the response that actually has the
+    /// bytes. That split is the whole point of following redirects here
+    /// rather than letting reqwest do it.
+    async fn send(
+        &self,
+        method: Method,
+        url: &str,
+        what: &str,
+    ) -> Result<(UpstreamHead, Response)> {
+        let mut head = UpstreamHead::default();
+        let mut url = url.to_owned();
+        let origin = origin_of(&url);
+
+        for _ in 0..MAX_REDIRECTS {
+            // A redirect to a presigned CDN URL is a different party. Sending
+            // the operator's Hub token to it would leak a credential to a host
+            // that has no business seeing one, and the presigned URL needs no
+            // authorization of ours.
+            let request = if origin_of(&url) == origin {
+                self.authorize(self.client.request(method.clone(), &url))
+            } else {
+                self.client.request(method.clone(), &url)
+            };
+            let response = request
+                .send()
+                .await
+                .map_err(|e| Error::Upstream(format!("{what}: {e}")))?;
+
+            let status = response.status();
+            let on_hub = origin_of(&url) == origin;
+            if status.is_redirection() {
+                // Metadata is only taken from the Hub. A CDN sets an `ETag` of
+                // its own -- for an LFS file the Hub sends none, so absorbing
+                // the CDN's would record a value no client would ever have
+                // seen from the Hub, and one the CDN is free to change.
+                if on_hub {
+                    head.absorb(&response);
+                }
+                let location = response
+                    .headers()
+                    .get(LOCATION)
+                    .and_then(|v| v.to_str().ok())
+                    .ok_or_else(|| {
+                        Error::Upstream(format!(
+                            "{what}: upstream returned {status} with no Location"
+                        ))
+                    })?;
+                url = resolve_location(&url, location)?;
+                continue;
+            }
+
+            let response = self.check(response, what).await?;
+            if on_hub {
+                head.absorb(&response);
+            }
+            // The length always comes from the response that actually has the
+            // bytes, whoever is serving them.
+            head.content_length = response.content_length().or(head.content_length);
+            return Ok((head, response));
+        }
+
+        Err(Error::Upstream(format!(
+            "{what}: upstream redirected more than {MAX_REDIRECTS} times"
+        )))
+    }
 }
 
-/// Read the replayable metadata out of an upstream response.
-fn head_of(response: &Response) -> UpstreamHead {
-    let headers = response.headers();
-    let string = |name: &str| {
-        headers
-            .get(name)
-            .and_then(|v| v.to_str().ok())
-            .map(str::to_owned)
-    };
-    let linked_etag = string(X_LINKED_ETAG);
-    UpstreamHead {
-        commit: string(X_REPO_COMMIT),
-        etag: headers
-            .get(ETAG)
-            .and_then(|v| v.to_str().ok())
-            .map(str::to_owned),
-        lfs: linked_etag.is_some(),
-        linked_etag,
-        linked_size: string(X_LINKED_SIZE).and_then(|v| v.parse().ok()),
-        content_length: response.content_length(),
+/// How many redirects to follow before giving up.
+///
+/// The Hub uses at most two in practice (resolve to the LFS bridge, bridge to
+/// the CDN); the limit exists so a redirect loop fails rather than hangs.
+const MAX_REDIRECTS: usize = 5;
+
+/// The scheme and authority of a URL, for deciding whether a redirect crosses
+/// into somebody else's origin.
+fn origin_of(url: &str) -> String {
+    let (scheme, rest) = url.split_once("://").unwrap_or(("", url));
+    let authority = rest.split('/').next().unwrap_or_default();
+    format!("{scheme}://{authority}")
+}
+
+/// Resolve a `Location` against the URL it came from.
+///
+/// The Hub uses relative redirects for `resolve/{ref}` to `resolve/{commit}`
+/// and absolute ones for the CDN, so both shapes have to work.
+fn resolve_location(base: &str, location: &str) -> Result<String> {
+    if location.contains("://") {
+        return Ok(location.to_owned());
+    }
+    let origin = origin_of(base);
+    if let Some(absolute) = location.strip_prefix('/') {
+        return Ok(format!("{origin}/{absolute}"));
+    }
+    let parent = base.rsplit_once('/').map(|(head, _)| head).unwrap_or(base);
+    Ok(format!("{parent}/{location}"))
+}
+
+/// Merging the replayable metadata out of each response in a redirect chain.
+trait Absorb {
+    /// Take any field this response carries that is not already recorded.
+    ///
+    /// First writer wins, so the Hub's own values survive a later hop.
+    fn absorb(&mut self, response: &Response);
+}
+
+impl Absorb for UpstreamHead {
+    fn absorb(&mut self, response: &Response) {
+        let headers = response.headers();
+        let string = |name: &str| {
+            headers
+                .get(name)
+                .and_then(|v| v.to_str().ok())
+                .map(str::to_owned)
+        };
+        if self.commit.is_none() {
+            self.commit = string(X_REPO_COMMIT);
+        }
+        if self.etag.is_none() {
+            self.etag = headers
+                .get(ETAG)
+                .and_then(|v| v.to_str().ok())
+                .map(str::to_owned);
+        }
+        if self.linked_etag.is_none() {
+            self.linked_etag = string(X_LINKED_ETAG);
+        }
+        if self.linked_size.is_none() {
+            self.linked_size = string(X_LINKED_SIZE).and_then(|v| v.parse().ok());
+        }
+        // The Hub sends `X-Linked-Etag` for every file -- a git blob SHA-1 for
+        // an ordinary one, an LFS oid for an LFS-backed one -- and sends
+        // `X-Linked-Size` only for the latter. That header is therefore the
+        // signal, not the linked etag's presence.
+        self.lfs = self.lfs || self.linked_size.is_some();
     }
 }
 
@@ -233,13 +356,8 @@ impl Upstream for HubClient {
     ) -> Result<UpstreamHead> {
         let url = self.resolve_url(repo_type, repo_id, revision, path);
         let what = format!("HEAD {repo_id}@{revision}/{path}");
-        let response = self
-            .authorize(self.client.head(&url))
-            .send()
-            .await
-            .map_err(|e| Error::Upstream(format!("{what}: {e}")))?;
-        let response = self.check(response, &what).await?;
-        Ok(head_of(&response))
+        let (head, _) = self.send(Method::HEAD, &url, &what).await?;
+        Ok(head)
     }
 
     async fn get_file(
@@ -251,13 +369,7 @@ impl Upstream for HubClient {
     ) -> Result<UpstreamBody> {
         let url = self.resolve_url(repo_type, repo_id, revision, path);
         let what = format!("GET {repo_id}@{revision}/{path}");
-        let response = self
-            .authorize(self.client.get(&url))
-            .send()
-            .await
-            .map_err(|e| Error::Upstream(format!("{what}: {e}")))?;
-        let response = self.check(response, &what).await?;
-        let head = head_of(&response);
+        let (head, response) = self.send(Method::GET, &url, &what).await?;
         debug!(%url, size = ?head.content_length, lfs = head.lfs, "upstream transfer starting");
 
         let stream = response
@@ -407,6 +519,36 @@ mod tests {
         assert!(url.ends_with("/a%20b/c.json"), "{url}");
     }
 
+    #[test]
+    fn an_absolute_redirect_replaces_the_url_and_a_relative_one_extends_it() {
+        let base = "https://huggingface.co/gpt2/resolve/main/tf_model.h5";
+        assert_eq!(
+            resolve_location(base, "https://cdn.hf.co/blob?sig=x").unwrap(),
+            "https://cdn.hf.co/blob?sig=x"
+        );
+        assert_eq!(
+            resolve_location(base, "/gpt2/resolve/abc123/tf_model.h5").unwrap(),
+            "https://huggingface.co/gpt2/resolve/abc123/tf_model.h5"
+        );
+        assert_eq!(
+            resolve_location(base, "other.h5").unwrap(),
+            "https://huggingface.co/gpt2/resolve/main/other.h5"
+        );
+    }
+
+    #[test]
+    fn a_cdn_redirect_is_recognised_as_a_different_origin() {
+        assert_eq!(
+            origin_of("https://huggingface.co/gpt2/resolve/main/a.bin"),
+            "https://huggingface.co"
+        );
+        assert_ne!(
+            origin_of("https://us.aws.cdn.hf.co/xet-bridge-us/abc"),
+            origin_of("https://huggingface.co/gpt2"),
+            "the token must not follow a redirect off the Hub"
+        );
+    }
+
     #[test]
     fn a_sealed_upstream_refuses_without_reaching_the_network() {
         let sealed = SealedUpstream::new("https://huggingface.co");