fix: handle production source failures and reduce ZIP work

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);
+        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", 0, 39) === "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;
   }
 
@@ -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) {