fix: bound metadata work and distinguish cancelled requests (#951)

src/core/AnonymizedFile.ts

@@ -1,3 +1,4 @@
+import { isResponseCancellation } from "./response-cancellation";
 import { canReadCachedRepository } from "./cached-repository-access";
 import { probeStreamerCache } from "./streamer-cache-probe";
 import { notifyOwnerAccessProblem } from "./owner-notifications";
@@ -485,6 +486,7 @@ export default class AnonymizedFile {
             if (res.destroyed || res.writableEnded) cancel();
             try { return await probeStreamerCache(url, payload, headers, cancellation.signal); }
             catch (err) {
+              if (isResponseCancellation(err, res)) { headersDone(); return undefined; }
               headersDone(true);
               const { error } = streamerErrorToAnonymous(err as Error & {
                 response?: { statusCode?: number; body?: unknown };
@@ -531,6 +533,7 @@ export default class AnonymizedFile {
           resStream.once("error", () => headersDone(true));
           resStream.once("close", () => headersDone());
           await streamResponse(resStream, res, (err: Error) => {
+              if (isResponseCancellation(err, res)) return;
               const { error } = streamerErrorToAnonymous(
                 err as Error & {
                   response?: { statusCode?: number; body?: unknown };

src/core/GitHubUtils.ts

@@ -1,3 +1,4 @@
+import { InvalidCredentialBackoff } from "./invalid-credential-backoff";
 import { isGitHubRateLimitError, OctokitRequestErrorLike } from "./github-rate-limit";
 import { CredentialValidation } from "./credential-validation";
 import { refreshLegacyToken } from "./legacy-token-refresh";
@@ -202,6 +203,8 @@ export async function getRedisGateResetAt(tokenKey: string): Promise<number> {
   }
 }
 
+const invalidCredentials = new InvalidCredentialBackoff();
+
 export function octokit(token: string) {
   const context = githubTokenContext(token);
   const oct = new ThrottledOctokit({
@@ -272,7 +275,11 @@ export function octokit(token: string) {
     }
     throw err;
   });
-  oct.hook.wrap("request", (request, options) => measureStage("upstream", async () => request(options)));
+  oct.hook.wrap("request", (request, options) => {
+    const run = () => measureStage("upstream", async () => request(options));
+    // Managed handles may rotate during renewal; never poison a handle with an old token's denial.
+    return githubTokenContext(token) ? run() : invalidCredentials.run(token, run);
+  });
   return oct;
 }
 
@@ -311,7 +318,7 @@ const checkedRepositoryTokens = new WeakMap<Repository, string>();
 
 export async function getToken(repository: Repository) {
   try {
-    return await measureStage("authorization", () => resolveRepositoryToken(repository));
+    return await measureStage("credential_resolution", () => resolveRepositoryToken(repository));
   } catch (error) {
     void notifyOwnerAccessProblem(repository.owner.id, error, { kind: "repository", id: String(repository.model._id) });
     throw error;

src/core/Repository.ts

@@ -1,7 +1,7 @@
 import { REPOSITORY_ID_COLLATION } from "./repository-id";
 import { reserveRepository } from "./billing/quotas";
 import { hasAutomaticUpdates } from "./billing/expiration";
-import { FILE_TREE_BATCH_SIZE, FILE_LISTING_LIMIT, deleteRowsInBatches } from "./file-tree";
+import { FILE_TREE_BATCH_SIZE, FILE_LISTING_LIMIT, deleteRowsInBatches, clearStagedRowsInBatches } from "./file-tree";
 import { SourceBase, sourceStorageSize } from "./source/Source";
 import { notifyOwnerAccessProblem, withOwnerAccessRecovery } from "./owner-notifications";
 import AnonymizedPathModel from "./model/anonymized-path";
@@ -341,8 +341,7 @@ export default class Repository {
       this.model.treeGeneration = generation;
       if (isConnected) {
         // A crash here is safe: maintenance recognizes the active generation.
-        await FileModel.updateMany({ repoId: this.repoId, treeGeneration: generation, treeStaged: true },
-          { $unset: { treeStaged: "" } }).exec().catch(error => logger.warn("staged tree completion deferred", serializeError(error)));
+        await clearStagedRowsInBatches(FileModel.collection, this.repoId, generation).catch(error => logger.warn("staged tree completion deferred", serializeError(error)));
       }
       this.model.emptyTreeGeneration = fileCount ? undefined : generation;
       this.model.truncatedFolders = truncatedFolders;
@@ -434,7 +433,7 @@ export default class Repository {
     const terminal = current?.status && [RepositoryStatus.EXPIRING, RepositoryStatus.EXPIRED,
       RepositoryStatus.REMOVING, RepositoryStatus.REMOVED].includes(current.status);
     if (current?.treeGeneration === generation && !terminal) {
-      await FileModel.updateMany({ repoId, treeGeneration: generation, treeStaged: true }, { $unset: { treeStaged: "" } }).exec();
+      await clearStagedRowsInBatches(FileModel.collection, repoId, generation);
     } else if (terminal || !current?.stagedFileTrees?.some(stage => stage.generation === generation && stage.until > now)) {
       await deleteRowsInBatches(FileModel.collection, { repoId, treeGeneration: generation, treeStaged: true });
     }

src/core/file-submodules.ts

@@ -0,0 +1,15 @@
+import FileModel from "./model/files/files.model";
+
+export async function hasVisibleSubmodules(repoId: string, generation?: string): Promise<boolean> {
+  const scope = { repoId, treeGeneration: generation || { $exists: false } };
+  // Separate probes prevent the OR planner choosing a scan of every generation row.
+  if (await FileModel.exists({ ...scope, name: ".gitmodules", path: "" })) return true;
+  const filter = { ...scope, submodule: { $exists: true, $ne: "included" } };
+  try { return !!await FileModel.exists(filter).hint("file_submodule_status"); }
+  catch (error) {
+    // Rolling upgrades may receive requests before the additive index is installed.
+    const failure = error as { code?: number; message?: string };
+    if (failure.code !== 2 || !/hint.*(?:index|existing)/i.test(failure.message || "")) throw error;
+    return !!await FileModel.exists(filter);
+  }
+}

src/core/file-tree.ts

@@ -30,3 +30,18 @@ export async function deleteRowsInBatches(
   logger.info("metadata cleanup deferred", { repoId: filter.repoId, deleted, ms: Date.now() - started });
   return false;
 }
+
+/** Staging markers are bookkeeping; leave the remainder discoverable by maintenance. */
+export async function clearStagedRowsInBatches(
+  collection: Pick<mongo.Collection, "find" | "updateMany">,
+  repoId: string, treeGeneration: string, maxBatches = 10
+): Promise<void> {
+  const filter = { repoId, treeGeneration, treeStaged: true };
+  for (let batch = 0; batch < maxBatches; batch++) {
+    const rows = await collection.find(filter, { projection: { _id: 1 }, limit: FILE_TREE_BATCH_SIZE }).toArray();
+    if (!rows.length) return;
+    await collection.updateMany({ ...filter, _id: { $in: rows.map(row => row._id) } },
+      { $unset: { treeStaged: "" } }, { hint: "_id_" });
+    await setImmediate();
+  }
+}

src/core/invalid-credential-backoff.ts

@@ -0,0 +1,17 @@
+import { createHash } from "crypto";
+import { ExpiringMap } from "./expiring-map";
+
+/** Cache only denials, keyed by the exact credential. Never cache successful access. */
+export class InvalidCredentialBackoff {
+  private readonly rejected: ExpiringMap<boolean>;
+  constructor(now = () => Date.now()) { this.rejected = new ExpiringMap<boolean>(2000, now); }
+  async run<T>(token: string, request: () => Promise<T>): Promise<T> {
+    const key = createHash("sha256").update(token).digest("hex");
+    if (this.rejected.get(key)) throw Object.assign(new Error("token_expired"), { status: 401 });
+    try { return await request(); }
+    catch (error) {
+      if ((error as { status?: number })?.status === 401) this.rejected.set(key, true, 60_000);
+      throw error;
+    }
+  }
+}

src/core/model/files/files.schema.ts

@@ -19,6 +19,8 @@ const FileSchema = new Schema({
   },
 });
 
+FileSchema.index({ repoId: 1, treeGeneration: 1, submodule: 1 },
+  { name: "file_submodule_status", partialFilterExpression: { submodule: { $exists: true } } });
 FileSchema.index({ metadataPending: 1, repoId: 1, treeGeneration: 1 },
   { partialFilterExpression: { metadataPending: true } });
 FileSchema.index({ treeStaged: 1, repoId: 1, treeGeneration: 1 },

src/core/request-monitoring.ts

@@ -6,7 +6,7 @@ import { createLogger } from "./logger";
 
 export const LATENCY_BUCKETS = [10, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 30000, 60000];
 export type Service = "api" | "streamer";
-export type Stage = "repository" | "authorization" | "upstream" | "cache_lookup" | "cache_fill" | "worker_wait" | "anonymize" | "streamer_headers";
+export type Stage = "repository" | "credential_resolution" | "upstream" | "cache_lookup" | "cache_fill" | "worker_wait" | "anonymize" | "streamer_headers";
 export interface Metric {
   metric: string; route: string; method: string; count: number; sumMs: number;
   bytes: number; aborted: number; errors: number; slow: number; buckets: number[];

src/core/response-cancellation.ts

@@ -0,0 +1,16 @@
+/** Suppress only transport cancellation after the downstream has gone away. */
+export function isResponseCancellation(error: unknown, response?: { destroyed?: boolean; req?: { aborted?: boolean } }): boolean {
+  if (!response?.destroyed) return false;
+  const seen = new Set<unknown>();
+  for (let value = error; value && typeof value === "object" && !seen.has(value);) {
+    seen.add(value);
+    const item = value as { code?: string; message?: string; cause?: unknown };
+    if (item.message === "response_cancelled") return true;
+    // pipeline also destroys the response on upstream failure. A destroyed
+    // response alone does not prove that a premature close was client-driven.
+    if (response.req?.aborted && (item.code === "ERR_STREAM_PREMATURE_CLOSE" ||
+        item.code === "ABORT_ERR")) return true;
+    value = item.cause;
+  }
+  return false;
+}

src/scripts/create-performance-indexes.ts

@@ -19,6 +19,8 @@ async function main() {
   await Repository.collection.createIndex({ "coauthors.githubId": 1 });
   await File.collection.createIndex({ repoId: 1, treeGeneration: 1, path: 1, name: 1 });
   await File.collection.createIndex({ repoId: 1, treeGeneration: 1, path: 1, size: 1 }, { name: "file_counts_covering" });
+  await File.collection.createIndex({ repoId: 1, treeGeneration: 1, submodule: 1 },
+    { name: "file_submodule_status", partialFilterExpression: { submodule: { $exists: true } } });
   await File.collection.createIndex({ metadataPending: 1, repoId: 1, treeGeneration: 1 },
     { partialFilterExpression: { metadataPending: true } });
   await Repository.collection.createIndex({ retiredTreeGenerations: 1 }, { sparse: true });

src/server/routes/repository-public.ts

@@ -1,3 +1,4 @@
+import { hasVisibleSubmodules } from "../../core/file-submodules";
 import { FILE_LISTING_LIMIT } from "../../core/file-tree";
 import { folderCounts as scopedFolderCounts } from "../../core/folder-counts";
 import { repositoryFailureStatus } from "../../core/repository-failure-status";
@@ -402,14 +403,7 @@ router.get(
         ),
         // Legacy trees retain .gitmodules until their next tree refresh.
         hasSubmodules:
-          (await FileModel.exists({
-            repoId: repo.repoId,
-            treeGeneration: repo.model.treeGeneration || { $exists: false },
-            $or: [
-              { submodule: { $exists: true, $ne: "included" } },
-              { name: ".gitmodules", path: "" },
-            ],
-          })) != null,
+          await hasVisibleSubmodules(repo.repoId, repo.model.treeGeneration),
       });
     } catch (error) {
       handleError(error, res, req);

src/server/routes/route-utils.ts

@@ -1,3 +1,4 @@
+import { isResponseCancellation } from "../../core/response-cancellation";
 import * as express from "express";
 import AnonymousError from "../../core/AnonymousError";
 import * as db from "../database";
@@ -225,6 +226,7 @@ export function handleError(
   res?: express.Response,
   req?: express.Request
 ) {
+  if (isResponseCancellation(error, res)) return;
   printError(error, req);
   let errorCode = error;
   if (error instanceof Error) {

test/credential-validation.test.js

@@ -73,8 +73,10 @@ describe('conditional credential validation', () => {
       expect(requests[1].headers.authorization).to.equal(`token ${token}`);
       status = 401;
       expect(await checkToken(token)).to.equal(false);
+      expect(await checkToken(token)).to.equal(false);
+      expect(requests).to.have.length(3);
       status = 200;
-      expect(await checkToken(token)).to.equal(true);
+      expect(await checkToken(token + '-rotated')).to.equal(true);
       expect(requests[3].headers['if-none-match']).to.equal(undefined);
       expect(requests.every(request => request.url === 'https://api.github.com/user')).to.equal(true);
     } finally { global.fetch = originalFetch; }

test/production-log-findings.test.js

@@ -0,0 +1,112 @@
+const { setImmediate } = require("timers");
+const { expect } = require("chai");
+require("ts-node/register/transpile-only");
+const { isResponseCancellation } = require("../src/core/response-cancellation");
+const { InvalidCredentialBackoff } = require("../src/core/invalid-credential-backoff");
+const { clearStagedRowsInBatches } = require("../src/core/file-tree");
+const { redactSecrets } = require("../src/core/redact-secrets");
+
+describe("production log failure handling", () => {
+  it("suppresses cancellation only when the downstream is destroyed", () => {
+    for (const error of [new Error("response_cancelled"),
+      Object.assign(new Error("Premature close"), { code: "ERR_STREAM_PREMATURE_CLOSE" }),
+      { cause: new Error("response_cancelled") }]) {
+      expect(isResponseCancellation(error, { destroyed: true, req: { aborted: true } })).to.equal(true);
+      expect(isResponseCancellation(error, { destroyed: false })).to.equal(false);
+    }
+    expect(isResponseCancellation({ code: "ECONNRESET" }, { destroyed: true })).to.equal(false);
+    expect(isResponseCancellation({ code: "ERR_STREAM_PREMATURE_CLOSE" }, { destroyed: true })).to.equal(false);
+    const cycle = {}; cycle.cause = cycle;
+    expect(isResponseCancellation(cycle, { destroyed: true })).to.equal(false);
+  });
+  it("keeps credential timing while redacting authorization headers", () => {
+    expect(redactSecrets({ stages: { credential_resolution: 12 }, headers: { authorization: "secret" } }))
+      .to.deep.equal({ stages: { credential_resolution: 12 }, headers: { authorization: "[REDACTED]" } });
+  });
+  it("backs off exact invalid credentials but not rotated tokens or transient errors", async () => {
+    let now = 0;
+    const gate = new InvalidCredentialBackoff(() => now);
+    let calls = 0;
+    const request = async () => { calls++; throw Object.assign(new Error("invalid"), { status: 401 }); };
+    for (let i = 0; i < 3; i++) {
+      try { await gate.run("old", request); throw Error("accepted"); }
+      catch (error) { expect(error.status).to.equal(401); }
+    }
+    expect(calls).to.equal(1);
+    now = 60_001;
+    expect(await gate.run("old", async () => "rechecked")).to.equal("rechecked");
+    expect(await gate.run("new", async () => "ok")).to.equal("ok");
+    for (const status of [403, 429, 500, undefined]) {
+      try { await gate.run("temporary", async () => { throw { status }; }); } catch { /* expected upstream rejection */ }
+      expect(await gate.run("temporary", async () => "ok")).to.equal("ok");
+    }
+  });
+  it("yields and bounds staging writes, leaving remaining markers for maintenance", async () => {
+    let rows = Array.from({ length: 1201 }, (_, _id) => ({ _id }));
+    let writes = 0, yielded = false;
+    setImmediate(() => { yielded = true; });
+    const collection = {
+      find: (_filter, options) => ({ toArray: async () => rows.slice(0, options.limit) }),
+      updateMany: async (filter, update, options) => {
+        expect(filter.repoId).to.equal("repo");
+        expect(filter.treeGeneration).to.equal("generation");
+        expect(filter.treeStaged).to.equal(true);
+        expect(filter._id.$in.length).to.be.at.most(500);
+        expect(update).to.deep.equal({ $unset: { treeStaged: "" } });
+        expect(options.hint).to.equal("_id_");
+        rows = rows.filter(row => !filter._id.$in.includes(row._id));
+        writes++;
+      },
+    };
+    await clearStagedRowsInBatches(collection, "repo", "generation", 2);
+    expect(rows.length).to.equal(201); expect(writes).to.equal(2); expect(yielded).to.equal(true);
+    await clearStagedRowsInBatches(collection, "repo", "generation");
+    expect(rows).to.have.length(0);
+  });
+});
+
+const uri = process.env.PERFORMANCE_MONGO_URI;
+(uri ? describe : describe.skip)("indexed submodule checks", function () {
+  this.timeout(30000);
+  const mongoose = require("mongoose");
+  const File = require("../src/core/model/files/files.model").default;
+  const { hasVisibleSubmodules } = require("../src/core/file-submodules");
+  before(async () => {
+    await mongoose.connect(uri);
+    await File.createIndexes();
+    await File.insertMany(Array.from({length: 12000}, (_, i) => ({
+      repoId: "probe", treeGeneration: "current", path: "", name: "file" + i, size: 1,
+    })));
+  });
+  after(async () => { await File.deleteMany({repoId: "probe"}); await mongoose.disconnect(); });
+  it("avoids a whole-generation scan when no submodules exist", async () => {
+    const executions = [];
+    const original = File.exists;
+    File.exists = function(filter) {
+      const query = original.call(this, filter);
+      const exec = query.exec.bind(query);
+      query.exec = async function () {
+        executions.push(await query.clone().explain("executionStats"));
+        return exec();
+      };
+      return query;
+    };
+    try {
+      expect(await hasVisibleSubmodules("probe", "current")).to.equal(false);
+      expect(executions.length).to.equal(2);
+      for (const plan of executions) {
+        expect(plan.executionStats.totalDocsExamined).to.be.lessThan(10);
+        expect(plan.executionStats.totalKeysExamined).to.be.lessThan(10);
+      }
+    } finally { File.exists = original; }
+  });
+  it("preserves legacy root manifests and excludes included or other-generation markers", async () => {
+    await File.create({repoId:"probe", treeGeneration:"current", path:"", name:"included", submodule:"included"});
+    await File.create({repoId:"probe", treeGeneration:"other", path:"", name:"blocked", submodule:"unavailable"});
+    expect(await hasVisibleSubmodules("probe","current")).to.equal(false);
+    await File.create({repoId:"probe", treeGeneration:"current", path:"", name:"blocked", submodule:"unavailable"});
+    expect(await hasVisibleSubmodules("probe","current")).to.equal(true);
+    await File.create({repoId:"probe", path:"", name:".gitmodules"});
+    expect(await hasVisibleSubmodules("probe")).to.equal(true);
+  });
+});