perf: serve cached OAuth snapshots without GitHub calls

src/core/AnonymizedFile.ts

@@ -1,3 +1,5 @@
+import { canReadCachedRepository } from "./cached-repository-access";
+import { probeStreamerCache } from "./streamer-cache-probe";
 import { notifyOwnerAccessProblem } from "./owner-notifications";
 import { sendSharedContent } from "./cached-response";
 import { streamResponse } from "./response-stream";
@@ -458,37 +460,48 @@ export default class AnonymizedFile {
       try {
         if (config.STREAMER_ENTRYPOINT) {
           // use the streamer service
-          const [sha, size, token] = await Promise.all([
-            this.sha(),
-            this.size(),
-            this.repository.getToken(),
-          ]);
+          const [sha, size] = await Promise.all([this.sha(), this.size()]);
+          const payload = {
+            sha,
+            size,
+            upstream: (await this.getFileInfo()).upstream,
+            repoFullName: this.repository.model.source.repositoryName,
+            commit: this.repository.model.source.commit,
+            cacheGeneration: `${this.repository.model.treeGeneration || "legacy"}:${this.repository.model.anonymizeDate?.toISOString() || ""}`,
+            cacheRevision: this.repository.model.contentCacheRevision,
+            branch: this.repository.model.source.branch,
+            repoId: this.repository.repoId,
+            filePath: this.filePath,
+            anonymizerOptions: anonymizer.opt,
+          };
+          const url = join(config.STREAMER_ENTRYPOINT, "api");
+          const headers = { ...requestHeaders(), "accept-encoding": res.req?.get?.("accept-encoding") || "identity" };
           const headersDone = startStage("streamer_headers");
-          const resStream = got
-            .stream(join(config.STREAMER_ENTRYPOINT, "api"), {
-              method: "POST",
-              headers: { ...requestHeaders(), "accept-encoding": res.req?.get?.("accept-encoding") || "identity" },
-              decompress: false,
-              json: {
-                sha,
-                size,
-                upstream: (await this.getFileInfo()).upstream,
-                token: await githubTokenForStreamer(token, this.repository.model.source.repositoryName),
-                repoFullName: this.repository.model.source.repositoryName,
-                commit: this.repository.model.source.commit,
-        cacheGeneration: `${this.repository.model.treeGeneration || "legacy"}:${this.repository.model.anonymizeDate?.toISOString() || ""}`,
-                cacheRevision: this.repository.model.contentCacheRevision,
-                branch: this.repository.model.source.branch,
-                repoId: this.repository.repoId,
-                filePath: this.filePath,
-                anonymizerOptions: anonymizer.opt,
-              },
-            });
+          const probe = async () => {
+            if (!config.SHARED_CACHE_ENABLED || !await canReadCachedRepository(this.repository)) return undefined;
+            const cancellation = new AbortController();
+            const cancel = () => cancellation.abort();
+            res.once("close", cancel);
+            if (res.destroyed || res.writableEnded) cancel();
+            try { return await probeStreamerCache(url, payload, headers, cancellation.signal); }
+            finally { res.removeListener("close", cancel); }
+          };
+          const cached = await probe();
+          if (cached) {
+            try {
+              if (!await canReadCachedRepository(this.repository)) throw new AnonymousError("connection_changed", { httpStatus: 409 });
+            } catch (error) { cached.stream.destroy(); throw error; }
+          }
+          if (res.destroyed || res.writableEnded) { cached?.stream.destroy(); headersDone(); resolve(); return; }
+          const resStream = cached?.stream || got.stream(url, {
+            method: "POST", headers, decompress: false,
+            json: { ...payload, token: await githubTokenForStreamer(await this.repository.getToken(), this.repository.model.source.repositoryName) },
+          });
           // Forward Content-Type from the streamer's upstream response.
           // got.stream(...).pipe(res) forwards body bytes only — without
           // this, the parent response has no Content-Type and the browser
           // guesses (text renders as download, images as octet-stream).
-          resStream.on("response", (upstream: { headers: Record<string, string | string[] | undefined> }) => {
+          const forwardHeaders = (upstream: { headers: Record<string, string | string[] | undefined> }) => {
             headersDone();
             if (res.headersSent) return;
             for (const header of ["content-encoding", "content-length", "vary", "accept-ranges"]) {
@@ -503,7 +516,9 @@ export default class AnonymizedFile {
               if (fallback) res.contentType(fallback);
               else if (isTextFile(this.anonymizedPath)) res.contentType("text/plain");
             }
-          });
+          };
+          if (cached) forwardHeaders(cached.response);
+          else resStream.on("response", forwardHeaders);
           resStream.once("error", () => headersDone(true));
           resStream.once("close", () => headersDone());
           await streamResponse(resStream, res, (err: Error) => {

src/core/cached-repository-access.ts

@@ -0,0 +1,36 @@
+import Repository from "./Repository";
+import AnonymousError from "./AnonymousError";
+import { isConnected } from "../server/database";
+import RepositoryModel from "./model/anonymizedRepositories/anonymizedRepositories.model";
+import UserModel from "./model/users/users.model";
+import CredentialModel from "./model/credentials/credentials.model";
+import config from "../config";
+import { APP_PROVIDER, appError } from "./github-app";
+
+/** Local access to an already published snapshot, not permission to fetch upstream. */
+export async function canReadCachedRepository(repository: Repository): Promise<boolean> {
+  // Unpersisted/test objects must use the existing authorization path.
+  if (!isConnected || repository.model.isNew) return false;
+  await repository.check();
+  const current = await RepositoryModel.findById(repository.model._id).select("owner githubAccess").lean();
+  if (!current || String(current.owner) !== repository.owner.id ||
+      current.githubAccess?.revision !== repository.model.githubAccess?.revision) {
+    throw new AnonymousError("connection_changed", { httpStatus: 409 });
+  }
+  const user = await UserModel.findById(repository.owner.id).select("status").lean();
+  if (!user || user.status === "removed" || user.status === "banned") {
+    throw new AnonymousError("not_authorized", { httpStatus: 403 });
+  }
+  const binding = current.githubAccess;
+  const grant = await CredentialModel.findOne({ ownerId: repository.owner.id,
+    provider: binding?.kind === "github-app" ? APP_PROVIDER : "github" }).select("revoked").lean();
+  if (grant?.revoked) throw new AnonymousError("not_authorized", { httpStatus: 403 });
+  if (binding?.kind === "github-app") {
+    if (!config.GITHUB_APP_ENABLED || !grant) throw appError();
+    // Public visibility and user membership changes are not fully covered by
+    // local webhook state. Keep remote authorization for these bindings until
+    // a durable per-repository permission reconciliation mechanism exists.
+    return false;
+  }
+  return true;
+}

src/core/source/GitHubStream.ts

@@ -337,7 +337,8 @@ export default class GitHubStream extends GitHubBase {
       | Promise<{ sha: string; size?: number; upstream?: UpstreamFile }>
       | { sha: string; size?: number; upstream?: UpstreamFile }
       | Promise<string>
-      | string
+      | string,
+    cacheOnly = false
   ) {
     const meta = await fileMeta();
     const expected: { sha: string; size?: number; upstream?: UpstreamFile } =
@@ -396,6 +397,23 @@ export default class GitHubStream extends GitHubBase {
       }
     };
     const shared = sharedBlobCache();
+    // A cache-only probe must never fill, promote legacy data or contact GitHub.
+    // The API has checked local access; lifecycle/revision fences still apply here.
+    if (cacheOnly) {
+      await assertActive();
+      if (expected.size != null && expected.size > config.MAX_FILE_SIZE) throw new AnonymousError("file_too_big", { httpStatus: 413 });
+      const hit = /^[a-f0-9]{40}$/.test(expected.sha) ? await shared?.lookup(`git-${expected.sha}`) : undefined;
+      if (!hit) throw new AnonymousError("cache_miss", { httpStatus: 409 });
+      // Resolved LFS payloads may need additional billing admission. Let the
+      // normal path reserve that expansion before serving it.
+      if (config.BILLING_ENABLED && hit.cachedBlob!.record.size > (expected.size || 0)) {
+        hit.destroy(); throw new AnonymousError("cache_miss", { httpStatus: 409 });
+      }
+      try { await assertActive(); } catch (error) { hit.destroy(); throw error; }
+      observeCache("source_hit", hit.cachedBlob!.record.size);
+      recordCacheHit("source", true);
+      return hit;
+    }
     if (shared && /^[a-f0-9]{40}$/.test(expected.sha)) {
       await assertActive();
       if (expected.size != null && expected.size > config.MAX_FILE_SIZE) throw new AnonymousError("file_too_big", { httpStatus: 413 });

src/core/streamer-cache-probe.ts

@@ -0,0 +1,24 @@
+import got, { Response } from "got";
+
+/** Wait only for headers; the body remains a backpressured stream. */
+export async function probeStreamerCache(url: string, json: Record<string, unknown>, headers: Record<string, string>, signal?: AbortSignal) {
+  const stream = got.stream(url, { method: "POST", json: { ...json, token: "", cacheOnly: true },
+    headers, timeout: { response: 15000 }, decompress: false });
+  const responseReady = new Promise<Response>((resolve, reject) => {
+    stream.once("response", resolve);
+    stream.once("error", reject);
+  });
+  const abort = () => stream.destroy(new Error("response_cancelled"));
+  signal?.addEventListener("abort", abort, { once: true });
+  stream.once("close", () => signal?.removeEventListener("abort", abort));
+  if (signal?.aborted) abort();
+  try {
+    const response = await responseReady;
+    return { stream, response };
+  } catch (error) {
+    stream.destroy();
+    const response = (error as { response?: Response }).response;
+    if (response?.statusCode === 409 && response.headers["x-anonymous-cache-miss"] === "1") return undefined;
+    throw error;
+  }
+}

src/streamer/route.ts

@@ -91,7 +91,8 @@ router.post("/", async (req: express.Request, res: express.Response) => {
     const content = await source.getFileContentCache(
       filePath,
       repoId,
-      () => ({ sha: fileSha, size: fileSize, upstream: req.body.upstream })
+      () => ({ sha: fileSha, size: fileSize, upstream: req.body.upstream }),
+      req.body.cacheOnly === true
     );
     if (res.destroyed || res.writableEnded) { content.destroy(); anonymizer.destroy(); return; }
     const mime = lookup(filePath);
@@ -136,6 +137,12 @@ router.post("/", async (req: express.Request, res: express.Response) => {
         }
       });
   } catch (error) {
+    anonymizer.destroy();
+    if (req.body.cacheOnly === true && error instanceof Error && error.message === "cache_miss") {
+      res.setHeader("X-Anonymous-Cache-Miss", "1");
+      res.status(409).end();
+      return;
+    }
     handleError(error, res);
   }
 });