diff --git a/src/core/GitHubUtils.ts b/src/core/GitHubUtils.ts index 97e5868..e0f562e 100644 --- a/src/core/GitHubUtils.ts +++ b/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 { +function abortableGateLookup(lookup: Promise, signal: AbortSignal): Promise { + 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 { + signal?.throwIfAborted(); const key = githubQuotaKey(token); const localGate = tokenGates.get(key); let waitMs = 0; @@ -96,7 +108,9 @@ async function waitForTokenGate(token: string): Promise { 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 { waitMs, resetAt: new Date(resetAt).toISOString(), }); - await new Promise((resolve) => setTimeout(resolve, waitMs)); + await delay(waitMs, undefined, { signal }); if (localGate) tokenGates.delete(key); } diff --git a/src/core/archive-entry.ts b/src/core/archive-entry.ts index d2060e3..f8b5be5 100644 --- a/src/core/archive-entry.ts +++ b/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 { return new Promise((resolve, reject) => { diff --git a/src/core/logger.ts b/src/core/logger.ts index 39f89b6..b3be112 100644 --- a/src/core/logger.ts +++ b/src/core/logger.ts @@ -320,7 +320,10 @@ export function serializeError(err: unknown): Record { // 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(); diff --git a/src/core/repository-failure-status.ts b/src/core/repository-failure-status.ts index 443868e..3e78184 100644 --- a/src/core/repository-failure-status.ts +++ b/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; } } diff --git a/src/core/request-monitoring.ts b/src/core/request-monitoring.ts index 9c794e0..97c60e0 100644 --- a/src/core/request-monitoring.ts +++ b/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); }; diff --git a/src/core/source/GitHubStream.ts b/src/core/source/GitHubStream.ts index 0a8a532..7b92778 100644 --- a/src/core/source/GitHubStream.ts +++ b/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>(); -const missingContent = new ExpiringMap(2048); +const missingContent = new ExpiringMap(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( 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) { diff --git a/src/core/source/github-content-miss.ts b/src/core/source/github-content-miss.ts new file mode 100644 index 0000000..91574f2 --- /dev/null +++ b/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(2048); +const pending = new Map>(); + +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 { + 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 => { + 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, + }); +} diff --git a/src/core/storage/FileSystem.ts b/src/core/storage/FileSystem.ts index 692ea07..e5dc774 100644 --- a/src/core/storage/FileSystem.ts +++ b/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)); diff --git a/src/core/storage/S3.ts b/src/core/storage/S3.ts index 6cb7e26..35a0555 100644 --- a/src/core/storage/S3.ts +++ b/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); diff --git a/src/core/zipStream.ts b/src/core/zipStream.ts index ade41b6..9aa55a7 100644 --- a/src/core/zipStream.ts +++ b/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); } diff --git a/src/server/index.ts b/src/server/index.ts index d66a2d1..2c39be6 100644 --- a/src/server/index.ts +++ b/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)); diff --git a/src/server/server-options.ts b/src/server/server-options.ts new file mode 100644 index 0000000..dc4f6c6 --- /dev/null +++ b/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(); +}; diff --git a/test/error-codes.test.js b/test/error-codes.test.js index ea1f52a..006617f 100644 --- a/test/error-codes.test.js +++ b/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]; diff --git a/test/fixtures/github-gate-cancellation.js b/test/fixtures/github-gate-cancellation.js new file mode 100644 index 0000000..2800643 --- /dev/null +++ b/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. diff --git a/test/performance-regressions.test.js b/test/performance-regressions.test.js index 2735bb4..1c19db4 100644 --- a/test/performance-regressions.test.js +++ b/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); diff --git a/test/production-failure-fixes.test.js b/test/production-failure-fixes.test.js index f9646ab..e10eda9 100644 --- a/test/production-failure-fixes.test.js +++ b/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" }); }); diff --git a/test/production-followup-fixes.test.js b/test/production-followup-fixes.test.js new file mode 100644 index 0000000..af62acd --- /dev/null +++ b/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('private') }; + 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"); + }); + } +}); diff --git a/test/request-monitoring.test.js b/test/request-monitoring.test.js index 404bae9..ccebfd8 100644 --- a/test/request-monitoring.test.js +++ b/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); + } }); }