Merge pull request #886 from tdurieux/fix/production-followup-errors

fix: handle production source failures and reduce ZIP work

src/core/GitHubUtils.ts

@@ -15,6 +15,7 @@ import { getCredential, getCredentialToken } from "./credentials";
 import config from "../config";
 import { createLogger } from "./logger";
 import { measureStage } from "./request-monitoring";
+import { setTimeout as delay } from "timers/promises";
 
 const logger = createLogger("github");
 
@@ -85,7 +86,18 @@ export function getTokenGateResetAt(token: string): number {
   return gate.resetAt;
 }
 
-async function waitForTokenGate(token: string): Promise<void> {
+function abortableGateLookup(lookup: Promise<number>, signal: AbortSignal): Promise<number> {
+  return new Promise((resolve, reject) => {
+    const cancelled = () => { signal.removeEventListener("abort", cancelled); reject(signal.reason); };
+    if (signal.aborted) cancelled();
+    else signal.addEventListener("abort", cancelled, { once: true });
+    lookup.then(value => { signal.removeEventListener("abort", cancelled); resolve(value); },
+      error => { signal.removeEventListener("abort", cancelled); reject(error); });
+  });
+}
+
+async function waitForTokenGate(token: string, signal?: AbortSignal): Promise<void> {
+  signal?.throwIfAborted();
   const key = githubQuotaKey(token);
   const localGate = tokenGates.get(key);
   let waitMs = 0;
@@ -96,7 +108,9 @@ async function waitForTokenGate(token: string): Promise<void> {
     waitMs = resetAt - Date.now();
   }
 
-  const redisResetAt = await getRedisGateResetAt(key);
+  const lookup = getRedisGateResetAt(key);
+  const redisResetAt = await (signal ? abortableGateLookup(lookup, signal) : lookup);
+  signal?.throwIfAborted();
   if (redisResetAt > resetAt) {
     resetAt = redisResetAt;
     waitMs = resetAt - Date.now();
@@ -112,7 +126,7 @@ async function waitForTokenGate(token: string): Promise<void> {
     waitMs,
     resetAt: new Date(resetAt).toISOString(),
   });
-  await new Promise((resolve) => setTimeout(resolve, waitMs));
+  await delay(waitMs, undefined, { signal });
   if (localGate) tokenGates.delete(key);
 }
 

src/core/archive-entry.ts

@@ -1,5 +1,17 @@
-import { Archiver, EntryData } from "archiver";
+import { Archiver, EntryData, ZipEntryData } from "archiver";
+import { extname } from "path";
 import { Readable } from "stream";
+
+const compressedExtensions = new Set([
+  ".png", ".jpg", ".jpeg", ".gif", ".webp", ".avif", ".heif", ".heic",
+  ".mp4", ".m4v", ".mov", ".webm", ".mp3", ".m4a", ".aac", ".ogg", ".flac",
+  ".zip", ".gz", ".tgz", ".bz2", ".xz", ".7z", ".rar", ".zst", ".woff", ".woff2",
+]);
+
+/** Store already compressed assets; every entry still passes through anonymization. */
+export function zipEntryOptions(name: string, originalName = name): ZipEntryData {
+  return { name, store: compressedExtensions.has(extname(originalName).toLowerCase()) };
+}
 /** Allow only one active entry; compression and the response provide backpressure. */
 export function appendArchiveEntry(archive: Archiver, source: Readable, options: EntryData): Promise<void> {
   return new Promise((resolve, reject) => {

src/core/logger.ts

@@ -320,7 +320,10 @@ export function serializeError(err: unknown): Record<string, unknown> {
 
   // AnonymousError carries an httpStatus and an inner cause.
   if (typeof e.httpStatus === "number") out.httpStatus = e.httpStatus;
-  if (e.code !== undefined && e.code !== e.message) out.code = e.code;
+  // Octokit's deprecated `code` getter emits a stack on every read. Only
+  // inspect data properties so logging never invokes that getter.
+  const code = Object.getOwnPropertyDescriptor(e, "code")?.value;
+  if (code !== undefined && code !== e.message) out.code = code;
   if (typeof e.url === "function") {
     try {
       const u = e.url();

src/core/repository-failure-status.ts

@@ -6,6 +6,7 @@ export function repositoryFailureStatus(code?: string): number {
     case "token_expired": case "not_connected": return 401;
     case "github_rate_limit_exceeded": return 429;
     case "github_unavailable": return 502;
+    case "repo_empty": return 409;
     default: return 500;
   }
 }

src/core/request-monitoring.ts

@@ -148,8 +148,12 @@ export function monitorRequests(service: Service, metrics = requestMetrics,
       const slow = ms > (transfer ? 10000 : 1000);
       metrics.observe("request", route, method, ms, { bytes, aborted, error: res.statusCode >= 500, slow });
       if (firstByteMs != null) metrics.observe("first_byte", route, method, firstByteMs);
+      const responseMs = firstByteMs == null ? null : Math.max(0, ms - firstByteMs);
       const detail = { requestId: id, service, method, route, status: res.statusCode,
-        outcome: aborted ? "interrupted" : "completed", ms: Math.round(ms), firstByteMs: firstByteMs == null ? null : Math.round(firstByteMs), bytes, stages: trace.stages };
+        outcome: aborted ? "interrupted" : "completed", ms: Math.round(ms), firstByteMs: firstByteMs == null ? null : Math.round(firstByteMs),
+        responseMs: responseMs == null ? null : Math.round(responseMs),
+        bytesPerSecond: responseMs != null && responseMs >= 1 && bytes > 0 ? Math.round(bytes * 1000 / responseMs) : null,
+        bytes, stages: trace.stages };
       if (slow || aborted) log.warn(aborted ? "request interrupted" : "slow request", detail);
       else if (!route.includes("healthcheck")) log.info("request", detail);
     };

src/core/source/GitHubStream.ts

@@ -15,7 +15,7 @@ import { basename, dirname } from "path";
 
 import * as stream from "stream";
 import AnonymousError from "../AnonymousError";
-import { octokit, waitForTokenGate } from "../GitHubUtils";
+import { octokit, waitForTokenGate, RateLimitDelayError } from "../GitHubUtils";
 import FileModel from "../model/files/files.model";
 import { IFile } from "../model/files/files.types";
 import { createLogger, serializeError } from "../logger";
@@ -24,13 +24,20 @@ import config from "../../config";
 import { isConnected } from "../../server/database";
 import AnonymizedRepositoryModel from "../model/anonymizedRepositories/anonymizedRepositories.model";
 import { RepositoryStatus } from "../types";
+import { cachedContentMiss, classifyContentMiss } from "./github-content-miss";
 
 
 const logger = createLogger("gh-stream");
 
 const GH_API_CONCURRENCY = 6;
 const cacheFills = new Map<string, Promise<void>>();
-const missingContent = new ExpiringMap<boolean>(2048);
+const missingContent = new ExpiringMap<string>(2048);
+
+/** Stop upstream before Readable.from waits for the pending iterator to finish. */
+function stopSourceOnDestroy(output: stream.Readable, stop: () => void): void {
+  const destroy = output._destroy;
+  output._destroy = (error, callback) => { stop(); destroy.call(output, error, callback); };
+}
 
 async function pMap<T, R>(
   items: T[],
@@ -68,6 +75,7 @@ export function githubRawFileUrl(
 
 function downloadError(error: Error, filePath: string): AnonymousError {
   if (error instanceof AnonymousError) return error;
+  if (error instanceof RateLimitDelayError) return new AnonymousError("github_rate_limit_exceeded", { httpStatus: 429, cause: error });
   const response = error as Error & { response?: { statusCode?: number }; status?: number; httpStatus?: number };
   const status = response.response?.statusCode ?? response.status ?? response.httpStatus;
   const code = status === 401 || status === 403 ? "file_not_accessible"
@@ -115,10 +123,9 @@ export default class GitHubStream extends GitHubBase {
     }
   }
 
-  // GitHub's web raw URL auto-resolves Git LFS pointers via redirect to
-  // media.githubusercontent.com, with the auth header carried through. The
-  // blob endpoint above returns the raw pointer text instead, so we use this
-  // as the fallback for LFS files (#95).
+  // The web URL resolves public LFS pointers. Private web URLs can return
+  // 404 even when the REST API can read the file, so use the scoped contents
+  // API to obtain a short-lived download URL in that case.
   private downloadFileViaRaw(token: string, filePath: string) {
     const url = githubRawFileUrl(
       this.data.organization,
@@ -127,11 +134,73 @@ export default class GitHubStream extends GitHubBase {
       filePath
     );
     logger.debug("downloading via raw URL (LFS)", { url });
-    return got.stream(url, {
-      hooks: { beforeRequest: [async () => { await githubTokenContext(token)?.renew(); }] },
-      headers: !token || githubTokenContext(token)?.publicRepository ? {} : { authorization: `token ${token}` },
-      followRedirect: true,
-    });
+    const data = this.data;
+    const cancellation = new AbortController();
+    let active: stream.Readable | undefined;
+    const output = stream.Readable.from((async function* () {
+      try {
+        active = got.stream(url, {
+          hooks: { beforeRequest: [async () => { await githubTokenContext(token)?.renew(); }] },
+          headers: !token || githubTokenContext(token)?.publicRepository ? {} : { authorization: `token ${token}` },
+          followRedirect: true,
+        });
+        for await (const chunk of active) yield chunk;
+        return;
+      } catch (error) {
+        if (!token || githubTokenContext(token)?.publicRepository ||
+            (error as { response?: { statusCode?: number } }).response?.statusCode !== 404) throw error;
+        await waitForTokenGate(token, cancellation.signal);
+        const response = await octokit(token).repos.getContent({
+          owner: data.organization, repo: data.repoName,
+          path: filePath, ref: data.commit || "HEAD", mediaType: { format: "object" },
+          request: { signal: cancellation.signal, timeout: 30_000 },
+        }).catch(apiError => {
+          if (apiError instanceof AnonymousError) throw apiError;
+          const upstream = apiError as { status?: number; response?: { data?: { message?: string } } };
+          if (upstream.status === 422) {
+            const missingCommit = /^No commit found for SHA\b/.test(upstream.response?.data?.message || "");
+            throw new AnonymousError(missingCommit ? "commit_not_found" : "upstream_error", {
+              httpStatus: missingCommit ? 404 : 502, cause: apiError as Error,
+            });
+          }
+          throw apiError;
+        });
+        if (cancellation.signal.aborted) return;
+        const metadata = response.data as { type?: string; download_url?: string };
+        if (metadata.type !== "file" || !metadata.download_url) throw error;
+        const assertDownloadUrl = (value: URL) => {
+          if (value.protocol !== "https:" || value.username || value.password || value.port ||
+              !["raw.githubusercontent.com", "media.githubusercontent.com"].includes(value.hostname)) {
+            throw new AnonymousError("upstream_error", { httpStatus: 502 });
+          }
+        };
+        assertDownloadUrl(new URL(metadata.download_url));
+        // The API URL supplies its own authorization. Never forward the
+        // owner's token to the download host or a redirect destination.
+        active = got.stream(metadata.download_url, {
+          hooks: { beforeRedirect: [options => { assertDownloadUrl(options.url); }] },
+        });
+        let probe = Buffer.alloc(0);
+        let checked = false;
+        for await (const chunk of active) {
+          if (checked) { yield chunk; continue; }
+          probe = Buffer.concat([probe, Buffer.from(chunk)]);
+          if (probe.length < 150) continue;
+          if (probe.toString("utf8").startsWith("version https://git-lfs.github.com/spec/")) {
+            throw new AnonymousError("upstream_error", { httpStatus: 502 });
+          }
+          checked = true; yield probe;
+        }
+        if (!checked && probe.length) {
+          if (probe.toString("utf8").startsWith("version https://git-lfs.github.com/spec/")) {
+            throw new AnonymousError("upstream_error", { httpStatus: 502 });
+          }
+          yield probe;
+        }
+      }
+    })(), { objectMode: false });
+    stopSourceOnDestroy(output, () => { cancellation.abort(); active?.destroy(); });
+    return output;
   }
 
   private deferredRawDownload(token: string, filePath: string): stream.Readable {
@@ -141,7 +210,7 @@ export default class GitHubStream extends GitHubBase {
       active = download();
       for await (const chunk of active) yield chunk;
     })(), { objectMode: false });
-    output.once("close", () => active?.destroy());
+    stopSourceOnDestroy(output, () => active?.destroy());
     return output;
   }
 
@@ -225,7 +294,7 @@ export default class GitHubStream extends GitHubBase {
         probe = Buffer.concat([probe, bytes]);
         if (probe.length < 150) continue;
         decided = true;
-        if (probe.toString("utf8", 0, 39) === "version https://git-lfs.github.com/spec/") {
+        if (probe.toString("utf8").startsWith("version https://git-lfs.github.com/spec/")) {
           active = downloadRaw();
           for await (const raw of active) yield raw;
           return;
@@ -244,7 +313,7 @@ export default class GitHubStream extends GitHubBase {
     }
     const resolved = stream.Readable.from(resolve(), { objectMode: false });
     // Keep the generator lazy until the consumer has installed its handlers.
-    resolved.once("close", () => active.destroy());
+    stopSourceOnDestroy(resolved, () => active.destroy());
     return resolved;
   }
 
@@ -314,10 +383,13 @@ export default class GitHubStream extends GitHubBase {
       filling = coordinatedFill(key, cached, async () => {
         await assertActive();
         const token = await this.data.getToken();
+        const sourceMiss = cachedContentMiss(this.data, token);
+        if (sourceMiss && sourceMiss !== "file_not_found") throw new AnonymousError(sourceMiss, { httpStatus: 404 });
         const missKey = createHash("sha256").update(JSON.stringify([
           key, this.data.organization, this.data.repoName, token,
         ])).digest("hex");
-        if (missingContent.get(missKey)) throw new AnonymousError("file_not_found", { httpStatus: 404 });
+        const missed = missingContent.get(missKey);
+        if (missed) throw new AnonymousError(missed, { httpStatus: 404 });
         try {
           return await measureStage("upstream", async () => {
             const content = await this.downloadWithFallback(token, expected.sha, filePath)
@@ -342,7 +414,9 @@ export default class GitHubStream extends GitHubBase {
           });
         } catch (error) {
           if (error instanceof AnonymousError && error.message === "file_not_found" && error.httpStatus === 404) {
-            missingContent.set(missKey, true, 30_000);
+            const code = await classifyContentMiss(this.data, token);
+            missingContent.set(missKey, code, 30_000);
+            if (code !== "file_not_found") throw new AnonymousError(code, { httpStatus: 404, cause: error });
           }
           throw error;
         }
@@ -430,19 +504,37 @@ export default class GitHubStream extends GitHubBase {
     count = { request: 0, file: 0 },
     opt = { recursive: true, callback: () => {} }
   ) {
-    await waitForTokenGate(token);
-    const ghRes = await oct.git.getTree({
-      owner: this.data.organization,
-      repo: this.data.repoName,
-      tree_sha: sha,
-      recursive: opt.recursive === true ? "1" : undefined,
-    });
-    count.request++;
-    count.file += ghRes.data.tree.length;
-    if (opt.callback) {
-      opt.callback();
+    try {
+      await waitForTokenGate(token);
+      const ghRes = await oct.git.getTree({
+        owner: this.data.organization,
+        repo: this.data.repoName,
+        tree_sha: sha,
+        recursive: opt.recursive === true ? "1" : undefined,
+      });
+      count.request++;
+      count.file += ghRes.data.tree.length;
+      if (opt.callback) opt.callback();
+      return ghRes.data;
+    } catch (error) {
+      // Keep quota, credential and lifecycle errors intact. A failed tree
+      // request does not imply that the repository has disappeared.
+      if (error instanceof AnonymousError || error instanceof RateLimitDelayError) throw error;
+      const upstream = error as { status?: number; response?: { data?: { message?: string } } };
+      const status = upstream.status;
+      const invalidObject = status === 422 && /^Invalid object requested\. SHA must identify a commit or a tree\b/.test(upstream.response?.data?.message || "");
+      const code = status === 409 ? "repo_empty"
+        : status === 404 ? await classifyGitHubMissError(error, this.data)
+        : invalidObject ? "commit_not_found"
+        : status === 401 ? "token_expired"
+        : status === 403 ? "repo_not_accessible" : "github_unavailable";
+      throw new AnonymousError(code, {
+        httpStatus: status === 409 ? 409 : status === 404 || invalidObject ? 404
+          : status === 401 || status === 403 ? status : 502,
+        object: this.data,
+        cause: error as Error,
+      });
     }
-    return ghRes.data;
   }
 
   private async getTruncatedTree(
@@ -457,55 +549,17 @@ export default class GitHubStream extends GitHubBase {
       file: 0,
     };
     const output: IFile[] = [];
-    let data;
-    try {
-      data = await this.getGHTree(oct, token, sha, count, {
-        recursive: true,
-        callback: () => {
-          if (progress) {
-            progress("List file: " + count.file);
-          }
-        },
-      });
-      if (!data.truncated) return this.tree2Tree(data.tree, parentPath);
-      data = await this.getGHTree(oct, token, sha, count, {
-        recursive: false,
-        callback: () => {
-          if (progress) {
-            progress("List file: " + count.file);
-          }
-        },
-      });
-      if (data.truncated) {
-        this._truncatedFolders.push(parentPath);
-      }
-      output.push(...this.tree2Tree(data.tree, parentPath));
-    } catch (error) {
-      const status = (error as { status?: number }).status;
-      if (status === 409) {
-        logger.debug("getTree empty repo", serializeError(error));
-        throw new AnonymousError("repo_empty", {
-          httpStatus: 409,
-          object: this.data,
-          cause: error as Error,
-        });
-      }
-      if (status === 404) {
-        logger.debug("getTree miss", serializeError(error));
-        const code = await classifyGitHubMissError(error, this.data);
-        throw new AnonymousError(code, {
-          httpStatus: 404,
-          object: this.data,
-          cause: error as Error,
-        });
-      }
-      logger.warn("getTree failed", serializeError(error));
-      throw new AnonymousError("repo_not_found", {
-        httpStatus: status || 404,
-        object: this.data,
-        cause: error as Error,
-      });
-    }
+    let data = await this.getGHTree(oct, token, sha, count, {
+      recursive: true,
+      callback: () => { progress?.("List file: " + count.file); },
+    });
+    if (!data.truncated) return this.tree2Tree(data.tree, parentPath);
+    data = await this.getGHTree(oct, token, sha, count, {
+      recursive: false,
+      callback: () => { progress?.("List file: " + count.file); },
+    });
+    if (data.truncated) this._truncatedFolders.push(parentPath);
+    output.push(...this.tree2Tree(data.tree, parentPath));
     const subtrees: { sha: string; parentPath: string }[] = [];
     for (const file of data.tree) {
       if (file.type == "tree" && file.path && file.sha) {

src/core/source/github-content-miss.ts

@@ -0,0 +1,65 @@
+import { createHash } from "crypto";
+import AnonymousError from "../AnonymousError";
+import { ExpiringMap } from "../expiring-map";
+import { octokit, waitForTokenGate, RateLimitDelayError } from "../GitHubUtils";
+import { GitHubBaseData } from "./GitHubBase";
+
+type Miss = "file_not_found" | "repo_not_found" | "commit_not_found";
+const results = new ExpiringMap<Miss>(2048);
+const pending = new Map<string, Promise<Miss>>();
+
+function missKey(data: GitHubBaseData, token: string): string {
+  return createHash("sha256").update(JSON.stringify([
+    data.repoId, data.organization, data.repoName, data.commit, data.cacheGeneration, data.cacheRevision, token,
+  ])).digest("hex");
+}
+
+export function cachedContentMiss(data: GitHubBaseData, token: string): Miss | undefined {
+  return results.get(missKey(data, token));
+}
+
+/** A raw 404 can mean a missing file, lost repository access, or a removed revision. */
+export async function classifyContentMiss(data: GitHubBaseData, token: string): Promise<Miss> {
+  const key = missKey(data, token);
+  const cached = results.get(key);
+  if (cached) return cached;
+  const active = pending.get(key);
+  if (active) return active;
+  if (pending.size >= 64) throw new AnonymousError("cache_busy", { httpStatus: 503 });
+  const probe = (async (): Promise<Miss> => {
+    const oct = octokit(token);
+    const options = { owner: data.organization, repo: data.repoName, request: { timeout: 15_000 } };
+    try {
+      await waitForTokenGate(token);
+      await oct.repos.get(options);
+    } catch (error) {
+      if ((error as { status?: number }).status === 404) return "repo_not_found";
+      throw probeError(error);
+    }
+    try {
+      await waitForTokenGate(token);
+      // Only check that the revision resolves; do not download a potentially
+      // large commit diff just to classify a missing file.
+      await oct.repos.getCommit({ ...options, ref: data.commit || "HEAD", mediaType: { format: "sha" } });
+    } catch (error) {
+      const upstream = error as { status?: number; response?: { data?: { message?: string } } };
+      if (upstream.status === 404 || upstream.status === 409 || (upstream.status === 422 &&
+          /^No commit found for SHA\b/.test(upstream.response?.data?.message || ""))) return "commit_not_found";
+      throw probeError(error);
+    }
+    return "file_not_found";
+  })().then(result => { results.set(key, result, 30_000); return result; });
+  pending.set(key, probe);
+  try { return await probe; }
+  finally { if (pending.get(key) === probe) pending.delete(key); }
+}
+
+function probeError(error: unknown): AnonymousError {
+  if (error instanceof AnonymousError) return error;
+  if (error instanceof RateLimitDelayError) return new AnonymousError("github_rate_limit_exceeded", { httpStatus: 429, cause: error });
+  const status = (error as { status?: number }).status;
+  return new AnonymousError(status === 401 ? "token_expired" : status === 403 ? "repo_not_accessible" : "github_unavailable", {
+    httpStatus: status === 401 || status === 403 ? status : 502,
+    cause: error as Error,
+  });
+}

src/core/storage/FileSystem.ts

@@ -1,4 +1,4 @@
-import { appendArchiveEntry } from "../archive-entry";
+import { appendArchiveEntry, zipEntryOptions } from "../archive-entry";
 import config from "../../config";
 import * as fs from "fs";
 import { Extract } from "unzip-stream";
@@ -280,7 +280,7 @@ export default class FileSystem extends StorageBase {
           output = source.pipe(transformer);
           active = output;
         }
-        await appendArchiveEntry(archive, output, { name: filename.slice(dir ? dir.length + 1 : 0) });
+        await appendArchiveEntry(archive, output, zipEntryOptions(filename.slice(dir ? dir.length + 1 : 0), filename));
       }
       if (!stopped) await archive.finalize();
     })().catch(error => archive.destroy(error));

src/core/storage/S3.ts

@@ -1,4 +1,4 @@
-import { appendArchiveEntry } from "../archive-entry";
+import { appendArchiveEntry, zipEntryOptions } from "../archive-entry";
 import {
   GetObjectCommand,
   ListObjectsV2CommandOutput,
@@ -398,7 +398,7 @@ export default class S3Storage extends StorageBase {
             output = source.pipe(transformer);
             active = output;
           }
-          await appendArchiveEntry(archive, output, { name: filename.slice(dir.length) });
+          await appendArchiveEntry(archive, output, zipEntryOptions(filename.slice(dir.length), filename));
         }
         if (listing.IsTruncated && !next) throw new Error("Invalid S3 continuation token");
       } while (next && !stopped);

src/core/zipStream.ts

@@ -13,6 +13,7 @@ import {
 } from "./anonymize-utils";
 import { createLogger, serializeError } from "./logger";
 import { startStage } from "./request-monitoring";
+import { zipEntryOptions } from "./archive-entry";
 
 const logger = createLogger("zip-stream");
 
@@ -222,7 +223,7 @@ export async function streamAnonymizedZip(
           activeStreams.add(anonymizer);
           anonymizer.once("close", () => activeStreams.delete(anonymizer));
           const st = entry.pipe(anonymizer);
-          archive.append(st, { name: fileName });
+          archive.append(st, zipEntryOptions(fileName, entry.path));
         } catch (error) {
           fail(error as Error);
         }

src/server/index.ts

@@ -39,6 +39,7 @@ import config from "../config";
 import { resolveTrustProxy } from "./trustProxy";
 import { requestRateLimitKey } from "./rate-limit-key";
 import { createLogger, serializeError } from "../core/logger";
+import { serverOptions } from "./server-options";
 import { monitorRequests } from "../core/request-monitoring";
 import { startPerformanceMonitoring } from "../core/performance-monitoring";
 import { getAnonymizationPoolStats } from "../core/anonymization-pool";
@@ -113,6 +114,7 @@ function indexResponse(req: express.Request, res: express.Response) {
 export default async function start() {
   const app = express();
   app.use(monitorRequests("api"));
+  app.use(serverOptions);
   startPerformanceMonitoring("api", getAnonymizationPoolStats);
   app.set("query parser", "extended");
   app.use("/service", createReviewCapabilities(process.env.REVIEW_SERVICE_KEYS));

src/server/server-options.ts

@@ -0,0 +1,9 @@
+import { RequestHandler } from "express";
+
+/** The asterisk request target asks about the server, not a session or route. */
+export const serverOptions: RequestHandler = (req, res, next) => {
+  if (req.originalUrl !== "*") return next();
+  if (req.method !== "OPTIONS") { res.status(400).json({ error: "invalid_request" }); return; }
+  res.setHeader("Allow", "GET, HEAD, POST, PUT, PATCH, DELETE, OPTIONS");
+  res.status(204).end();
+};

test/error-codes.test.js

@@ -68,7 +68,9 @@ function extractBackendErrorCodes(files) {
     // Multi-line matching for AnonymousError calls that span lines
     // (e.g. ternary expressions where the string is on the next line)
     const multiLinePattern =
-      /AnonymousError\([\s\S]*?["']([a-zA-Z][a-zA-Z0-9]*(?:_[a-zA-Z0-9]+)+)["']/g;
+      // Stop at a closing parenthesis so a dynamic code cannot consume an
+      // unrelated string in a later statement, such as a monitoring label.
+      /AnonymousError\([^)]*?["']([a-zA-Z][a-zA-Z0-9]*(?:_[a-zA-Z0-9]+)+)["']/g;
     let m;
     while ((m = multiLinePattern.exec(content)) !== null) {
       const code = m[1];

test/fixtures/github-gate-cancellation.js

@@ -0,0 +1,61 @@
+const assert = require("node:assert/strict");
+const { Readable } = require("node:stream");
+const { setImmediate: nextTurn, setTimeout: delay } = require("node:timers/promises");
+const { setTimeout, clearTimeout } = require("node:timers");
+const redis = require("redis");
+let reads = 0;
+const resetAt = Date.now() + 10 * 60 * 1000;
+let readGate = async () => String(resetAt);
+const client = { isOpen: true, on() { return this; }, async connect() {},
+  async get() { reads++; return readGate(); }, destroy() { this.isOpen = false; } };
+redis.createClient = () => client;
+require("ts-node/register/transpile-only");
+const github = require("../../src/core/GitHubUtils");
+const GitHubStream = require("../../src/core/source/GitHubStream").default;
+const got = require("got");
+
+async function main() {
+  const preAborted = new AbortController(); preAborted.abort();
+  await assert.rejects(github.waitForTokenGate("token", preAborted.signal), { name: "AbortError" });
+  assert.equal(reads, 0);
+
+  const firstController = new AbortController(), secondController = new AbortController();
+  const first = github.waitForTokenGate("token", firstController.signal);
+  let secondSettled = false;
+  const second = github.waitForTokenGate("token", secondController.signal).finally(() => { secondSettled = true; });
+  await nextTurn(); firstController.abort();
+  await assert.rejects(first, { name: "AbortError" });
+  await delay(20); assert.equal(secondSettled, false);
+  secondController.abort(); await assert.rejects(second, { name: "AbortError" });
+
+  // A shared Redis lookup may remain pending. Cancelling one caller must
+  // release that caller without closing the shared client.
+  readGate = () => new Promise(() => {});
+  const lookupController = new AbortController();
+  const lookup = github.waitForTokenGate("token", lookupController.signal);
+  await nextTurn(); lookupController.abort();
+  await assert.rejects(lookup, { name: "AbortError" }); assert.equal(client.isOpen, true);
+
+  let entered;
+  const gateEntered = new Promise(resolve => { entered = resolve; });
+  readGate = async () => { entered(); return String(resetAt); };
+  let metadataCalls = 0;
+  github.octokit = () => ({ repos: { getContent: async () => { metadataCalls++; throw Error("unexpected metadata request"); } } });
+  got.stream = () => new Readable({ read() {
+    this.destroy(Object.assign(Error("private web 404"), { response: { statusCode: 404 } }));
+  } });
+  const source = new GitHubStream({ repoId: "gate-cancel", organization: "owner", repoName: "repo", commit: "pinned", getToken: () => "token" });
+  const input = source.deferredRawDownload("token", "file.bin");
+  const collected = (async () => { for await (const chunk of input) void chunk; })().catch(() => {});
+  await gateEntered; await nextTurn();
+  const closed = new Promise(resolve => input.once("close", resolve)); input.destroy();
+  await closed; await collected;
+  assert.equal(metadataCalls, 0); assert.equal(input.closed, true);
+}
+
+const watchdog = setTimeout(() => { console.error("Cancellation did not release the gate wait"); process.exit(1); }, 5000);
+watchdog.unref();
+main().then(() => { clearTimeout(watchdog); console.log("gate cancellations released"); },
+  error => { clearTimeout(watchdog); console.error(error); process.exit(1); });
+// Do not force a successful exit: any retained ten-minute timer must fail
+// the parent test's process timeout.

test/performance-regressions.test.js

@@ -127,6 +127,7 @@ describe("performance regressions", function () {
     [Object.assign(new Error("large"), { status: 422 }), "file_too_big", 422],
     [Object.assign(new Error("missing"), { httpStatus: 404 }), "file_not_found", 404],
     [Object.assign(new Error("reset"), { code: "ECONNRESET" }), "upstream_error", 502],
+    [new (require("../src/core/GitHubUtils").RateLimitDelayError)(Date.now() + 60000, "quota-key"), "github_rate_limit_exceeded", 429],
     [new (require("../src/core/AnonymousError").default)("file_not_accessible", { httpStatus: 403 }), "file_not_accessible", 403],
   ].entries()) {
     it(`preserves ${code}/${status} for failed downloads (${upstream.message})`, async () => {
@@ -136,6 +137,7 @@ describe("performance regressions", function () {
       let committed = false;
       stub(storage, "write", async (_repo, _path, stream) => { for await (const chunk of stream) void chunk; committed = true; });
       source.downloadWithFallback = async () => { setImmediate(() => input.destroy(upstream)); return input; };
+      stub(require("../src/core/source/github-content-miss"), "classifyContentMiss", async () => "file_not_found");
       try { await source.getFileContentCache("file.txt", source.data.repoId, () => ({ sha: "blob", size: 1 })); throw Error("expected failure"); }
       catch (error) {
         expect(error.message).to.equal(code); expect(error.httpStatus).to.equal(status);

test/production-failure-fixes.test.js

@@ -72,6 +72,7 @@ describe("production failure fixes", function () {
     now = 30; expect(cache.get("b")).to.equal(undefined); expect(cache.get("c")).to.equal(undefined);
   });
   it("caches confirmed file misses by credential and commit, without caching permission failures", async () => {
+    stub(require("../src/core/source/github-content-miss"), "classifyContentMiss", async () => "file_not_found");
     let token = "credential-a", attempts = 0, now = Date.now();
     stub(Date, "now", () => now);
     stub(storage, "fileInfo", async () => { throw Object.assign(Error("missing"), { code: "ENOENT" }); });

test/production-followup-fixes.test.js

@@ -0,0 +1,379 @@
+const { expect } = require("chai");
+const express = require("express");
+const session = require("express-session");
+const { Passport } = require("passport");
+const http = require("node:http");
+const { Readable } = require("node:stream");
+const { once } = require("node:events");
+const { randomBytes } = require("node:crypto");
+const { URL } = require("node:url");
+const { spawnSync } = require("node:child_process");
+const path = require("node:path");
+const archiver = require("archiver");
+const got = require("got");
+require("ts-node/register/transpile-only");
+const { serverOptions } = require("../src/server/server-options");
+const { serializeError } = require("../src/core/logger");
+const { repositoryFailureStatus } = require("../src/core/repository-failure-status");
+const AnonymousError = require("../src/core/AnonymousError").default;
+const github = require("../src/core/GitHubUtils");
+const GitHubStream = require("../src/core/source/GitHubStream").default;
+const FileSystem = require("../src/core/storage/FileSystem").default;
+const S3 = require("../src/core/storage/S3").default;
+const config = require("../src/config").default;
+const { AnonymizeTransformer } = require("../src/core/anonymize-utils");
+const { streamAnonymizedZip } = require("../src/core/zipStream");
+const { classifyContentMiss } = require("../src/core/source/github-content-miss");
+const storage = require("../src/core/storage").default;
+
+function request(server, path, method) {
+  return new Promise((resolve, reject) => {
+    http.request({ host: "127.0.0.1", port: server.address().port, path, method }, res => {
+      res.resume(); res.once("end", () => resolve({ status: res.statusCode, headers: res.headers }));
+      res.once("error", reject);
+    }).once("error", reject).end();
+  });
+}
+
+describe("production follow-up fixes", function () {
+  this.timeout(10000);
+  const restores = [];
+  function stub(object, key, value) {
+    const previous = object[key]; object[key] = value; restores.push(() => { object[key] = previous; });
+  }
+  afterEach(() => restores.splice(0).reverse().forEach(restore => restore()));
+
+  it("answers OPTIONS * before sessions while preserving route OPTIONS and authentication", async () => {
+    const app = express(), passport = new Passport();
+    let sessionRequests = 0;
+    app.use(serverOptions);
+    app.use((_req, _res, next) => { sessionRequests++; next(); });
+    app.use(session({ secret: "test-secret", resave: false, saveUninitialized: false }));
+    app.use(passport.initialize()); app.use(passport.session());
+    app.get("/protected", (req, res) => res.sendStatus(req.isAuthenticated() ? 200 : 401));
+    const server = await new Promise(resolve => { const s = app.listen(0, "127.0.0.1", () => resolve(s)); });
+    try {
+      const global = await request(server, "*", "OPTIONS");
+      expect(global.status).to.equal(204); expect(global.headers.allow).to.include("OPTIONS");
+      expect(global.headers["set-cookie"]).to.equal(undefined); expect(sessionRequests).to.equal(0);
+      expect((await request(server, "*", "GET")).status).to.equal(400); expect(sessionRequests).to.equal(0);
+      const route = await request(server, "/protected", "OPTIONS");
+      expect(route.status).to.equal(200); expect(route.headers.allow).to.equal("GET, HEAD");
+      expect((await request(server, "/protected", "GET")).status).to.equal(401);
+      expect(sessionRequests).to.equal(2);
+    } finally { server.closeAllConnections(); await new Promise(resolve => server.close(resolve)); }
+  });
+
+  function source(getTree, getRepo = async () => ({})) {
+    stub(github, "waitForTokenGate", async () => {});
+    stub(github, "octokit", () => ({ git: { getTree }, repos: { get: getRepo } }));
+    return new GitHubStream({ repoId: "followup", organization: "owner", repoName: "repo", commit: "pinned", getToken: () => "test-token" });
+  }
+  const invalidObject = () => Object.assign(Error("Invalid object requested"), {
+    status: 422, response: { data: { message: "Invalid object requested. SHA must identify a commit or a tree." } },
+  });
+  for (const [status, code, responseStatus] of [[401, "token_expired", 401], [403, "repo_not_accessible", 403],
+    [404, "commit_not_found", 404], [409, "repo_empty", 409], [500, "github_unavailable", 502],
+    [503, "github_unavailable", 502], [422, "github_unavailable", 502], [undefined, "github_unavailable", 502]]) {
+    it(`classifies tree failure ${status} without calling it a missing repository`, async () => {
+      const upstream = Object.assign(Error("upstream failure"), { status });
+      const s = source(async () => { throw upstream; });
+      const error = await s.getFiles().catch(error => error);
+      expect(error.message).to.equal(code); expect(error.httpStatus).to.equal(responseStatus);
+      expect(error.cause).to.equal(upstream); expect(s.data.commit).to.equal("pinned");
+    });
+  }
+  it("reports an invalid tree revision as a missing commit without changing the pinned snapshot", async () => {
+    const s = source(async () => { throw invalidObject(); });
+    const error = await s.getFiles().catch(error => error);
+    expect(error.message).to.equal("commit_not_found"); expect(error.httpStatus).to.equal(404);
+    expect(s.data.commit).to.equal("pinned"); expect(repositoryFailureStatus(error.message)).to.equal(404);
+  });
+  it("preserves quota errors from both the gate and tree request", async () => {
+    const quota = new AnonymousError("github_rate_limit_exceeded", { httpStatus: 429 });
+    const s = source(async () => { throw quota; });
+    expect(await s.getFiles().catch(error => error)).to.equal(quota);
+    github.waitForTokenGate = async () => { throw quota; };
+    expect(await s.getFiles().catch(error => error)).to.equal(quota);
+    const delay = new github.RateLimitDelayError(Date.now() + 60000, "quota-key");
+    github.waitForTokenGate = async () => { throw delay; };
+    expect(await s.getFiles().catch(error => error)).to.equal(delay);
+  });
+  it("classifies failures while traversing a truncated tree too", async () => {
+    const s = source(async ({ tree_sha }) => {
+      if (tree_sha === "pinned") return { data: { truncated: true, tree: [{ type: "tree", path: "folder", sha: "child" }] } };
+      throw Object.assign(Error("GitHub offline"), { status: 503 });
+    });
+    const error = await s.getFiles().catch(error => error);
+    expect(error.message).to.equal("github_unavailable"); expect(error.httpStatus).to.equal(502);
+  });
+  it("maps an empty repository consistently after its failure is persisted", () => {
+    expect(repositoryFailureStatus("repo_empty")).to.equal(409);
+  });
+  it("serializes HTTP errors without invoking a deprecated code getter", () => {
+    let reads = 0;
+    const error = Object.assign(Error("GitHub error"), { status: 422 });
+    Object.defineProperty(error, "code", { get() { reads++; throw Error("deprecated getter invoked"); } });
+    expect(serializeError(error)).to.include({ status: 422, message: "GitHub error" });
+    expect(reads).to.equal(0);
+    expect(serializeError(Object.assign(Error("reset"), { code: "ECONNRESET" }))).to.include({ code: "ECONNRESET" });
+  });
+
+  it("coalesces content-miss probes across paths and expires them so restored access is retried", async () => {
+    let now = Date.now(), repoCalls = 0, commitCalls = 0;
+    stub(Date, "now", () => now); stub(github, "waitForTokenGate", async () => {});
+    stub(github, "octokit", () => ({ repos: {
+      get: async () => { repoCalls++; }, getCommit: async options => { expect(options.mediaType.format).to.equal("sha"); commitCalls++; },
+    } }));
+    const data = { repoId: "miss-coalescing", organization: "owner", repoName: "repo", commit: "pinned" };
+    expect(await Promise.all(Array.from({ length: 20 }, () => classifyContentMiss(data, "a"))))
+      .to.deep.equal(Array(20).fill("file_not_found"));
+    expect(await classifyContentMiss(data, "a")).to.equal("file_not_found");
+    expect(repoCalls).to.equal(1); expect(commitCalls).to.equal(1);
+    await classifyContentMiss(data, "b"); expect(repoCalls).to.equal(2);
+    await classifyContentMiss({ ...data, commit: "new" }, "a"); expect(repoCalls).to.equal(3);
+    await classifyContentMiss({ ...data, cacheRevision: "new" }, "a"); expect(repoCalls).to.equal(4);
+    now += 30001; await classifyContentMiss(data, "a"); expect(repoCalls).to.equal(5);
+  });
+  for (const stage of ["repository", "commit"]) {
+    it(`reports a missing ${stage} instead of a missing file`, async () => {
+      let commits = 0;
+      stub(github, "waitForTokenGate", async () => {});
+      stub(github, "octokit", () => ({ repos: {
+        get: async () => { if (stage === "repository") throw Object.assign(Error("missing"), { status: 404 }); },
+        getCommit: async () => { commits++; throw Object.assign(Error("missing"), { status: 404 }); },
+      } }));
+      const data = { repoId: `missing-${stage}`, organization: "owner", repoName: "repo", commit: "pinned" };
+      expect(await classifyContentMiss(data, "a")).to.equal(stage === "repository" ? "repo_not_found" : "commit_not_found");
+      expect(commits).to.equal(stage === "repository" ? 0 : 1);
+    });
+  }
+  for (const status of [401, 403, 422, 503]) {
+    it(`does not cache a content-probe ${status} failure as a missing file`, async () => {
+      let calls = 0;
+      stub(github, "waitForTokenGate", async () => {});
+      stub(github, "octokit", () => ({ repos: { get: async () => { calls++; throw Object.assign(Error("failed"), { status }); } } }));
+      const data = { repoId: `probe-failure-${status}`, organization: "owner", repoName: "repo", commit: "pinned" };
+      for (let i = 0; i < 2; i++) {
+        const error = await classifyContentMiss(data, "a").catch(error => error);
+        expect(error.httpStatus).to.equal(status === 401 || status === 403 ? status : 502);
+      }
+      expect(calls).to.equal(2);
+    });
+  }
+  it("recognizes the GitHub invalid-commit response without retrying with HEAD", async () => {
+    let ref;
+    stub(github, "waitForTokenGate", async () => {});
+    stub(github, "octokit", () => ({ repos: {
+      get: async () => {}, getCommit: async options => {
+        ref = options.ref; throw Object.assign(Error("missing"), { status: 422, response: { data: { message: "No commit found for SHA: pinned" } } });
+      },
+    } }));
+    expect(await classifyContentMiss({ repoId: "invalid-commit-probe", organization: "owner", repoName: "repo", commit: "pinned" }, "a"))
+      .to.equal("commit_not_found");
+    expect(ref).to.equal("pinned");
+  });
+
+  function rawFallback(downloadUrl, content) {
+    const calls = [];
+    stub(github, "waitForTokenGate", async () => {});
+    stub(github, "octokit", () => ({ repos: { getContent: async options => {
+      expect(options).to.include({ owner: "owner", repo: "repo", path: "file.bin", ref: "pinned" });
+      return { data: { type: "file", download_url: downloadUrl } };
+    } } }));
+    stub(got, "stream", (url, options) => {
+      calls.push({ url, options });
+      if (calls.length === 1) return new Readable({ read() { this.destroy(Object.assign(Error("private web 404"), { response: { statusCode: 404 } })); } });
+      return Readable.from(Array.isArray(content) ? content : [content]);
+    });
+    const s = new GitHubStream({ repoId: "private-fallback", organization: "owner", repoName: "repo", commit: "pinned", getToken: () => "owner-secret" });
+    return { stream: s.downloadFileViaRaw("owner-secret", "file.bin"), calls };
+  }
+  it("recovers an accessible private file from the API download URL without forwarding the owner token", async () => {
+    const url = "https://raw.githubusercontent.com/owner/repo/pinned/file.bin?token=short-lived";
+    const { stream, calls } = rawFallback(url, Buffer.from("actual content"));
+    expect((await collect(stream)).toString()).to.equal("actual content");
+    expect(calls).to.have.length(2); expect(calls[1].url).to.equal(url);
+    expect(calls[1].options.headers).to.equal(undefined);
+    expect(() => calls[1].options.hooks.beforeRedirect[0]({ url: new URL("https://attacker.test/file") })).to.throw("upstream_error");
+    expect(serializeError(Object.assign(Error("failed"), { response: { url } })).url).not.to.include("short-lived");
+  });
+  for (const url of ["http://raw.githubusercontent.com/file", "https://attacker.test/file", "https://owner-secret@raw.githubusercontent.com/file"]) {
+    it(`rejects an unsafe contents download URL ${new URL(url).hostname}`, async () => {
+      const { stream, calls } = rawFallback(url, Buffer.from("should never be read"));
+      const error = await collect(stream).catch(error => error);
+      expect(error.message).to.equal("upstream_error"); expect(error.httpStatus).to.equal(502); expect(calls).to.have.length(1);
+    });
+  }
+  for (const mode of ["short", "long", "fragmented"]) {
+    it(`does not serve a ${mode} unresolved LFS pointer as file content`, async () => {
+      const extension = mode === "short" ? "" : "ext-0-test sha256:" + "b".repeat(64) + "\n";
+      const pointer = Buffer.from("version https://git-lfs.github.com/spec/v1\n" + extension + "oid sha256:" + "a".repeat(64) + "\nsize 10\n");
+      const chunks = mode === "fragmented" ? [pointer.subarray(0, 20), pointer.subarray(20, 150), pointer.subarray(150)] : pointer;
+      const { stream } = rawFallback("https://raw.githubusercontent.com/owner/repo/pinned/file.bin", chunks);
+      const error = await collect(stream).catch(error => error);
+      expect(error.message).to.equal("upstream_error"); expect(error.httpStatus).to.equal(502);
+    });
+  }
+  for (const mode of ["single", "fragmented"]) {
+    it(`resolves a long blob LFS pointer with ${mode} delivery instead of forwarding its text`, async () => {
+      const pointer = Buffer.from("version https://git-lfs.github.com/spec/v1\next-0-test sha256:" + "b".repeat(64) + "\noid sha256:" + "a".repeat(64) + "\nsize 10\n");
+      const s = new GitHubStream({ repoId: "long-lfs", organization: "owner", repoName: "repo", commit: "pinned", getToken: () => "token" });
+      let downloads = 0;
+      s.downloadFileViaRaw = () => { downloads++; return Readable.from(["resolved file"]); };
+      const chunks = mode === "fragmented" ? [pointer.subarray(0, 20), pointer.subarray(20, 150), pointer.subarray(150)] : [pointer];
+      expect((await collect(s.resolveLfsPointer(Readable.from(chunks), "token", "file.bin"))).toString()).to.equal("resolved file");
+      expect(downloads).to.equal(1);
+    });
+  }
+  it("releases cancelled gate timers and private streams without retaining process handles", function () {
+    this.timeout(20000);
+    const result = spawnSync(process.execPath, [path.join(__dirname, "fixtures/github-gate-cancellation.js")], {
+      timeout: 15000, encoding: "utf8", env: { ...process.env, NODE_ENV: "test", GITHUB_APP_ENABLED: "false", REDIS_HOSTNAME: "127.0.0.1", REDIS_PORT: "1" },
+    });
+    expect(result.error, result.stderr).to.equal(undefined);
+    expect(result.status, result.stderr + result.stdout).to.equal(0);
+    expect(result.stdout).to.include("gate cancellations released");
+  });
+  it("keeps public raw misses off the contents API", async () => {
+    stub(github, "octokit", () => { throw Error("unexpected anonymous REST request"); });
+    stub(got, "stream", () => new Readable({ read() { this.destroy(Object.assign(Error("missing"), { response: { statusCode: 404 } })); } }));
+    const s = new GitHubStream({ repoId: "public-raw", organization: "owner", repoName: "repo", commit: "pinned", getToken: () => "" });
+    const error = await collect(s.downloadFileViaRaw("", "file.bin")).catch(error => error);
+    expect(error.response.statusCode).to.equal(404);
+  });
+  it("does not start a private download after cancellation during the contents lookup", async () => {
+    let resolveMetadata, started, calls = 0;
+    const metadata = new Promise(resolve => { resolveMetadata = resolve; });
+    const lookupStarted = new Promise(resolve => { started = resolve; });
+    stub(github, "waitForTokenGate", async () => {});
+    stub(github, "octokit", () => ({ repos: { getContent: async () => { started(); return metadata; } } }));
+    stub(got, "stream", () => {
+      calls++; return new Readable({ read() { this.destroy(Object.assign(Error("missing"), { response: { statusCode: 404 } })); } });
+    });
+    const s = new GitHubStream({ repoId: "cancel-private", organization: "owner", repoName: "repo", commit: "pinned", getToken: () => "token" });
+    const input = s.downloadFileViaRaw("token", "file.bin");
+    const collected = collect(input).catch(error => error);
+    await lookupStarted; input.destroy();
+    resolveMetadata({ data: { type: "file", download_url: "https://raw.githubusercontent.com/owner/repo/pinned/file.bin" } });
+    await collected; expect(calls).to.equal(1);
+  });
+  for (const mode of ["raw", "lfs"]) {
+    it(`destroys a pending ${mode} upstream immediately when its consumer cancels`, async () => {
+      const { PassThrough } = require("node:stream");
+      const upstream = new PassThrough();
+      let started;
+      const ready = new Promise(resolve => { started = resolve; });
+      stub(got, "stream", () => { started(); return upstream; });
+      const s = new GitHubStream({ repoId: "pending-raw", organization: "owner", repoName: "repo", commit: "pinned", getToken: () => "token" });
+      const pointer = "version https://git-lfs.github.com/spec/v1\noid sha256:" + "a".repeat(64) + "\nsize 10\n";
+      const input = mode === "raw" ? s.deferredRawDownload("token", "file.bin")
+        : s.resolveLfsPointer(Readable.from([pointer]), "token", "file.bin");
+      const collected = collect(input).catch(error => error);
+      await ready; input.destroy();
+      expect(upstream.destroyed).to.equal(true);
+      await collected;
+    });
+  }
+  it("closes a real upstream HTTP socket when a nested raw stream is cancelled", async () => {
+    let respond, closed;
+    const responseStarted = new Promise(resolve => { respond = resolve; });
+    const socketClosed = new Promise(resolve => { closed = resolve; });
+    const server = http.createServer((req, res) => {
+      req.socket.once("close", closed); res.write("partial content");
+    });
+    await new Promise(resolve => server.listen(0, "127.0.0.1", resolve));
+    const originalStream = got.stream;
+    stub(got, "stream", () => {
+      const input = originalStream(`http://127.0.0.1:${server.address().port}`, { retry: { limit: 0 } });
+      input.once("response", respond); return input;
+    });
+    const s = new GitHubStream({ repoId: "cancel-socket", organization: "owner", repoName: "repo", commit: "pinned", getToken: () => "token" });
+    const input = s.deferredRawDownload("token", "file.bin");
+    const collected = collect(input).catch(error => error);
+    try {
+      await responseStarted; input.destroy();
+      await socketClosed; await collected;
+    } finally { input.destroy(); server.closeAllConnections(); await new Promise(resolve => server.close(resolve)); }
+  });
+  it("avoids file downloads for a known unavailable source and retries after the short TTL", async () => {
+    let now = Date.now(), attempts = 0;
+    stub(Date, "now", () => now); stub(github, "waitForTokenGate", async () => {});
+    stub(github, "octokit", () => ({ repos: { get: async () => { throw Object.assign(Error("missing repository"), { status: 404 }); } } }));
+    stub(storage, "fileInfo", async () => { throw Object.assign(Error("missing cache"), { code: "ENOENT" }); });
+    const s = new GitHubStream({ repoId: "source-miss-shortcut", organization: "owner", repoName: "repo", commit: "pinned", getToken: () => "token" });
+    s.downloadWithFallback = async () => { attempts++; throw Object.assign(Error("missing"), { response: { statusCode: 404 } }); };
+    for (const path of ["first", "second", "third"]) {
+      const error = await s.getFileContentCache(path, s.data.repoId, () => ({ sha: path })).catch(error => error);
+      expect(error.message).to.equal("repo_not_found"); expect(error.httpStatus).to.equal(404);
+    }
+    expect(attempts).to.equal(1);
+    now += 30001;
+    await s.getFileContentCache("fourth", s.data.repoId, () => ({ sha: "fourth" })).catch(() => {});
+    expect(attempts).to.equal(2);
+  });
+
+  // Read the central directory to check the actual compression methods,
+  // rather than just asserting the options passed to archiver.
+  function compressionMethods(zip) {
+    const end = zip.length - 22;
+    expect(zip.readUInt32LE(end)).to.equal(0x06054b50);
+    const methods = {};
+    let offset = zip.readUInt32LE(end + 16);
+    for (let i = 0; i < zip.readUInt16LE(end + 10); i++) {
+      expect(zip.readUInt32LE(offset)).to.equal(0x02014b50);
+      const nameLength = zip.readUInt16LE(offset + 28);
+      const name = zip.subarray(offset + 46, offset + 46 + nameLength).toString();
+      methods[name] = zip.readUInt16LE(offset + 10);
+      offset += 46 + nameLength + zip.readUInt16LE(offset + 30) + zip.readUInt16LE(offset + 32);
+    }
+    return methods;
+  }
+  async function collect(input) {
+    const chunks = []; for await (const chunk of input) chunks.push(chunk); return Buffer.concat(chunks);
+  }
+  async function unpack(zip) {
+    const files = {}, entries = [], parser = require("unzip-stream").Parse();
+    parser.on("entry", entry => { entries.push(collect(entry).then(body => { files[entry.path] = body; })); });
+    const done = once(parser, "finish"); Readable.from([zip]).pipe(parser);
+    await done; await Promise.all(entries); return files;
+  }
+  const binary = Buffer.concat([Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]), randomBytes(128 * 1024)]);
+  const contents = { "private.PNG": binary, "README.md": Buffer.from("private source content"),
+    "figure.svg": Buffer.from('<svg><text>private</text></svg>') };
+  const options = { terms: ["private"], image: true, pdf: true, link: true };
+  for (const backend of ["filesystem", "s3", "github"]) {
+    it(`avoids recompressing binary assets in ${backend} ZIPs and still anonymizes text`, async () => {
+      let zip;
+      if (backend === "github") {
+        const fixture = archiver("zip");
+        const collected = collect(fixture);
+        for (const [name, body] of Object.entries(contents)) fixture.append(body, { name: `repo/${name}` });
+        await fixture.finalize(); const input = await collected;
+        stub(got, "stream", () => Readable.from([input]));
+        const { PassThrough } = require("node:stream");
+        const response = new PassThrough(); const output = collect(response);
+        await streamAnonymizedZip({ repoId: "test", organization: "owner", repoName: "public", commit: "pinned",
+          getToken: () => "", anonymizerOptions: options }, response);
+        zip = await output;
+      } else {
+        if (backend === "s3") stub(config, "S3_BUCKET", "test-bucket");
+        const storage = backend === "filesystem" ? new FileSystem() : new S3();
+        const names = Object.keys(contents);
+        storage.read = async (_repo, name) => Readable.from([contents[name]]);
+        if (backend === "filesystem") storage.listFiles = async () => names.map(name => ({ name, path: "", size: contents[name].length }));
+        else {
+          storage.client = () => ({ listObjectsV2: async () => ({ Contents: names.map(name => ({ Key: storage.repoPath("test") + name })) }) });
+        }
+        zip = await collect(await storage.archive("test", "", {
+          fileTransformer: name => new AnonymizeTransformer({ ...options, filePath: name }),
+        }));
+      }
+      const methods = compressionMethods(zip), files = await unpack(zip);
+      const binaryName = backend === "github" ? "XXXX-1.PNG" : "private.PNG";
+      expect(methods[binaryName]).to.equal(0); expect(files[binaryName]).to.deep.equal(binary);
+      expect(methods["README.md"]).to.equal(8); expect(files["README.md"].toString()).to.equal("XXXX-1 source content");
+      expect(methods["figure.svg"]).to.equal(8); expect(files["figure.svg"].toString()).to.include("XXXX-1");
+    });
+  }
+});

test/request-monitoring.test.js

@@ -95,6 +95,19 @@ describe("bounded request monitoring", function () {
     expect(metrics.drain().find(row => row.data.metric === "request").data.bytes).to.equal(0);
   });
 
+  it("reports time after headers and delivery rate without conflating them with startup time", async function () {
+    app.get("/api/repo/:repoId/file/*path", async (_req, res) => {
+      await delay(30); res.flushHeaders(); await delay(40); res.end(Buffer.alloc(4096));
+    });
+    await listen();
+    await fetch(server, "/api/repo/test/file/asset.mp4");
+    const detail = logs[0][1];
+    expect(detail.firstByteMs).to.be.at.least(20);
+    expect(detail.responseMs).to.be.at.least(30);
+    expect(detail.ms).to.be.closeTo(detail.firstByteMs + detail.responseMs, 1);
+    expect(detail.bytesPerSecond).to.be.closeTo(detail.bytes * 1000 / detail.responseMs, detail.bytes * 1000 / detail.responseMs * .05);
+  });
+
   for (const headersSent of [false, true]) {
     it(`records a disconnect ${headersSent ? "after" : "before"} headers exactly once`, async function () {
       let arrived; const ready = new Promise(resolve => { arrived = resolve; });
@@ -107,7 +120,11 @@ describe("bounded request monitoring", function () {
       expect(logs[0][1].outcome).to.equal("interrupted");
       const row = metrics.drain().find(row => row.data.metric === "request").data;
       expect(row.count).to.equal(1); expect(row.aborted).to.equal(1); expect(activeRequestCount()).to.equal(0);
-      if (!headersSent) expect(logs[0][1].firstByteMs).to.equal(null);
+      if (!headersSent) {
+        expect(logs[0][1].firstByteMs).to.equal(null);
+        expect(logs[0][1].responseMs).to.equal(null);
+        expect(logs[0][1].bytesPerSecond).to.equal(null);
+      }
     });
   }