Skip to main content

hf_fetch_model/
http_range.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! HTTP Range `Read + Seek` substrate for remote inspection.
4//!
5//! [`RangeReader`] adapts a byte-range transport into [`std::io::Read`] +
6//! [`std::io::Seek`], so format parsers that consume a reader (e.g.
7//! `anamnesis::inspect_npz_from_reader`) can inspect a remote file's
8//! metadata without downloading it. The transport is abstracted behind
9//! [`RangeFetcher`]; production code uses [`HttpRangeFetcher`] (HTTP Range
10//! requests against the `HuggingFace` CDN), and unit tests use an in-memory
11//! fetcher — the buffering, budgeting, and `Seek` semantics are exercised
12//! entirely offline.
13//!
14//! # Access pattern optimisations
15//!
16//! Tuned for archive-directory reads (`ZIP` central directory + per-entry
17//! headers — the `NPZ` / `PTH` layout):
18//!
19//! - **Tail prefetch** — the first read landing in the last
20//!   [`TAIL_PREFETCH_BYTES`] of the file fetches that whole tail once and
21//!   caches it. The `ZIP` end-of-central-directory scan and the central
22//!   directory itself are then served from memory.
23//! - **Read-ahead** — short sequential reads are coalesced into
24//!   [`READAHEAD_BYTES`]-sized window fetches, so parsing a 30-byte local
25//!   header followed by a ~120-byte `NPY` header costs one request, not two.
26//! - **No re-fetch** — reads covered by the cached tail or the current
27//!   window never re-issue a request.
28//!
29//! # Safety budgets
30//!
31//! The reader enforces two hard caps — [`MAX_RANGE_REQUESTS`] requests and
32//! [`MAX_TRANSFER_BUDGET`] bytes fetched — so a pathological or adversarial
33//! archive layout (a seek storm, an absurdly large central directory) fails
34//! fast with a clear error instead of degenerating into a full download.
35//! [`RangeReader::with_limits`] overrides the defaults for callers with
36//! different budgets.
37//!
38//! # Error channel
39//!
40//! `Read` / `Seek` must return [`std::io::Error`], which flattens the crate's
41//! typed [`FetchError`]. The reader therefore stores the most recent fetch
42//! failure; after a failed parse, [`RangeReader::take_last_error`] recovers
43//! the typed error (e.g. an HTTP `403` that the CLI upgrades into a
44//! gated-repo diagnosis).
45
46use std::io::{self, Read, Seek, SeekFrom};
47use std::time::Duration;
48
49use serde::Serialize;
50
51use crate::chunked;
52use crate::error::FetchError;
53
54// -----------------------------------------------------------------------
55// Tuning constants
56// -----------------------------------------------------------------------
57
58/// Size of the cached file tail fetched by the first read near end-of-file.
59///
60/// 64 KiB covers the `ZIP` end-of-central-directory record (max comment
61/// length 65 535 bytes) plus the central directory of typical `NPZ`
62/// archives, so the whole directory scan costs one request.
63pub const TAIL_PREFETCH_BYTES: u64 = 64 * 1024;
64
65/// Minimum window size for a fetch serving a sequential read.
66///
67/// Short header-sized reads (tens of bytes) are widened to this size so
68/// adjacent structures (`ZIP` local header + `NPY` header) arrive in one
69/// request.
70pub const READAHEAD_BYTES: u64 = 4 * 1024;
71
72/// Default cap on total bytes fetched over the reader's lifetime.
73///
74/// Remote inspection of a well-formed archive transfers well under 1 MiB;
75/// 32 MiB leaves two orders of magnitude of headroom (e.g. a many-thousand
76/// entry central directory) while guaranteeing that no inspect silently
77/// degenerates into a full-file download.
78pub const MAX_TRANSFER_BUDGET: u64 = 32 * 1024 * 1024;
79
80/// Default cap on the number of range fetches over the reader's lifetime.
81///
82/// A well-formed inspect needs a handful of requests; 256 tolerates a
83/// fragmented directory layout while stopping request storms from
84/// adversarial seek patterns.
85pub const MAX_RANGE_REQUESTS: u32 = 256;
86
87/// Wall-clock budget for a single HTTP range request (headers + body).
88///
89/// Bounds a stalled response body — the connect phase is separately bounded
90/// by the client's TCP connect timeout.
91const RANGE_REQUEST_TIMEOUT: Duration = Duration::from_secs(60);
92
93// -----------------------------------------------------------------------
94// Transport abstraction
95// -----------------------------------------------------------------------
96
97/// A transport that serves byte ranges of one remote file.
98///
99/// Implemented by [`HttpRangeFetcher`] for production and by in-memory
100/// fetchers in tests. The contract: `fetch(start, end_inclusive)` returns
101/// exactly `end_inclusive - start + 1` bytes of the file's content at
102/// `start`, and `total_size` is the file's full length in bytes, known
103/// up front.
104pub trait RangeFetcher {
105    /// Fetches the inclusive byte range `start..=end_inclusive`.
106    ///
107    /// # Errors
108    ///
109    /// Returns [`FetchError::Http`] when the transport fails (network
110    /// error, unexpected HTTP status, validation mismatch).
111    fn fetch(&mut self, start: u64, end_inclusive: u64) -> Result<Vec<u8>, FetchError>;
112
113    /// Total size of the remote file in bytes.
114    fn total_size(&self) -> u64;
115
116    /// Requests spent by the transport outside [`RangeFetcher::fetch`]
117    /// (e.g. the eager probe that resolved the file size and classified
118    /// access). Included in [`RangeReader::stats`] so the reported request
119    /// count is honest.
120    fn extra_requests(&self) -> u32 {
121        0
122    }
123}
124
125/// Transfer statistics for one [`RangeReader`] lifetime.
126///
127/// Rendered by the CLI as provenance (e.g. `remote (6 range requests,
128/// 136.0 KiB fetched)` — the live-measured cost of inspecting a 72 MiB
129/// `NPZ`) — the on-screen proof that an inspect read metadata, not weights.
130#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
131#[non_exhaustive]
132pub struct RangeStats {
133    /// Total HTTP requests issued: range fetches (attempts, including
134    /// failed ones) plus the transport's probe requests.
135    pub requests: u32,
136    /// Total content bytes fetched across all successful range requests.
137    /// Excludes HTTP header overhead and probe responses (single-byte
138    /// probes; negligible).
139    pub bytes_fetched: u64,
140}
141
142// -----------------------------------------------------------------------
143// RangeReader core
144// -----------------------------------------------------------------------
145
146/// A cached contiguous extent of the remote file.
147struct Extent {
148    /// Absolute file offset of `data[0]`.
149    start: u64,
150    /// The fetched bytes.
151    data: Vec<u8>,
152}
153
154impl Extent {
155    /// One past the last absolute offset covered by this extent.
156    fn end(&self) -> u64 {
157        // CAST: usize → u64, buffer length always fits (usize ≤ 64 bits)
158        #[allow(clippy::as_conversions)]
159        self.start.saturating_add(self.data.len() as u64)
160    }
161
162    /// Whether `pos` falls inside this extent.
163    fn contains(&self, pos: u64) -> bool {
164        pos >= self.start && pos < self.end()
165    }
166
167    /// Copies bytes starting at absolute offset `pos` into `buf`, returning
168    /// the count copied (`None` if `pos` is outside the extent).
169    fn copy_at(&self, pos: u64, buf: &mut [u8]) -> Option<usize> {
170        if !self.contains(pos) {
171            return None;
172        }
173        let offset = usize::try_from(pos.checked_sub(self.start)?).ok()?;
174        let available = self.data.len().checked_sub(offset)?;
175        let n = buf.len().min(available);
176        let src = self.data.get(offset..offset.checked_add(n)?)?;
177        let dst = buf.get_mut(..n)?;
178        dst.copy_from_slice(src);
179        Some(n)
180    }
181}
182
183/// `Read + Seek` over a [`RangeFetcher`], with tail caching, read-ahead,
184/// and hard safety budgets.
185///
186/// See the [module docs](self) for the access-pattern and budget design.
187pub struct RangeReader<F: RangeFetcher> {
188    /// The transport serving byte ranges.
189    fetcher: F,
190    /// Current logical read position (may exceed the file size after a
191    /// permissive seek; reads there return `Ok(0)`).
192    pos: u64,
193    /// Most recent read-ahead window.
194    window: Option<Extent>,
195    /// Cached file tail (`ZIP` end-of-central-directory region).
196    tail: Option<Extent>,
197    /// Range fetches attempted (successful or failed).
198    requests: u32,
199    /// Content bytes fetched across successful range fetches.
200    bytes_fetched: u64,
201    /// Lifetime cap on `requests`.
202    max_requests: u32,
203    /// Lifetime cap on `bytes_fetched`.
204    max_transfer_bytes: u64,
205    /// Most recent typed transport error (see [`Self::take_last_error`]).
206    last_error: Option<FetchError>,
207}
208
209impl<F: RangeFetcher> RangeReader<F> {
210    /// Creates a reader with the default safety budgets
211    /// ([`MAX_RANGE_REQUESTS`], [`MAX_TRANSFER_BUDGET`]).
212    #[must_use]
213    pub const fn new(fetcher: F) -> Self {
214        Self::with_limits(fetcher, MAX_RANGE_REQUESTS, MAX_TRANSFER_BUDGET)
215    }
216
217    /// Creates a reader with explicit safety budgets.
218    ///
219    /// `max_requests` caps the number of range fetches; `max_transfer_bytes`
220    /// caps the total content bytes fetched. Exceeding either fails the
221    /// offending read with a [`std::io::Error`] naming the cap.
222    #[must_use]
223    pub const fn with_limits(fetcher: F, max_requests: u32, max_transfer_bytes: u64) -> Self {
224        Self {
225            fetcher,
226            pos: 0,
227            window: None,
228            tail: None,
229            requests: 0,
230            bytes_fetched: 0,
231            max_requests,
232            max_transfer_bytes,
233            last_error: None,
234        }
235    }
236
237    /// Transfer statistics so far (range fetches + transport probes).
238    #[must_use]
239    pub fn stats(&self) -> RangeStats {
240        RangeStats {
241            requests: self.requests.saturating_add(self.fetcher.extra_requests()),
242            bytes_fetched: self.bytes_fetched,
243        }
244    }
245
246    /// Total size of the remote file in bytes.
247    #[must_use]
248    pub fn total_size(&self) -> u64 {
249        self.fetcher.total_size()
250    }
251
252    /// Takes the most recent typed transport error, if any.
253    ///
254    /// `Read` flattens [`FetchError`] into [`std::io::Error`] strings; after
255    /// a parse fails, callers use this to recover the typed error (e.g. to
256    /// let the CLI upgrade an HTTP `403` into a gated-repo diagnosis).
257    #[must_use]
258    pub const fn take_last_error(&mut self) -> Option<FetchError> {
259        self.last_error.take()
260    }
261
262    /// Fetches `start..=end_inclusive` through the budget checks, recording
263    /// stats and stashing typed errors.
264    fn checked_fetch(&mut self, start: u64, end_inclusive: u64) -> io::Result<Vec<u8>> {
265        let len = end_inclusive
266            .checked_sub(start)
267            .and_then(|d| d.checked_add(1))
268            .ok_or_else(|| {
269                io::Error::new(
270                    io::ErrorKind::InvalidInput,
271                    format!("invalid range {start}..={end_inclusive}"),
272                )
273            })?;
274
275        if self.requests >= self.max_requests {
276            return Err(io::Error::other(format!(
277                "range request cap exceeded ({} requests): pathological archive \
278                 layout or seek storm; refusing further fetches",
279                self.max_requests
280            )));
281        }
282        if self.bytes_fetched.saturating_add(len) > self.max_transfer_bytes {
283            return Err(io::Error::other(format!(
284                "range transfer budget exceeded ({} bytes fetched, {len} more \
285                 requested, cap {}): metadata inspection should never read \
286                 this much; refusing further fetches",
287                self.bytes_fetched, self.max_transfer_bytes
288            )));
289        }
290
291        self.requests = self.requests.saturating_add(1);
292        match self.fetcher.fetch(start, end_inclusive) {
293            Ok(data) => {
294                // CAST: usize → u64, buffer length always fits (usize ≤ 64 bits)
295                #[allow(clippy::as_conversions)]
296                let got = data.len() as u64;
297                if got != len {
298                    return Err(io::Error::other(format!(
299                        "range fetcher returned {got} bytes for a {len}-byte \
300                         range ({start}..={end_inclusive})"
301                    )));
302                }
303                self.bytes_fetched = self.bytes_fetched.saturating_add(len);
304                Ok(data)
305            }
306            Err(fetch_err) => {
307                let io_err = io::Error::other(fetch_err.to_string());
308                self.last_error = Some(fetch_err);
309                Err(io_err)
310            }
311        }
312    }
313
314    /// Serves `buf` from the cached tail or window, if `pos` is covered.
315    fn copy_cached(&self, buf: &mut [u8]) -> Option<usize> {
316        if let Some(w) = &self.window
317            && let Some(n) = w.copy_at(self.pos, buf)
318        {
319            return Some(n);
320        }
321        if let Some(t) = &self.tail
322            && let Some(n) = t.copy_at(self.pos, buf)
323        {
324            return Some(n);
325        }
326        None
327    }
328
329    /// Advances the read position by `n` copied bytes.
330    fn advance_pos(&mut self, n: usize) {
331        // CAST: usize → u64, copy count bounded by buf length
332        #[allow(clippy::as_conversions)]
333        {
334            self.pos = self.pos.saturating_add(n as u64);
335        }
336    }
337
338    /// Copies from cache immediately after a fetch that must cover `pos`.
339    ///
340    /// A miss here means the fetch/cache bookkeeping is internally
341    /// inconsistent; surfaced as an error rather than a panic.
342    fn serve_from_cache_after_fetch(&mut self, buf: &mut [u8]) -> io::Result<usize> {
343        let n = self.copy_cached(buf).ok_or_else(|| {
344            io::Error::other(format!(
345                "internal range cache inconsistency at offset {}",
346                self.pos
347            ))
348        })?;
349        self.advance_pos(n);
350        Ok(n)
351    }
352}
353
354impl<F: RangeFetcher> Read for RangeReader<F> {
355    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
356        if buf.is_empty() {
357            return Ok(0);
358        }
359        let total = self.fetcher.total_size();
360        if self.pos >= total {
361            return Ok(0); // EOF (also covers permissive past-end seeks)
362        }
363
364        // 1. Serve from cached extents (window, then tail) — no request.
365        if let Some(n) = self.copy_cached(buf) {
366            self.advance_pos(n);
367            return Ok(n);
368        }
369
370        // 2. First read near end-of-file: prefetch and cache the tail.
371        let tail_start = total.saturating_sub(TAIL_PREFETCH_BYTES);
372        if self.tail.is_none() && self.pos >= tail_start {
373            let data = self.checked_fetch(tail_start, total.saturating_sub(1))?;
374            self.tail = Some(Extent {
375                start: tail_start,
376                data,
377            });
378            return self.serve_from_cache_after_fetch(buf);
379        }
380
381        // 3. Read-ahead window fetch at the current position.
382        // CAST: usize → u64, buffer length always fits (usize ≤ 64 bits)
383        #[allow(clippy::as_conversions)]
384        let want = (buf.len() as u64).max(READAHEAD_BYTES);
385        let mut end = self
386            .pos
387            .saturating_add(want)
388            .saturating_sub(1)
389            .min(total.saturating_sub(1));
390        // Never overlap the cached tail — those bytes are already paid for.
391        if let Some(t) = &self.tail
392            && self.pos < t.start
393        {
394            end = end.min(t.start.saturating_sub(1));
395        }
396        let data = self.checked_fetch(self.pos, end)?;
397        self.window = Some(Extent {
398            start: self.pos,
399            data,
400        });
401        self.serve_from_cache_after_fetch(buf)
402    }
403}
404
405impl<F: RangeFetcher> Seek for RangeReader<F> {
406    fn seek(&mut self, pos: SeekFrom) -> io::Result<u64> {
407        let total = self.fetcher.total_size();
408        let target: i128 = match pos {
409            SeekFrom::Start(p) => i128::from(p),
410            SeekFrom::End(delta) => i128::from(total).saturating_add(i128::from(delta)),
411            SeekFrom::Current(delta) => i128::from(self.pos).saturating_add(i128::from(delta)),
412        };
413        if target < 0 {
414            return Err(io::Error::new(
415                io::ErrorKind::InvalidInput,
416                format!("seek to negative offset {target}"),
417            ));
418        }
419        let new_pos = u64::try_from(target).map_err(|_| {
420            io::Error::new(
421                io::ErrorKind::InvalidInput,
422                format!("seek offset {target} exceeds u64::MAX"),
423            )
424        })?;
425        self.pos = new_pos;
426        Ok(new_pos)
427    }
428}
429
430// -----------------------------------------------------------------------
431// HTTP transport
432// -----------------------------------------------------------------------
433
434/// [`RangeReader`] over [`HttpRangeFetcher`] — the production remote-inspect
435/// substrate.
436pub type HttpRangeReader = RangeReader<HttpRangeFetcher>;
437
438/// HTTP Range transport against a `HuggingFace`-hosted file.
439///
440/// Opened via [`HttpRangeReader::open`]. One authenticated, redirect-free
441/// probe (reusing the chunked-download probe) resolves the file size and
442/// classifies access failures eagerly (gated repos, no Range support).
443/// **Every range request then targets the HF `/resolve` URL** and follows
444/// the redirect to a freshly-signed CDN URL; `reqwest` strips the
445/// `Authorization` header on the cross-host redirect, so the `Bearer`
446/// token is never sent to the CDN host.
447///
448/// Per-request re-resolution is **required**, not just convenient:
449/// Xet-backed repos sign each CDN URL for the *exact* `Range` header that
450/// minted it (a `ByteRange` policy condition in the signed URL), so a CDN
451/// URL stored from the probe can never serve any other range. Re-resolving
452/// also makes signed-URL expiry a non-issue — every request gets a fresh
453/// signature. Discovered live against `google/gemma-scope-2b-pt-res`
454/// (v0.11.0 dogfooding).
455///
456/// # Response validation (every request)
457///
458/// - Status must be `206 Partial Content`. A `200` (server ignoring the
459///   `Range` header) is rejected **without reading the body** — the guard
460///   against silently pulling the full file.
461/// - `Content-Range` must match the requested range and the probed total
462///   size.
463/// - The body is streamed with a hard cap of the requested length;
464///   over- or under-delivery is an error.
465/// - The response `ETag` (when present) must stay identical across
466///   requests — a mid-inspect change upstream aborts with a clear error.
467pub struct HttpRangeFetcher {
468    /// Runtime handle used to drive async `reqwest` calls from the
469    /// blocking `Read` context (`spawn_blocking` thread).
470    handle: tokio::runtime::Handle,
471    /// Authenticated, redirect-following client. The auth header rides
472    /// only to the HF host — `reqwest` removes sensitive headers when a
473    /// redirect crosses hosts.
474    client: reqwest::Client,
475    /// The `HF` `/resolve` URL every range request targets.
476    hf_url: String,
477    /// Filename for error messages.
478    filename: String,
479    /// First `ETag` observed on a range response — later responses must
480    /// match it (self-consistency across requests).
481    response_etag: Option<String>,
482    /// The etag observed at probe time, before any range request — the
483    /// same value `chunked::probe_range_support` uses for the blob path in
484    /// the `hf-hub` cache. Read by `inspect --cache-headers` to key its
485    /// header cache without an extra round trip; distinct from
486    /// `response_etag`, which validates self-consistency across the range
487    /// requests this reader itself issues.
488    probe_etag: String,
489    /// Total file size from the probe's `Content-Range`.
490    total_size: u64,
491    /// Probe requests spent (no-redirect probe + CDN size fetch).
492    extra: u32,
493}
494
495impl HttpRangeReader {
496    /// Opens a range reader over `filename` in `repo_id` at `revision`
497    /// (default `main`).
498    ///
499    /// Performs the probe eagerly, so failures (missing file, gated repo,
500    /// no Range support) surface here as typed errors rather than
501    /// mid-parse `io::Error`s.
502    ///
503    /// Must be called from within a `tokio` runtime; the returned reader is
504    /// then handed to a blocking context (`tokio::task::spawn_blocking`),
505    /// where its `Read` / `Seek` impls drive async requests via the
506    /// captured runtime handle.
507    ///
508    /// # Errors
509    ///
510    /// Returns [`FetchError::Http`] if the probe fails, the repository or
511    /// file is inaccessible (including gated repos: `returned status
512    /// 401/403` errors, which the CLI upgrades into a gated-repo
513    /// diagnosis), or the server does not support Range requests.
514    pub async fn open(
515        repo_id: &str,
516        revision: Option<&str>,
517        filename: &str,
518        token: Option<&str>,
519    ) -> Result<Self, FetchError> {
520        Self::open_with_limits(
521            repo_id,
522            revision,
523            filename,
524            token,
525            MAX_RANGE_REQUESTS,
526            MAX_TRANSFER_BUDGET,
527        )
528        .await
529    }
530
531    /// Opens a range reader with explicit safety budgets (see
532    /// [`RangeReader::with_limits`]) instead of the [`MAX_RANGE_REQUESTS`] /
533    /// [`MAX_TRANSFER_BUDGET`] defaults [`HttpRangeReader::open`] uses.
534    ///
535    /// The defaults are tuned for archive-header parsing (`inspect`'s
536    /// `.safetensors` / `.gguf` / `.npz` / `.pth` paths) — a handful of
537    /// requests, well under 1 MiB. A caller with its own, larger,
538    /// user-facing byte budget (`hf-fm peek`'s `--max`) needs its transport
539    /// budget sized to match, or safety limits tuned for header parsing cut
540    /// it off first with a misleading "pathological archive layout" error.
541    ///
542    /// Same preconditions as [`HttpRangeReader::open`] (must be called from
543    /// within a `tokio` runtime; the reader is then driven from a blocking
544    /// context).
545    ///
546    /// # Errors
547    ///
548    /// Same as [`HttpRangeReader::open`].
549    pub async fn open_with_limits(
550        repo_id: &str,
551        revision: Option<&str>,
552        filename: &str,
553        token: Option<&str>,
554        max_requests: u32,
555        max_transfer_bytes: u64,
556    ) -> Result<Self, FetchError> {
557        let fetcher = HttpRangeFetcher::open(repo_id, revision, filename, token).await?;
558        Ok(RangeReader::with_limits(
559            fetcher,
560            max_requests,
561            max_transfer_bytes,
562        ))
563    }
564
565    /// The etag observed at probe time, before any range request.
566    ///
567    /// Lets a caller key a cache entry (`inspect --cache-headers`) on the
568    /// same etag `hf-hub`'s own blob layout uses, without an extra round
569    /// trip beyond the probe [`HttpRangeReader::open`] already made.
570    #[must_use]
571    pub fn probe_etag(&self) -> &str {
572        self.fetcher.probe_etag()
573    }
574}
575
576impl HttpRangeFetcher {
577    /// The etag observed at probe time, before any range request.
578    #[must_use]
579    fn probe_etag(&self) -> &str {
580        // BORROW: explicit .as_str() instead of Deref coercion
581        self.probe_etag.as_str()
582    }
583
584    /// Probes `filename` and constructs the transport (see
585    /// [`HttpRangeReader::open`]).
586    ///
587    /// # Errors
588    ///
589    /// Returns [`FetchError::Http`] if the probe fails, the file is
590    /// inaccessible, or the server does not support Range requests.
591    async fn open(
592        repo_id: &str,
593        revision: Option<&str>,
594        filename: &str,
595        token: Option<&str>,
596    ) -> Result<Self, FetchError> {
597        let rev = revision.unwrap_or("main");
598        let hf_url = chunked::build_download_url(repo_id, rev, filename);
599        let client = chunked::build_client(token)?;
600
601        let info = chunked::probe_range_support(
602            client.clone(),
603            hf_url.clone(),
604            // BORROW: explicit String::from for Option<&str> → Option<String>
605            token.map(String::from),
606        )
607        .await?;
608        let Some(info) = info else {
609            return Err(Self::classify_no_range_support(&client, &hf_url, filename).await);
610        };
611
612        Ok(Self {
613            handle: tokio::runtime::Handle::current(),
614            client,
615            hf_url,
616            // BORROW: explicit .to_owned() for the owned field
617            filename: filename.to_owned(),
618            response_etag: None,
619            probe_etag: info.etag,
620            total_size: info.content_length,
621            extra: 2, // no-redirect probe + CDN size fetch
622        })
623    }
624
625    /// Distinguishes "no Range support" from an access failure after the
626    /// probe declined.
627    ///
628    /// The chunked probe returns `None` for **any** non-redirect,
629    /// non-`206` response — including a gated repo's `401`/`403`. One
630    /// follow-up request recovers the real status so gated repos get the
631    /// actionable diagnosis instead of a misleading "no Range support".
632    async fn classify_no_range_support(
633        client: &reqwest::Client,
634        url: &str,
635        filename: &str,
636    ) -> FetchError {
637        let result = client
638            .get(url)
639            .header(reqwest::header::RANGE, "bytes=0-0")
640            .timeout(RANGE_REQUEST_TIMEOUT)
641            .send()
642            .await;
643        match result {
644            Ok(resp) => {
645                let status = resp.status();
646                if status.is_client_error() || status.is_server_error() {
647                    FetchError::Http(format!(
648                        "Range request for {filename} returned status {status}"
649                    ))
650                } else {
651                    FetchError::Http(format!(
652                        "server does not support Range requests for {filename}"
653                    ))
654                }
655            }
656            Err(e) => FetchError::Http(format!("failed to probe {filename}: {e}")),
657        }
658    }
659
660    /// Issues one validated range request against the `/resolve` URL,
661    /// following the redirect to a freshly-signed CDN URL.
662    ///
663    /// Returns the body plus the response `ETag` (cleaned), which the
664    /// caller records for cross-request consistency.
665    fn fetch_once(
666        &self,
667        start: u64,
668        end_inclusive: u64,
669    ) -> Result<(Vec<u8>, Option<String>), FetchError> {
670        let expected_len = end_inclusive
671            .checked_sub(start)
672            .and_then(|d| d.checked_add(1))
673            .ok_or_else(|| {
674                FetchError::Http(format!(
675                    "invalid range {start}..={end_inclusive} for {}",
676                    self.filename
677                ))
678            })?;
679        let expected_usize = usize::try_from(expected_len).map_err(|_| {
680            FetchError::Http(format!(
681                "range length {expected_len} exceeds addressable memory for {}",
682                self.filename
683            ))
684        })?;
685
686        let range_value = format!("bytes={start}-{end_inclusive}");
687        // BORROW: explicit .as_str() instead of Deref coercion
688        let filename = self.filename.as_str();
689        let total_size = self.total_size;
690
691        self.handle.block_on(async {
692            let resp = self
693                .client
694                .get(self.hf_url.as_str())
695                // BORROW: explicit .as_str() instead of Deref coercion
696                .header(reqwest::header::RANGE, range_value.as_str())
697                .timeout(RANGE_REQUEST_TIMEOUT)
698                .send()
699                .await
700                .map_err(|e| {
701                    FetchError::Http(format!("failed to send Range request for {filename}: {e}"))
702                })?;
703
704            let status = resp.status();
705            if status == reqwest::StatusCode::OK {
706                // Body deliberately not read: a 200 means the server ignored
707                // the Range header and is offering the FULL file.
708                return Err(FetchError::Http(format!(
709                    "server ignored the Range header for {filename} (status 200 \
710                     for bytes={start}-{end_inclusive}); refusing to read the full file"
711                )));
712            }
713            if status != reqwest::StatusCode::PARTIAL_CONTENT {
714                return Err(FetchError::Http(format!(
715                    "Range request for {filename} returned status {status}"
716                )));
717            }
718
719            let content_range = resp
720                .headers()
721                .get(reqwest::header::CONTENT_RANGE)
722                .and_then(|v| v.to_str().ok())
723                // BORROW: explicit .to_owned() — header value outlives the response borrow
724                .map(str::to_owned)
725                .ok_or_else(|| {
726                    FetchError::Http(format!("missing Content-Range header for {filename}"))
727                })?;
728            let (cr_start, cr_end, cr_total) = parse_content_range(content_range.as_str())
729                .ok_or_else(|| {
730                    FetchError::Http(format!(
731                        "invalid Content-Range header for {filename}: {content_range}"
732                    ))
733                })?;
734            if cr_start != start || cr_end != end_inclusive || cr_total != total_size {
735                return Err(FetchError::Http(format!(
736                    "Content-Range mismatch for {filename}: requested \
737                     bytes={start}-{end_inclusive} of {total_size}, server answered {content_range}"
738                )));
739            }
740
741            let etag = resp
742                .headers()
743                .get(reqwest::header::ETAG)
744                .and_then(|v| v.to_str().ok())
745                .map(clean_etag);
746
747            // Stream the body with a hard cap of the requested length.
748            let mut data: Vec<u8> = Vec::with_capacity(expected_usize);
749            let mut resp = resp;
750            while let Some(chunk) = resp.chunk().await.map_err(|e| {
751                FetchError::Http(format!("failed to read Range response for {filename}: {e}"))
752            })? {
753                if data.len().saturating_add(chunk.len()) > expected_usize {
754                    return Err(FetchError::Http(format!(
755                        "server sent more than the requested {expected_len} bytes \
756                         for {filename} (bytes={start}-{end_inclusive}); aborting"
757                    )));
758                }
759                data.extend_from_slice(&chunk);
760            }
761            if data.len() != expected_usize {
762                return Err(FetchError::Http(format!(
763                    "server returned {} bytes for a {expected_len}-byte range \
764                     of {filename} (bytes={start}-{end_inclusive})",
765                    data.len()
766                )));
767            }
768
769            Ok((data, etag))
770        })
771    }
772
773    /// Records / checks the response `ETag` for cross-request consistency.
774    fn check_response_etag(&mut self, etag: Option<String>) -> Result<(), FetchError> {
775        if let Some(current) = etag {
776            match &self.response_etag {
777                Some(previous) if *previous != current => {
778                    return Err(FetchError::Http(format!(
779                        "{} changed upstream during inspect (etag {previous} \
780                         became {current})",
781                        self.filename
782                    )));
783                }
784                Some(_) => {} // EXPLICIT: etag unchanged — nothing to record
785                None => self.response_etag = Some(current),
786            }
787        }
788        Ok(())
789    }
790}
791
792impl RangeFetcher for HttpRangeFetcher {
793    fn fetch(&mut self, start: u64, end_inclusive: u64) -> Result<Vec<u8>, FetchError> {
794        // Each fetch re-resolves through /resolve (fresh signed CDN URL),
795        // so there is no stored-signature expiry to manage; failures
796        // surface directly with their real HTTP status.
797        let (data, etag) = self.fetch_once(start, end_inclusive)?;
798        self.check_response_etag(etag)?;
799        Ok(data)
800    }
801
802    fn total_size(&self) -> u64 {
803        self.total_size
804    }
805
806    fn extra_requests(&self) -> u32 {
807        self.extra
808    }
809}
810
811/// Parses a `Content-Range: bytes S-E/T` value into `(S, E, T)`.
812///
813/// Returns `None` on any deviation from that exact form (including the
814/// `bytes */T` unsatisfied-range form, which is never valid for a `206`).
815fn parse_content_range(value: &str) -> Option<(u64, u64, u64)> {
816    let rest = value.strip_prefix("bytes ")?;
817    let (range, total) = rest.split_once('/')?;
818    let (start, end) = range.split_once('-')?;
819    Some((
820        start.trim().parse().ok()?,
821        end.trim().parse().ok()?,
822        total.trim().parse().ok()?,
823    ))
824}
825
826/// Normalises an `ETag` value: strips the weak-validator prefix and quotes.
827///
828/// Matches the probe's normalisation (`etag.replace('"', "")`), so probe
829/// and response etags compare in the same representation.
830fn clean_etag(raw: &str) -> String {
831    raw.strip_prefix("W/").unwrap_or(raw).replace('"', "")
832}
833
834// -----------------------------------------------------------------------
835// Tests
836// -----------------------------------------------------------------------
837
838#[cfg(test)]
839mod tests {
840    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::indexing_slicing)]
841
842    use super::*;
843
844    /// In-memory fetcher over a byte vector, logging every fetched range.
845    struct InMemoryFetcher {
846        data: Vec<u8>,
847        calls: Vec<(u64, u64)>,
848    }
849
850    impl InMemoryFetcher {
851        fn new(data: Vec<u8>) -> Self {
852            Self {
853                data,
854                calls: Vec::new(),
855            }
856        }
857    }
858
859    impl RangeFetcher for InMemoryFetcher {
860        fn fetch(&mut self, start: u64, end_inclusive: u64) -> Result<Vec<u8>, FetchError> {
861            self.calls.push((start, end_inclusive));
862            let s = usize::try_from(start).unwrap();
863            let e = usize::try_from(end_inclusive).unwrap();
864            self.data
865                .get(s..=e)
866                .map(<[u8]>::to_vec)
867                .ok_or_else(|| FetchError::Http(format!("bad range {start}..={end_inclusive}")))
868        }
869
870        fn total_size(&self) -> u64 {
871            u64::try_from(self.data.len()).unwrap()
872        }
873    }
874
875    /// A fetcher that always fails with an HTTP-status-shaped error.
876    struct FailingFetcher {
877        size: u64,
878    }
879
880    impl RangeFetcher for FailingFetcher {
881        fn fetch(&mut self, _start: u64, _end_inclusive: u64) -> Result<Vec<u8>, FetchError> {
882            Err(FetchError::Http(
883                "Range request for x.npz returned status 403 Forbidden".to_owned(),
884            ))
885        }
886
887        fn total_size(&self) -> u64 {
888            self.size
889        }
890    }
891
892    /// A fetcher that returns fewer bytes than requested.
893    struct ShortFetcher {
894        size: u64,
895    }
896
897    impl RangeFetcher for ShortFetcher {
898        fn fetch(&mut self, _start: u64, _end_inclusive: u64) -> Result<Vec<u8>, FetchError> {
899            Ok(vec![0u8; 1])
900        }
901
902        fn total_size(&self) -> u64 {
903            self.size
904        }
905    }
906
907    fn sample_data(len: usize) -> Vec<u8> {
908        // CAST: usize → u8, deliberate wrapping for a recognisable pattern
909        #[allow(clippy::as_conversions, clippy::cast_possible_truncation)]
910        (0..len).map(|i| (i % 251) as u8).collect()
911    }
912
913    // ---------- Seek semantics ----------
914
915    #[test]
916    fn seek_start_end_current_semantics() {
917        let mut r = RangeReader::new(InMemoryFetcher::new(sample_data(1000)));
918        assert_eq!(r.seek(SeekFrom::Start(10)).unwrap(), 10);
919        assert_eq!(r.seek(SeekFrom::Current(5)).unwrap(), 15);
920        assert_eq!(r.seek(SeekFrom::Current(-15)).unwrap(), 0);
921        assert_eq!(r.seek(SeekFrom::End(0)).unwrap(), 1000);
922        assert_eq!(r.seek(SeekFrom::End(-1000)).unwrap(), 0);
923        // Past-EOF seek is permitted; reads there return 0.
924        assert_eq!(r.seek(SeekFrom::End(50)).unwrap(), 1050);
925        let mut buf = [0u8; 4];
926        assert_eq!(r.read(&mut buf).unwrap(), 0);
927    }
928
929    #[test]
930    fn seek_negative_is_invalid_input() {
931        let mut r = RangeReader::new(InMemoryFetcher::new(sample_data(100)));
932        let err = r.seek(SeekFrom::Current(-1)).unwrap_err();
933        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
934        let err = r.seek(SeekFrom::End(-101)).unwrap_err();
935        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
936    }
937
938    #[test]
939    fn empty_file_reads_zero_with_no_requests() {
940        let mut r = RangeReader::new(InMemoryFetcher::new(Vec::new()));
941        let mut buf = [0u8; 8];
942        assert_eq!(r.read(&mut buf).unwrap(), 0);
943        assert_eq!(r.stats().requests, 0);
944    }
945
946    // ---------- Read-ahead and caching ----------
947
948    #[test]
949    fn sequential_small_reads_coalesce_into_one_window_fetch() {
950        // 100 KiB file, reads far from the tail region.
951        let mut r = RangeReader::new(InMemoryFetcher::new(sample_data(100 * 1024)));
952        let mut buf = [0u8; 16];
953        for i in 0..10 {
954            r.read_exact(&mut buf).unwrap();
955            assert_eq!(buf[0], sample_data(100 * 1024)[i * 16]);
956        }
957        // 160 bytes of sequential reads << 4 KiB read-ahead → one request.
958        assert_eq!(r.stats().requests, 1);
959        assert_eq!(r.fetcher.calls[0], (0, READAHEAD_BYTES - 1));
960    }
961
962    #[test]
963    fn tail_region_read_prefetches_tail_once() {
964        let size: u64 = 1024 * 1024; // 1 MiB
965        let mut r = RangeReader::new(InMemoryFetcher::new(sample_data(
966            usize::try_from(size).unwrap(),
967        )));
968        // Mimic the ZIP EOCD scan: seek near the end, read a few bytes.
969        r.seek(SeekFrom::End(-22)).unwrap();
970        let mut buf = [0u8; 22];
971        r.read_exact(&mut buf).unwrap();
972        assert_eq!(r.stats().requests, 1);
973        assert_eq!(r.fetcher.calls[0], (size - TAIL_PREFETCH_BYTES, size - 1));
974        // Every further read inside the tail is served from cache.
975        r.seek(SeekFrom::End(-4096)).unwrap();
976        let mut big = [0u8; 4096];
977        r.read_exact(&mut big).unwrap();
978        assert_eq!(r.stats().requests, 1);
979        assert_eq!(r.stats().bytes_fetched, TAIL_PREFETCH_BYTES);
980    }
981
982    #[test]
983    fn window_reuse_after_seek_back() {
984        let mut r = RangeReader::new(InMemoryFetcher::new(sample_data(64 * 1024)));
985        let mut buf = [0u8; 128];
986        r.read_exact(&mut buf).unwrap();
987        assert_eq!(r.stats().requests, 1);
988        // Seek back inside the fetched window: no new request.
989        r.seek(SeekFrom::Start(32)).unwrap();
990        r.read_exact(&mut buf).unwrap();
991        assert_eq!(r.stats().requests, 1);
992        let expected = sample_data(64 * 1024);
993        assert_eq!(&buf[..], &expected[32..160]);
994    }
995
996    #[test]
997    fn window_fetch_never_overlaps_cached_tail() {
998        let size: u64 = 200 * 1024;
999        let mut r = RangeReader::new(InMemoryFetcher::new(sample_data(
1000            usize::try_from(size).unwrap(),
1001        )));
1002        // Prime the tail cache.
1003        r.seek(SeekFrom::End(-10)).unwrap();
1004        let mut small = [0u8; 10];
1005        r.read_exact(&mut small).unwrap();
1006        // Large read starting just before the tail: the window fetch must
1007        // stop at the tail boundary, and the rest is served from the tail.
1008        let tail_start = size - TAIL_PREFETCH_BYTES;
1009        r.seek(SeekFrom::Start(tail_start - 100)).unwrap();
1010        let mut big = vec![0u8; 4096];
1011        r.read_exact(&mut big).unwrap();
1012        assert_eq!(r.fetcher.calls[1], (tail_start - 100, tail_start - 1));
1013        let expected = sample_data(usize::try_from(size).unwrap());
1014        let from = usize::try_from(tail_start - 100).unwrap();
1015        assert_eq!(&big[..], &expected[from..from + 4096]);
1016    }
1017
1018    // ---------- Safety budgets ----------
1019
1020    #[test]
1021    fn request_cap_is_enforced() {
1022        let data = sample_data(10 * 1024 * 1024);
1023        let mut r = RangeReader::with_limits(InMemoryFetcher::new(data), 3, u64::MAX);
1024        let mut buf = [0u8; 8];
1025        // Three widely-spaced reads consume the three allowed requests
1026        // (spacing > READAHEAD_BYTES so no window reuse).
1027        for i in 0u64..3 {
1028            r.seek(SeekFrom::Start(i * 100 * 1024)).unwrap();
1029            r.read_exact(&mut buf).unwrap();
1030        }
1031        r.seek(SeekFrom::Start(1024 * 1024)).unwrap();
1032        let err = r.read(&mut buf).unwrap_err();
1033        assert!(
1034            err.to_string()
1035                .contains("range request cap exceeded (3 requests)"),
1036            "unexpected error: {err}"
1037        );
1038    }
1039
1040    #[test]
1041    fn transfer_budget_is_enforced() {
1042        let data = sample_data(10 * 1024 * 1024);
1043        // Budget of 6 KiB: one 4 KiB window fits, the second does not.
1044        let mut r = RangeReader::with_limits(InMemoryFetcher::new(data), u32::MAX, 6 * 1024);
1045        let mut buf = [0u8; 8];
1046        r.read_exact(&mut buf).unwrap();
1047        r.seek(SeekFrom::Start(1024 * 1024)).unwrap();
1048        let err = r.read(&mut buf).unwrap_err();
1049        assert!(
1050            err.to_string().contains("range transfer budget exceeded"),
1051            "unexpected error: {err}"
1052        );
1053    }
1054
1055    // ---------- Error channel ----------
1056
1057    #[test]
1058    fn fetch_error_surfaces_as_io_and_is_recoverable_typed() {
1059        let mut r = RangeReader::new(FailingFetcher { size: 1024 });
1060        let mut buf = [0u8; 8];
1061        let err = r.read(&mut buf).unwrap_err();
1062        assert!(err.to_string().contains("returned status 403"));
1063        let typed = r.take_last_error().expect("typed error must be stored");
1064        assert!(matches!(typed, FetchError::Http(msg)
1065            if msg.contains("returned status 403 Forbidden")));
1066        // Taken once — the slot is now empty.
1067        assert!(r.take_last_error().is_none());
1068    }
1069
1070    #[test]
1071    fn short_fetch_is_a_contract_error() {
1072        let mut r = RangeReader::new(ShortFetcher { size: 1024 * 1024 });
1073        let mut buf = [0u8; 8];
1074        let err = r.read(&mut buf).unwrap_err();
1075        assert!(
1076            err.to_string().contains("bytes for a"),
1077            "unexpected error: {err}"
1078        );
1079    }
1080
1081    // ---------- Pure helpers ----------
1082
1083    #[test]
1084    fn parse_content_range_accepts_the_exact_206_form() {
1085        assert_eq!(parse_content_range("bytes 0-7/1234"), Some((0, 7, 1234)));
1086        assert_eq!(
1087            parse_content_range("bytes 100-199/200"),
1088            Some((100, 199, 200))
1089        );
1090    }
1091
1092    #[test]
1093    fn parse_content_range_rejects_deviant_forms() {
1094        assert_eq!(parse_content_range("bytes */1234"), None);
1095        assert_eq!(parse_content_range("bytes 0-7/*"), None);
1096        assert_eq!(parse_content_range("0-7/1234"), None);
1097        assert_eq!(parse_content_range("bytes 7/1234"), None);
1098        assert_eq!(parse_content_range(""), None);
1099    }
1100
1101    #[test]
1102    fn clean_etag_strips_quotes_and_weak_prefix() {
1103        assert_eq!(clean_etag("\"abc123\""), "abc123");
1104        assert_eq!(clean_etag("W/\"abc123\""), "abc123");
1105        assert_eq!(clean_etag("abc123"), "abc123");
1106    }
1107
1108    // ---------- NPZ end-to-end over the reader ----------
1109
1110    /// Builds a minimal `NPY` v1.0 payload: magic, version, padded header
1111    /// dict, and zeroed data.
1112    fn npy_bytes(descr: &str, shape_literal: &str, data_len: usize) -> Vec<u8> {
1113        let dict =
1114            format!("{{'descr': '{descr}', 'fortran_order': False, 'shape': {shape_literal}, }}");
1115        // Pad so (magic 6 + version 2 + len 2 + header) % 64 == 0, per spec.
1116        let unpadded = 10 + dict.len() + 1; // +1 for the trailing '\n'
1117        let padding = (64 - unpadded % 64) % 64;
1118        let header_len = dict.len() + padding + 1;
1119        let mut out = Vec::with_capacity(10 + header_len + data_len);
1120        out.extend_from_slice(b"\x93NUMPY\x01\x00");
1121        out.extend_from_slice(&u16::try_from(header_len).unwrap().to_le_bytes());
1122        out.extend_from_slice(dict.as_bytes());
1123        out.extend(std::iter::repeat_n(b' ', padding));
1124        out.push(b'\n');
1125        out.extend(std::iter::repeat_n(0u8, data_len));
1126        out
1127    }
1128
1129    /// Builds a stored (uncompressed) ZIP archive from `(name, payload)`
1130    /// entries: local headers, central directory, EOCD. CRC fields are
1131    /// zero — the inspect path never reads entry data, so they are unused.
1132    fn stored_zip(entries: &[(&str, Vec<u8>)]) -> Vec<u8> {
1133        let mut out = Vec::new();
1134        let mut offsets = Vec::new();
1135        for (name, payload) in entries {
1136            offsets.push(u32::try_from(out.len()).unwrap());
1137            let size = u32::try_from(payload.len()).unwrap();
1138            out.extend_from_slice(&0x0403_4b50u32.to_le_bytes()); // LFH sig
1139            out.extend_from_slice(&20u16.to_le_bytes()); // version needed
1140            out.extend_from_slice(&0u16.to_le_bytes()); // flags
1141            out.extend_from_slice(&0u16.to_le_bytes()); // method: stored
1142            out.extend_from_slice(&0u32.to_le_bytes()); // mod time + date
1143            out.extend_from_slice(&0u32.to_le_bytes()); // CRC-32 (unused)
1144            out.extend_from_slice(&size.to_le_bytes()); // compressed size
1145            out.extend_from_slice(&size.to_le_bytes()); // uncompressed size
1146            out.extend_from_slice(&u16::try_from(name.len()).unwrap().to_le_bytes());
1147            out.extend_from_slice(&0u16.to_le_bytes()); // extra len
1148            out.extend_from_slice(name.as_bytes());
1149            out.extend_from_slice(payload);
1150        }
1151        let cd_offset = u32::try_from(out.len()).unwrap();
1152        for ((name, payload), lfh_offset) in entries.iter().zip(&offsets) {
1153            let size = u32::try_from(payload.len()).unwrap();
1154            out.extend_from_slice(&0x0201_4b50u32.to_le_bytes()); // CDFH sig
1155            out.extend_from_slice(&20u16.to_le_bytes()); // version made by
1156            out.extend_from_slice(&20u16.to_le_bytes()); // version needed
1157            out.extend_from_slice(&0u16.to_le_bytes()); // flags
1158            out.extend_from_slice(&0u16.to_le_bytes()); // method: stored
1159            out.extend_from_slice(&0u32.to_le_bytes()); // mod time + date
1160            out.extend_from_slice(&0u32.to_le_bytes()); // CRC-32 (unused)
1161            out.extend_from_slice(&size.to_le_bytes()); // compressed size
1162            out.extend_from_slice(&size.to_le_bytes()); // uncompressed size
1163            out.extend_from_slice(&u16::try_from(name.len()).unwrap().to_le_bytes());
1164            out.extend_from_slice(&0u16.to_le_bytes()); // extra len
1165            out.extend_from_slice(&0u16.to_le_bytes()); // comment len
1166            out.extend_from_slice(&0u16.to_le_bytes()); // disk number
1167            out.extend_from_slice(&0u16.to_le_bytes()); // internal attrs
1168            out.extend_from_slice(&0u32.to_le_bytes()); // external attrs
1169            out.extend_from_slice(&lfh_offset.to_le_bytes());
1170            out.extend_from_slice(name.as_bytes());
1171        }
1172        let cd_size = u32::try_from(out.len()).unwrap() - cd_offset;
1173        let n = u16::try_from(entries.len()).unwrap();
1174        out.extend_from_slice(&0x0605_4b50u32.to_le_bytes()); // EOCD sig
1175        out.extend_from_slice(&0u16.to_le_bytes()); // this disk
1176        out.extend_from_slice(&0u16.to_le_bytes()); // CD start disk
1177        out.extend_from_slice(&n.to_le_bytes()); // entries this disk
1178        out.extend_from_slice(&n.to_le_bytes()); // entries total
1179        out.extend_from_slice(&cd_size.to_le_bytes());
1180        out.extend_from_slice(&cd_offset.to_le_bytes());
1181        out.extend_from_slice(&0u16.to_le_bytes()); // comment len
1182        out
1183    }
1184
1185    #[test]
1186    fn npz_inspect_over_range_reader_reads_metadata_not_data() {
1187        // Two arrays; the big one carries 600 KiB of data the inspect
1188        // must never fetch.
1189        let npz = stored_zip(&[
1190            ("w_enc.npy", npy_bytes("<f4", "(2, 3)", 2 * 3 * 4)),
1191            ("b_dec.npy", npy_bytes("<f4", "(150, 1024)", 600 * 1024)),
1192        ]);
1193        let total = u64::try_from(npz.len()).unwrap();
1194        let mut reader = RangeReader::new(InMemoryFetcher::new(npz));
1195
1196        let info = anamnesis::inspect_npz_from_reader(&mut reader)
1197            .expect("synthetic NPZ must inspect cleanly");
1198
1199        assert_eq!(info.tensors.len(), 2);
1200        let names: Vec<&str> = info.tensors.iter().map(|t| t.name.as_str()).collect();
1201        assert!(names.contains(&"w_enc"), "names: {names:?}");
1202        assert!(names.contains(&"b_dec"), "names: {names:?}");
1203        let b_dec = info.tensors.iter().find(|t| t.name == "b_dec").unwrap();
1204        assert_eq!(b_dec.shape, vec![150, 1024]);
1205
1206        // The efficiency property this module exists for: metadata-only
1207        // transfer, a small number of requests, no weight data fetched.
1208        let stats = reader.stats();
1209        assert!(
1210            stats.requests <= 8,
1211            "expected a handful of range requests, got {}",
1212            stats.requests
1213        );
1214        assert!(
1215            stats.bytes_fetched < total / 4,
1216            "fetched {} of {total} bytes — inspect must not read tensor data",
1217            stats.bytes_fetched
1218        );
1219    }
1220
1221    // ---------- Safetensors end-to-end over the reader ----------
1222
1223    /// Builds a synthetic safetensors buffer: an 8-byte little-endian length
1224    /// prefix, a JSON header with `count` tensor entries (deliberately large
1225    /// enough to exceed the reader's 4 KiB read-ahead window, mirroring a
1226    /// real many-tensor model), and a dummy data section the inspect must
1227    /// never fetch. Returns `(buffer, header_len)` so callers can assert on
1228    /// the exact header size instead of hardcoding it.
1229    fn synthetic_safetensors(count: usize, data_section_len: usize) -> (Vec<u8>, usize) {
1230        use std::fmt::Write as _;
1231
1232        let mut entries = String::new();
1233        let mut offset: u64 = 0;
1234        for i in 0..count {
1235            if i > 0 {
1236                entries.push(',');
1237            }
1238            let start = offset;
1239            let end = offset + 16;
1240            offset = end;
1241            let _ = write!(
1242                entries,
1243                "\"tensor_{i}\":{{\"dtype\":\"F32\",\"shape\":[4],\"data_offsets\":[{start},{end}]}}"
1244            );
1245        }
1246        let header_bytes = format!("{{{entries}}}").into_bytes();
1247        let header_len = header_bytes.len();
1248
1249        let mut buf = Vec::with_capacity(8 + header_len + data_section_len);
1250        buf.extend_from_slice(&u64::try_from(header_len).unwrap().to_le_bytes());
1251        buf.extend_from_slice(&header_bytes);
1252        buf.extend(std::iter::repeat_n(0u8, data_section_len));
1253        (buf, header_len)
1254    }
1255
1256    #[test]
1257    fn safetensors_inspect_over_range_reader_reads_metadata_not_data() {
1258        // 100 tensor entries push the JSON header past the 4 KiB read-ahead
1259        // window, so this also exercises the second-fetch path a real
1260        // many-tensor model triggers. 256 KiB of dummy tensor data the
1261        // inspect must never touch.
1262        let (buf, header_len) = synthetic_safetensors(100, 256 * 1024);
1263        assert!(
1264            header_len > 4096,
1265            "fixture must exceed the 4 KiB read-ahead window to exercise \
1266             the second-fetch path, got {header_len} bytes"
1267        );
1268        let total = u64::try_from(buf.len()).unwrap();
1269        let mut reader = RangeReader::new(InMemoryFetcher::new(buf));
1270
1271        let header = anamnesis::parse_safetensors_header_from_reader(&mut reader)
1272            .expect("synthetic safetensors header must parse cleanly");
1273        assert_eq!(header.tensors.len(), 100);
1274
1275        // The efficiency property this module exists for: metadata-only
1276        // transfer, exactly two requests (the header exceeds the first
1277        // window fetch, forcing a second), no data section fetched.
1278        let stats = reader.stats();
1279        assert_eq!(
1280            stats.requests, 2,
1281            "a {header_len}-byte header should need exactly two range \
1282             requests (first window fetch covers 0..4096, second covers \
1283             the rest); got {}",
1284            stats.requests
1285        );
1286        assert!(
1287            stats.bytes_fetched < total / 4,
1288            "fetched {} of {total} bytes — inspect must not read tensor data",
1289            stats.bytes_fetched
1290        );
1291    }
1292
1293    #[test]
1294    fn safetensors_small_header_fits_in_one_read_ahead_window() {
1295        // A handful of tensors keeps the header well under 4 KiB, so the
1296        // length prefix and the JSON header are both served by the same
1297        // initial window fetch — one request total.
1298        let (buf, header_len) = synthetic_safetensors(3, 1024);
1299        assert!(
1300            header_len < 4096,
1301            "fixture must stay under the 4 KiB read-ahead window, got {header_len} bytes"
1302        );
1303        let mut reader = RangeReader::new(InMemoryFetcher::new(buf));
1304
1305        let header = anamnesis::parse_safetensors_header_from_reader(&mut reader)
1306            .expect("synthetic safetensors header must parse cleanly");
1307        assert_eq!(header.tensors.len(), 3);
1308        assert_eq!(reader.stats().requests, 1);
1309    }
1310
1311    // ---------- GGUF end-to-end over the reader ----------
1312
1313    /// In-memory `GGUF` v3 byte-stream builder — mirrors the layout
1314    /// anamnesis's parser expects (magic, version, counts, tensor-info
1315    /// table, alignment-padded data section). Minimal: no metadata KV
1316    /// entries, `F32` tensors only (`GGML_TYPE_F32 = 0`).
1317    struct GgufBuilder {
1318        buf: Vec<u8>,
1319    }
1320
1321    impl GgufBuilder {
1322        fn new(tensor_count: u64) -> Self {
1323            let mut buf = Vec::new();
1324            buf.extend_from_slice(b"GGUF");
1325            buf.extend_from_slice(&3u32.to_le_bytes()); // version
1326            buf.extend_from_slice(&tensor_count.to_le_bytes());
1327            buf.extend_from_slice(&0u64.to_le_bytes()); // kv_count = 0
1328            Self { buf }
1329        }
1330
1331        fn push_f32_tensor_info(&mut self, name: &str, shape: &[u64], relative_offset: u64) {
1332            self.buf
1333                .extend_from_slice(&u64::try_from(name.len()).unwrap().to_le_bytes());
1334            self.buf.extend_from_slice(name.as_bytes());
1335            self.buf
1336                .extend_from_slice(&u32::try_from(shape.len()).unwrap().to_le_bytes());
1337            for d in shape {
1338                self.buf.extend_from_slice(&d.to_le_bytes());
1339            }
1340            self.buf.extend_from_slice(&0u32.to_le_bytes()); // GGML_TYPE_F32 = 0
1341            self.buf.extend_from_slice(&relative_offset.to_le_bytes());
1342        }
1343
1344        fn pad_to_alignment(&mut self, alignment: usize) {
1345            let rem = self.buf.len() % alignment;
1346            if rem != 0 {
1347                self.buf.extend(std::iter::repeat_n(0u8, alignment - rem));
1348            }
1349        }
1350
1351        fn push_zeros(&mut self, n: usize) {
1352            self.buf.extend(std::iter::repeat_n(0u8, n));
1353        }
1354
1355        fn finish(self) -> Vec<u8> {
1356            self.buf
1357        }
1358    }
1359
1360    /// Builds a minimal `GGUF` v3 byte stream: two `F32` tensors, no
1361    /// metadata, default 32-byte alignment. `b_dec` carries 600 KiB of data
1362    /// the front-matter parse must never fetch — same magnitude as the NPZ
1363    /// fixture above, for an apples-to-apples `bytes_fetched` comparison.
1364    fn synthetic_gguf() -> Vec<u8> {
1365        let mut b = GgufBuilder::new(2);
1366        b.push_f32_tensor_info("w_enc", &[2, 3], 0);
1367        b.push_f32_tensor_info("b_dec", &[153_600], 32);
1368        b.pad_to_alignment(32); // end of tensor-info table -> data_section_start
1369        b.push_zeros(24); // w_enc data (2*3*4 bytes)
1370        b.push_zeros(8); // pad up to b_dec's relative offset 32
1371        b.push_zeros(153_600 * 4); // b_dec data (600 KiB)
1372        b.finish()
1373    }
1374
1375    #[test]
1376    fn gguf_front_matter_over_range_reader_reads_metadata_not_data() {
1377        let gguf = synthetic_gguf();
1378        let total = u64::try_from(gguf.len()).unwrap();
1379        let mut reader = RangeReader::new(InMemoryFetcher::new(gguf));
1380
1381        let front = anamnesis::parse_gguf_front_matter_from_reader(&mut reader)
1382            .expect("synthetic GGUF must parse cleanly");
1383
1384        assert_eq!(front.tensor_infos.len(), 2);
1385        let names: Vec<&str> = front.tensor_infos.iter().map(|t| t.name.as_str()).collect();
1386        assert!(names.contains(&"w_enc"), "names: {names:?}");
1387        assert!(names.contains(&"b_dec"), "names: {names:?}");
1388        let b_dec = front
1389            .tensor_infos
1390            .iter()
1391            .find(|t| t.name == "b_dec")
1392            .unwrap();
1393        assert_eq!(b_dec.shape, vec![153_600]);
1394
1395        // The efficiency property this module exists for: metadata-only
1396        // transfer, a small number of requests, no weight data fetched.
1397        // Measured: exactly 1 range request, 64 KiB fetched (bounded by the
1398        // reader's internal `BufReader` capacity) out of the 600 KiB payload.
1399        let stats = reader.stats();
1400        assert!(
1401            stats.requests <= 8,
1402            "expected a handful of range requests, got {}",
1403            stats.requests
1404        );
1405        assert!(
1406            stats.bytes_fetched < total / 4,
1407            "fetched {} of {total} bytes — front-matter parse must not read tensor data",
1408            stats.bytes_fetched
1409        );
1410    }
1411
1412    // ---------- PTH end-to-end over the reader ----------
1413
1414    /// Raw `data.pkl` bytes extracted from anamnesis's real
1415    /// `tests/fixtures/pth_reference/algzoo_rnn_small.pth` fixture — a
1416    /// genuine `torch.save()`-produced pickle stream (448 bytes), not a
1417    /// hand-rolled one. Describes 3 `F32` tensors: `rnn.weight_ih_l0`
1418    /// `[2, 1]`, `rnn.weight_hh_l0` `[2, 2]`, `linear.weight` `[2, 2]` (see
1419    /// that fixture's sibling `algzoo_rnn_small_reference.json`). Reusing a
1420    /// real pickle stream avoids hand-crafting pickle VM opcodes — the same
1421    /// reasoning anamnesis's own `bench_pth_inspect` fixture uses. The
1422    /// surrounding ZIP archive below is synthetic, with a large dummy
1423    /// storage entry the front-matter parse must never fetch.
1424    const ALGZOO_RNN_SMALL_DATA_PKL: &[u8] = &[
1425        0x80, 0x02, 0x63, 0x63, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x0a,
1426        0x4f, 0x72, 0x64, 0x65, 0x72, 0x65, 0x64, 0x44, 0x69, 0x63, 0x74, 0x0a, 0x71, 0x00, 0x29,
1427        0x52, 0x71, 0x01, 0x28, 0x58, 0x10, 0x00, 0x00, 0x00, 0x72, 0x6e, 0x6e, 0x2e, 0x77, 0x65,
1428        0x69, 0x67, 0x68, 0x74, 0x5f, 0x69, 0x68, 0x5f, 0x6c, 0x30, 0x71, 0x02, 0x63, 0x74, 0x6f,
1429        0x72, 0x63, 0x68, 0x2e, 0x5f, 0x75, 0x74, 0x69, 0x6c, 0x73, 0x0a, 0x5f, 0x72, 0x65, 0x62,
1430        0x75, 0x69, 0x6c, 0x64, 0x5f, 0x74, 0x65, 0x6e, 0x73, 0x6f, 0x72, 0x5f, 0x76, 0x32, 0x0a,
1431        0x71, 0x03, 0x28, 0x28, 0x58, 0x07, 0x00, 0x00, 0x00, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67,
1432        0x65, 0x71, 0x04, 0x63, 0x74, 0x6f, 0x72, 0x63, 0x68, 0x0a, 0x46, 0x6c, 0x6f, 0x61, 0x74,
1433        0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x0a, 0x71, 0x05, 0x58, 0x01, 0x00, 0x00, 0x00,
1434        0x30, 0x71, 0x06, 0x58, 0x06, 0x00, 0x00, 0x00, 0x63, 0x75, 0x64, 0x61, 0x3a, 0x30, 0x71,
1435        0x07, 0x4b, 0x0a, 0x74, 0x71, 0x08, 0x51, 0x4b, 0x00, 0x4b, 0x02, 0x4b, 0x01, 0x86, 0x71,
1436        0x09, 0x4b, 0x01, 0x4b, 0x01, 0x86, 0x71, 0x0a, 0x89, 0x68, 0x00, 0x29, 0x52, 0x71, 0x0b,
1437        0x74, 0x71, 0x0c, 0x52, 0x71, 0x0d, 0x58, 0x10, 0x00, 0x00, 0x00, 0x72, 0x6e, 0x6e, 0x2e,
1438        0x77, 0x65, 0x69, 0x67, 0x68, 0x74, 0x5f, 0x68, 0x68, 0x5f, 0x6c, 0x30, 0x71, 0x0e, 0x68,
1439        0x03, 0x28, 0x28, 0x68, 0x04, 0x68, 0x05, 0x68, 0x06, 0x58, 0x06, 0x00, 0x00, 0x00, 0x63,
1440        0x75, 0x64, 0x61, 0x3a, 0x30, 0x71, 0x0f, 0x4b, 0x0a, 0x74, 0x71, 0x10, 0x51, 0x4b, 0x02,
1441        0x4b, 0x02, 0x4b, 0x02, 0x86, 0x71, 0x11, 0x4b, 0x02, 0x4b, 0x01, 0x86, 0x71, 0x12, 0x89,
1442        0x68, 0x00, 0x29, 0x52, 0x71, 0x13, 0x74, 0x71, 0x14, 0x52, 0x71, 0x15, 0x58, 0x0d, 0x00,
1443        0x00, 0x00, 0x6c, 0x69, 0x6e, 0x65, 0x61, 0x72, 0x2e, 0x77, 0x65, 0x69, 0x67, 0x68, 0x74,
1444        0x71, 0x16, 0x68, 0x03, 0x28, 0x28, 0x68, 0x04, 0x68, 0x05, 0x58, 0x01, 0x00, 0x00, 0x00,
1445        0x31, 0x71, 0x17, 0x58, 0x06, 0x00, 0x00, 0x00, 0x63, 0x75, 0x64, 0x61, 0x3a, 0x30, 0x71,
1446        0x18, 0x4b, 0x04, 0x74, 0x71, 0x19, 0x51, 0x4b, 0x00, 0x4b, 0x02, 0x4b, 0x02, 0x86, 0x71,
1447        0x1a, 0x4b, 0x02, 0x4b, 0x01, 0x86, 0x71, 0x1b, 0x89, 0x68, 0x00, 0x29, 0x52, 0x71, 0x1c,
1448        0x74, 0x71, 0x1d, 0x52, 0x71, 0x1e, 0x75, 0x7d, 0x71, 0x1f, 0x58, 0x09, 0x00, 0x00, 0x00,
1449        0x5f, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x71, 0x20, 0x68, 0x00, 0x29, 0x52,
1450        0x71, 0x21, 0x28, 0x58, 0x00, 0x00, 0x00, 0x00, 0x71, 0x22, 0x7d, 0x71, 0x23, 0x58, 0x07,
1451        0x00, 0x00, 0x00, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x71, 0x24, 0x4b, 0x01, 0x73,
1452        0x58, 0x03, 0x00, 0x00, 0x00, 0x72, 0x6e, 0x6e, 0x71, 0x25, 0x7d, 0x71, 0x26, 0x68, 0x24,
1453        0x4b, 0x01, 0x73, 0x58, 0x06, 0x00, 0x00, 0x00, 0x6c, 0x69, 0x6e, 0x65, 0x61, 0x72, 0x71,
1454        0x27, 0x7d, 0x71, 0x28, 0x68, 0x24, 0x4b, 0x01, 0x73, 0x75, 0x73, 0x62, 0x2e,
1455    ];
1456
1457    #[test]
1458    fn pth_front_matter_over_range_reader_reads_metadata_not_data() {
1459        // The real `data.pkl` plus a 600 KiB dummy storage entry the
1460        // front-matter parse must never fetch — same magnitude as the
1461        // NPZ/GGUF fixtures above, for an apples-to-apples comparison.
1462        let pth = stored_zip(&[
1463            ("archive/data.pkl", ALGZOO_RNN_SMALL_DATA_PKL.to_vec()),
1464            ("archive/data/bulk", vec![0u8; 600 * 1024]),
1465        ]);
1466        let total = u64::try_from(pth.len()).unwrap();
1467        let mut reader = RangeReader::new(InMemoryFetcher::new(pth));
1468
1469        let front = anamnesis::parse_pth_front_matter_from_reader(&mut reader)
1470            .expect("real algzoo data.pkl must parse cleanly");
1471
1472        assert_eq!(front.tensors.len(), 3);
1473        let names: Vec<&str> = front.tensors.iter().map(|t| t.name.as_str()).collect();
1474        assert!(names.contains(&"rnn.weight_ih_l0"), "names: {names:?}");
1475        assert!(names.contains(&"rnn.weight_hh_l0"), "names: {names:?}");
1476        assert!(names.contains(&"linear.weight"), "names: {names:?}");
1477        let hh = front
1478            .tensors
1479            .iter()
1480            .find(|t| t.name == "rnn.weight_hh_l0")
1481            .unwrap();
1482        assert_eq!(hh.shape, vec![2, 2]);
1483        assert!(!front.big_endian);
1484
1485        // The efficiency property this module exists for: metadata-only
1486        // transfer, a small number of requests, no bulk storage data
1487        // fetched.
1488        let stats = reader.stats();
1489        assert!(
1490            stats.requests <= 8,
1491            "expected a handful of range requests, got {}",
1492            stats.requests
1493        );
1494        assert!(
1495            stats.bytes_fetched < total / 4,
1496            "fetched {} of {total} bytes — front-matter parse must not read tensor data",
1497            stats.bytes_fetched
1498        );
1499    }
1500}