From d508ab17545f34c032149e7b997286ffe2cb9400 Mon Sep 17 00:00:00 2001 From: Thomas Durieux <5577568+tdurieux@users.noreply.github.com> Date: Fri, 13 Nov 2026 11:38:26 +0100 Subject: [PATCH] fix: bound metadata work and distinguish cancelled requests (#951) --- src/core/AnonymizedFile.ts | 3 + src/core/GitHubUtils.ts | 11 ++- src/core/Repository.ts | 7 +- src/core/file-submodules.ts | 15 +++ src/core/file-tree.ts | 15 +++ src/core/invalid-credential-backoff.ts | 17 ++++ src/core/model/files/files.schema.ts | 2 + src/core/request-monitoring.ts | 2 +- src/core/response-cancellation.ts | 16 ++++ src/scripts/create-performance-indexes.ts | 2 + src/server/routes/repository-public.ts | 10 +- src/server/routes/route-utils.ts | 2 + test/credential-validation.test.js | 4 +- test/production-log-findings.test.js | 112 ++++++++++++++++++++++ 14 files changed, 202 insertions(+), 16 deletions(-) create mode 100644 src/core/file-submodules.ts create mode 100644 src/core/invalid-credential-backoff.ts create mode 100644 src/core/response-cancellation.ts create mode 100644 test/production-log-findings.test.js diff --git a/src/core/AnonymizedFile.ts b/src/core/AnonymizedFile.ts index 0ad2a35..2c96be7 100644 --- a/src/core/AnonymizedFile.ts +++ b/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 }; diff --git a/src/core/GitHubUtils.ts b/src/core/GitHubUtils.ts index c2a699f..8f63839 100644 --- a/src/core/GitHubUtils.ts +++ b/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 { } } +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(); 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; diff --git a/src/core/Repository.ts b/src/core/Repository.ts index 01a614f..63424ed 100644 --- a/src/core/Repository.ts +++ b/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 }); } diff --git a/src/core/file-submodules.ts b/src/core/file-submodules.ts new file mode 100644 index 0000000..ef31cf6 --- /dev/null +++ b/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 { + 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); + } +} diff --git a/src/core/file-tree.ts b/src/core/file-tree.ts index d9d8165..20f4c7a 100644 --- a/src/core/file-tree.ts +++ b/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, + repoId: string, treeGeneration: string, maxBatches = 10 +): Promise { + 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(); + } +} diff --git a/src/core/invalid-credential-backoff.ts b/src/core/invalid-credential-backoff.ts new file mode 100644 index 0000000..ccf5c5d --- /dev/null +++ b/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; + constructor(now = () => Date.now()) { this.rejected = new ExpiringMap(2000, now); } + async run(token: string, request: () => Promise): Promise { + 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; + } + } +} diff --git a/src/core/model/files/files.schema.ts b/src/core/model/files/files.schema.ts index 9979e52..42d3948 100644 --- a/src/core/model/files/files.schema.ts +++ b/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 }, diff --git a/src/core/request-monitoring.ts b/src/core/request-monitoring.ts index c2b899c..f4ca19b 100644 --- a/src/core/request-monitoring.ts +++ b/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[]; diff --git a/src/core/response-cancellation.ts b/src/core/response-cancellation.ts new file mode 100644 index 0000000..5ec895e --- /dev/null +++ b/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(); + 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; +} diff --git a/src/scripts/create-performance-indexes.ts b/src/scripts/create-performance-indexes.ts index 8ad5c3f..374f261 100644 --- a/src/scripts/create-performance-indexes.ts +++ b/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 }); diff --git a/src/server/routes/repository-public.ts b/src/server/routes/repository-public.ts index 90fbc40..375db07 100644 --- a/src/server/routes/repository-public.ts +++ b/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); diff --git a/src/server/routes/route-utils.ts b/src/server/routes/route-utils.ts index cbed9e5..3355176 100644 --- a/src/server/routes/route-utils.ts +++ b/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) { diff --git a/test/credential-validation.test.js b/test/credential-validation.test.js index 7ed5177..7a9c1c9 100644 --- a/test/credential-validation.test.js +++ b/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; } diff --git a/test/production-log-findings.test.js b/test/production-log-findings.test.js new file mode 100644 index 0000000..e70699f --- /dev/null +++ b/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); + }); +});