Stream shard downloads and reject on JS decompress failure

Address review on the compressed-artifact handling:

- fetch_perf_report.py: download_url streamed the whole shard body into
  memory (resp.read() then write). Compressed perf artifacts exist only
  for the large shards, and up to 8 shards download in parallel, so this
  could hold hundreds of MiB. Stream the response through the matching
  decompressor into dest instead, peeking the magic bytes via a small
  _PrefixedReader.
- fetch_ci_report.js: maybeDecompress could throw from the async 'end'
  handler (missing zstd binary, corrupt archive), escaping as an uncaught
  exception. Wrap it in try/catch and reject so it surfaces normally.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

.claude/tools/fetch_ci_report.js

@@ -115,8 +115,14 @@ function fetchUrl(urlString, credentials = null) {
       const chunks = [];
       stream.on('data', chunk => chunks.push(chunk));
       stream.on('end', () => {
-        const body = maybeDecompress(Buffer.concat(chunks)).toString('utf8');
-        resolve(body);
+        // maybeDecompress may throw (missing zstd binary, corrupt archive); a throw from this
+        // async handler would escape the Promise as an uncaught exception, so surface it as a
+        // normal rejection instead.
+        try {
+          resolve(maybeDecompress(Buffer.concat(chunks)).toString('utf8'));
+        } catch (err) {
+          reject(err);
+        }
       });
       stream.on('error', reject);
     });

.claude/tools/fetch_perf_report.py

@@ -92,14 +92,74 @@ def fetch_url(url):
     return _read_url_bytes(url).decode("utf-8")
 
 
+class _PrefixedReader:
+    """File-like wrapper that yields `prefix` bytes before the rest of `stream`.
+
+    Lets us peek at the magic bytes for format detection and still stream the whole
+    body through a decompressor without ever holding it fully in memory.
+    """
+
+    def __init__(self, prefix, stream):
+        self._prefix = prefix
+        self._stream = stream
+
+    def read(self, size=-1):
+        if not self._prefix:
+            return self._stream.read(size)
+        if size is None or size < 0:
+            data, self._prefix = self._prefix + self._stream.read(), b""
+            return data
+        if size <= len(self._prefix):
+            data, self._prefix = self._prefix[:size], self._prefix[size:]
+            return data
+        data, self._prefix = self._prefix + self._stream.read(size - len(self._prefix)), b""
+        return data
+
+
+def _stream_to_file(resp, dest):
+    """Stream `resp` to `dest`, transparently decompressing zstd/gzip by magic bytes.
+
+    Compressed perf artifacts exist precisely for the large shards, and several shards
+    download in parallel, so the body is streamed in chunks rather than buffered.
+    """
+    header = resp.read(4)
+    source = _PrefixedReader(header, resp)
+    with open(dest, "wb") as f:
+        if header == b"\x28\xb5\x2f\xfd":  # zstd magic
+            try:
+                import zstandard  # optional dependency
+                shutil.copyfileobj(zstandard.ZstdDecompressor().stream_reader(source), f)
+            except ImportError:
+                proc = subprocess.Popen(["zstd", "-dcq"], stdin=subprocess.PIPE, stdout=f,
+                                        stderr=subprocess.PIPE)
+                shutil.copyfileobj(source, proc.stdin)
+                proc.stdin.close()
+                _, stderr = proc.communicate(timeout=120)
+                if proc.returncode != 0:
+                    raise RuntimeError(f"zstd decompression failed: {stderr.decode('utf-8', 'replace')}")
+        elif header[:2] == b"\x1f\x8b":  # gzip magic
+            with gzip.GzipFile(fileobj=source) as gz:
+                shutil.copyfileobj(gz, f)
+        else:
+            shutil.copyfileobj(source, f)
+
+
 def download_url(url, dest):
     """Download a URL to a file, transparently decompressing. Raises on failure.
 
-    Shard-local failures are isolated by the `download_shard` worker, not here.
+    Falls back to `url + '.zst'` on 404/403. Shard-local failures are isolated by the
+    `download_shard` worker, not here.
     """
-    data = _read_url_bytes(url)
-    with open(dest, "wb") as f:
-        f.write(data)
+    candidates = [url] if url.endswith(".zst") else [url, url + ".zst"]
+    last_error = None
+    for candidate in candidates:
+        try:
+            with urlopen(candidate, timeout=60) as resp:
+                _stream_to_file(resp, dest)
+                return
+        except HTTPError as e:
+            last_error = f"HTTP {e.code}: {e.reason} for {candidate}"
+    raise RuntimeError(last_error)
 
 
 # ---------------------------------------------------------------------------

.claude/tools/test_fetch_perf_report.py

@@ -164,6 +164,34 @@ def test_maybe_decompress_handles_plain_gzip_zstd():
     assert fpr.maybe_decompress(zst) == plain
 
 
+def test_stream_to_file_handles_plain_gzip_zstd():
+    import gzip as _gzip
+    import io as _io
+    import subprocess as _sp
+
+    plain = b"metric\tleft\tright\n" + b"memory_usage\t100\t90\n" * 10000  # >4 bytes, multi-chunk
+    zst = _sp.run(["zstd", "-cq"], input=plain, capture_output=True).stdout
+    variants = {"plain": plain, "gzip": _gzip.compress(plain), "zstd": zst}
+
+    with tempfile.TemporaryDirectory() as tmp:
+        for name, body in variants.items():
+            dest = os.path.join(tmp, name)
+            # A BytesIO stands in for the urlopen response: same .read(size) contract.
+            fpr._stream_to_file(_io.BytesIO(body), dest)
+            with open(dest, "rb") as f:
+                assert f.read() == plain, name
+
+
+def test_prefixed_reader_reassembles_stream():
+    import io as _io
+
+    payload = bytes(range(256)) * 8
+    reader = fpr._PrefixedReader(payload[:4], _io.BytesIO(payload[4:]))
+    # Small reads across the prefix boundary and a final read-all must round-trip.
+    got = reader.read(2) + reader.read(5) + reader.read()
+    assert got == payload
+
+
 def test_download_shard_isolates_failures():
     import gzip as _gzip
     import subprocess as _sp
@@ -193,5 +221,7 @@ if __name__ == "__main__":
     test_classification_matches_compare_sh()
     test_summary_counts()
     test_maybe_decompress_handles_plain_gzip_zstd()
+    test_stream_to_file_handles_plain_gzip_zstd()
+    test_prefixed_reader_reassembles_stream()
     test_download_shard_isolates_failures()
     print("All fetch_perf_report tests passed (or skipped).")