diff --git a/docs/performance-monitoring.md b/docs/performance-monitoring.md index 7bfa57c..625dbfc 100644 --- a/docs/performance-monitoring.md +++ b/docs/performance-monitoring.md @@ -88,3 +88,40 @@ validators avoid fetching and rendering content again; changed terms or content invalidate them. Raw Markdown served after a rendering failure has a separate validator, so later requests retry rendering when the transient failure clears. + +## Production failures and restarts + +Run `node scripts/check-production-logs.js 15` from a machine with Node and the +Docker CLI to summarize recent application logs and container states. For a +remote Docker host, use +`DOCKER_HOST=ssh://ubuntu@host node scripts/check-production-logs.js 15`. +The optional +window is 1–1440 minutes. The report includes completed 5xx responses, interrupted +requests, token-refresh failures, missing-file warnings, native unhandled stream +errors, and Docker restart counts. It prints counts and container metadata; +request URLs, repository IDs, file paths, credentials, and Docker environment +variables are excluded. Restart counts cover the container's lifetime, while log +counts cover the selected window. Compare restart counts with the last deployment +baseline even when every container is currently healthy. + +The API and streamers log `process_fatal_error` through Node's +`uncaughtExceptionMonitor`. This records fatal errors before the default process +exit; it does not resume execution or swallow uncaught exceptions. Native stderr +may describe the same crash as well, so fatal log indicators are not a count of +unique process exits. Use Docker restart counts to confirm exits. + +Legacy OAuth reset requests are coalesced by owner, credential, and OAuth app. +A failed 401/404 reset backs off for 30 minutes; transient failures back off for +one minute. A replaced credential or changed OAuth configuration gets a new key. +GitHub App and personal access tokens do not use the OAuth reset endpoint. Refresh +failure leaves the credential unchanged and continues through normal token +validation and fallback. + +Confirmed GitHub content 404s back off for 30 seconds in a bounded process-local +cache. The key includes repository, path, commit, blob SHA, cache generation, +revision, and credential. Permission failures and transport errors are not cached. +This reduces repeated upstream misses without changing the file tree or treating +an unavailable file as a successful cache fill. Persistent 404s still require +checking the repository's source snapshot. Application rate limits remain in +force; 429 replies now include `rate_limited`, `resetAt`, and `Retry-After` so +clients can wait for the actual store reset. diff --git a/scripts/check-production-logs.js b/scripts/check-production-logs.js new file mode 100644 index 0000000..dc77d27 --- /dev/null +++ b/scripts/check-production-logs.js @@ -0,0 +1,63 @@ +const { spawn, execFileSync } = require("node:child_process"); +const { createInterface } = require("node:readline"); + +function logSummary() { + return { completed: 0, interrupted: 0, completed5xx: 0, fatalErrors: 0, errors: 0, + warnings: 0, tokenRefreshFailures: 0, missingFiles: 0, rateLimited: 0, connectionResets: 0 }; +} +function observeLine(summary, line) { + // Older releases rely on Node's stderr. New releases also emit a redacted + // process_fatal_error event before Node terminates. + const match = /^\S+ (INFO|WARN|ERROR) \[([^\]]+)\] .*? (\{.*\})$/.exec(line); + if (!match) { + if (/Unhandled ['"]error['"] event|unhandled(?:promise)?rejection|FATAL ERROR|heap out of memory/i.test(line)) summary.fatalErrors++; + return; + } + let data; + try { data = JSON.parse(match[3]); } catch { return; } + const [, level, scope] = match; + if (level === "ERROR") summary.errors++; + if (level === "WARN") summary.warnings++; + if (data.code === "process_fatal_error" || /startup failed/i.test(line)) summary.fatalErrors++; + if (data.code === "token_refresh_failed") summary.tokenRefreshFailures++; + if (scope === "anonymized-file" && data.code === "file_not_found") summary.missingFiles++; + if (scope === "requests" && typeof data.ms === "number") { + if (data.outcome === "completed") { summary.completed++; if (data.status >= 500) summary.completed5xx++; } + else summary.interrupted++; + if (data.status === 429) summary.rateLimited++; + } + if (/ECONNRESET|socket hang up/i.test(line)) summary.connectionResets++; +} +async function readLogs(name, since) { + const summary = logSummary(); + const child = spawn("docker", ["logs", "--since", since, name], { stdio: ["ignore", "pipe", "pipe"] }); + const completion = new Promise((resolve, reject) => { + child.once("error", reject); + child.once("close", code => code === 0 ? resolve() : reject(new Error("docker logs failed"))); + }); + for (const output of [child.stdout, child.stderr]) { + createInterface({ input: output }).on("line", line => observeLine(summary, line)); + } + await completion; + return summary; +} +function containerSummary(container) { + return { name: container.Name.replace(/^\//, ""), healthy: container.State.Health?.Status === "healthy", + running: container.State.Running, restartCount: container.RestartCount, + oomKilled: container.State.OOMKilled, startedAt: container.State.StartedAt }; +} +async function main(minutes) { + if (!Number.isInteger(minutes) || minutes < 1 || minutes > 1440) throw new Error("Minutes must be between 1 and 1440"); + const since = new Date(Date.now() - minutes * 60000).toISOString(); + const names = execFileSync("docker", ["ps", "-a", "--filter", "label=com.docker.compose.project=anonymous_github", "--format", "{{.Names}}"], { encoding: "utf8" }) + .trim().split("\n").filter(name => /^anonymous_github-(anonymous_github|streamer)-\d+$/.test(name)); + if (!names.length) throw new Error("No application containers found"); + const containers = JSON.parse(execFileSync("docker", ["inspect", ...names], { encoding: "utf8" })); + const services = []; + for (const container of containers) services.push({ ...containerSummary(container), logs: await readLogs(container.Name, since) }); + console.log(JSON.stringify({ checkedAt: new Date().toISOString(), minutes, services }, null, 2)); +} +module.exports = { logSummary, observeLine, containerSummary }; +if (require.main === module) main(Number(process.argv[2] || 15)).catch(() => { + console.error("Production log check failed; check Docker access and the requested window."); process.exitCode = 1; +}); diff --git a/src/core/GitHubUtils.ts b/src/core/GitHubUtils.ts index d45e89b..84959b7 100644 --- a/src/core/GitHubUtils.ts +++ b/src/core/GitHubUtils.ts @@ -1,3 +1,4 @@ +import { refreshLegacyToken } from "./legacy-token-refresh"; import AnonymizedRepositoryModel from "./model/anonymizedRepositories/anonymizedRepositories.model"; import { isConnected } from "../server/database"; import { githubQuotaKey, githubTokenContext } from "./github-token-context"; @@ -8,7 +9,7 @@ import { createClient, RedisClientType } from "redis"; import AnonymousError from "./AnonymousError"; import Repository from "./Repository"; -import { getCredential, replaceCredential, getCredentialToken } from "./credentials"; +import { getCredential, getCredentialToken } from "./credentials"; import config from "../config"; import { createLogger } from "./logger"; import { measureStage } from "./request-monitoring"; @@ -337,56 +338,11 @@ async function resolveRepositoryToken(repository: Repository) { credential?.persisted && (!tokenAge || tokenAge < new Date(Date.now() - 1000 * 60 * 60 * 24 * 7)) ) { - const url = `https://api.github.com/applications/${config.CLIENT_ID}/token`; - const headers = { - Accept: "application/vnd.github+json", - "X-GitHub-Api-Version": "2022-11-28", - }; - - const res = await fetch(url, { - method: "PATCH", - body: JSON.stringify({ - access_token: ownerAccessToken, - }), - credentials: "include", - headers: { - ...headers, - Authorization: - "Basic " + - Buffer.from( - config.CLIENT_ID + ":" + config.CLIENT_SECRET - ).toString("base64"), - }, - }); - // Only persist a refreshed token if GitHub actually returned a - // valid one. Without this guard, a 4xx/5xx error body (revoked - // OAuth, rate limit, transient outage) silently overwrites the - // user's stored token with `undefined`, which then propagates as - // `Authorization: token undefined` to every subsequent API call — - // 401 even on public repos, and the config.GITHUB_TOKEN fallback - // below is unreachable because the token field is no longer falsy. - if (res.ok) { - const resBody = (await res.json().catch(() => null)) as - | { token?: unknown } - | null; - const refreshed = - resBody && typeof resBody.token === "string" && resBody.token.length > 0 - ? resBody.token - : null; - if (refreshed) { - if (await replaceCredential(repository.owner.id, ownerAccessToken, refreshed)) { - checkedRepositoryTokens.set(repository, refreshed); - return refreshed; - } - return (await getCredentialToken(repository.owner.id)) || config.GITHUB_TOKEN; - } + const refreshed = await refreshLegacyToken(repository.owner.id, ownerAccessToken); + if (refreshed !== null) { + checkedRepositoryTokens.set(repository, refreshed); + return refreshed; } - logger.warn("token refresh failed; falling back", { - code: "token_refresh_failed", - httpStatus: res.status, - username: repository.owner.model.username, - }); - // fall through to the checkToken path / config.GITHUB_TOKEN } const check = await checkToken(ownerAccessToken); if (check) { diff --git a/src/core/expiring-map.ts b/src/core/expiring-map.ts new file mode 100644 index 0000000..ec076e1 --- /dev/null +++ b/src/core/expiring-map.ts @@ -0,0 +1,17 @@ +/** A bounded cache for short-lived failures. Keys must include access and content versions. */ +export class ExpiringMap { + private entries = new Map(); + constructor(private capacity: number, private now = () => Date.now()) {} + get(key: string): T | undefined { + const entry = this.entries.get(key); + if (!entry) return undefined; + if (entry.expires <= this.now()) { this.entries.delete(key); return undefined; } + return entry.value; + } + set(key: string, value: T, ttl: number) { + for (const [key, entry] of this.entries) if (entry.expires <= this.now()) this.entries.delete(key); + this.entries.delete(key); + if (this.entries.size >= this.capacity) this.entries.delete(this.entries.keys().next().value!); + this.entries.set(key, { value, expires: this.now() + ttl }); + } +} diff --git a/src/core/legacy-token-refresh.ts b/src/core/legacy-token-refresh.ts new file mode 100644 index 0000000..b2721c9 --- /dev/null +++ b/src/core/legacy-token-refresh.ts @@ -0,0 +1,52 @@ +import { createHash } from "crypto"; +import config from "../config"; +import { getCredentialToken, replaceCredential } from "./credentials"; +import { ExpiringMap } from "./expiring-map"; +import { createLogger } from "./logger"; + +const logger = createLogger("github"); +const cooldowns = new ExpiringMap(1024); +const pending = new Map>(); + +/** Only legacy OAuth tokens belong to the OAuth application's reset endpoint. */ +export async function refreshLegacyToken(ownerId: string, previous: string): Promise { + if (!config.GITHUB_OAUTH_ENABLED || !config.CLIENT_ID || !config.CLIENT_SECRET || + config.CLIENT_ID === "CLIENT_ID" || config.CLIENT_SECRET === "CLIENT_SECRET" || + /^(gh[psur]_|github_pat_)/.test(previous)) return null; + const key = createHash("sha256").update(JSON.stringify([ownerId, previous, config.CLIENT_ID, config.CLIENT_SECRET])).digest("hex"); + if (cooldowns.get(key)) return null; + const running = pending.get(key); + if (running) return running; + if (pending.size >= 1024) return null; + const refresh = async () => { + let status = 502; + try { + const res = await fetch(`https://api.github.com/applications/${encodeURIComponent(config.CLIENT_ID)}/token`, { + method: "PATCH", body: JSON.stringify({ access_token: previous }), + signal: AbortSignal.timeout(5000), + headers: { Accept: "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28", + Authorization: "Basic " + Buffer.from(config.CLIENT_ID + ":" + config.CLIENT_SECRET).toString("base64") }, + }); + status = res.status; + if (res.ok) { + const body = await res.json().catch(() => null) as { token?: unknown } | null; + if (typeof body?.token === "string" && body.token.length) { + if (await replaceCredential(ownerId, previous, body.token)) return body.token; + return (await getCredentialToken(ownerId)) || config.GITHUB_TOKEN; + } + } else { + // Release the response connection even when only the status matters. + await res.body?.cancel(); + } + } catch { + // The existing token can still work during a transient reset-endpoint failure. + } + cooldowns.set(key, true, status === 401 || status === 404 ? 30 * 60_000 : 60_000); + logger.warn("token refresh failed; backing off", { code: "token_refresh_failed", httpStatus: status }); + return null; + }; + const work = refresh(); + pending.set(key, work); + try { return await work; } + finally { if (pending.get(key) === work) pending.delete(key); } +} diff --git a/src/core/process-monitoring.ts b/src/core/process-monitoring.ts new file mode 100644 index 0000000..e33e2da --- /dev/null +++ b/src/core/process-monitoring.ts @@ -0,0 +1,11 @@ +import { createLogger, serializeError } from "./logger"; + +/** Observe fatal errors without overriding Node's default exit behavior. */ +export function installFatalErrorLogging(service: "api" | "streamer") { + const logger = createLogger("process"); + const report = (error: Error, origin: string) => logger.error("process fatal error", { + ...serializeError(error), code: "process_fatal_error", service, origin, + }); + process.on("uncaughtExceptionMonitor", report); + return () => process.removeListener("uncaughtExceptionMonitor", report); +} diff --git a/src/core/repository-failure-status.ts b/src/core/repository-failure-status.ts new file mode 100644 index 0000000..443868e --- /dev/null +++ b/src/core/repository-failure-status.ts @@ -0,0 +1,11 @@ +/** Persisted repository failures need the same client status as the original error. */ +export function repositoryFailureStatus(code?: string): number { + switch (code) { + case "repo_not_found": case "branch_not_found": case "commit_not_found": return 404; + case "repo_not_accessible": case "repository_not_accessible": return 403; + case "token_expired": case "not_connected": return 401; + case "github_rate_limit_exceeded": return 429; + case "github_unavailable": return 502; + default: return 500; + } +} diff --git a/src/core/source/GitHubStream.ts b/src/core/source/GitHubStream.ts index 897d697..4d0b36a 100644 --- a/src/core/source/GitHubStream.ts +++ b/src/core/source/GitHubStream.ts @@ -1,3 +1,4 @@ +import { ExpiringMap } from "../expiring-map"; import { contentGenerationPrefix, contentRetirementMarker } from "../content-generation"; import { createHash } from "crypto"; import { coordinatedFill } from "../cache-coordination"; @@ -29,6 +30,7 @@ const logger = createLogger("gh-stream"); const GH_API_CONCURRENCY = 6; const cacheFills = new Map>(); +const missingContent = new ExpiringMap(2048); async function pMap( items: T[], @@ -64,6 +66,15 @@ export function githubRawFileUrl( )}/raw/${encodeURIComponent(commit)}/${encodedPath}`; } +function downloadError(error: Error, filePath: string): AnonymousError { + if (error instanceof AnonymousError) return 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" + : status === 404 ? "file_not_found" : status === 422 ? "file_too_big" : "upstream_error"; + return new AnonymousError(code, { httpStatus: Number.isInteger(status) && status! >= 400 && status! < 500 ? status : 502, cause: error, object: filePath }); +} + export default class GitHubStream extends GitHubBase { type: "GitHubDownload" | "GitHubStream" | "Zip" = "GitHubStream"; @@ -123,6 +134,17 @@ export default class GitHubStream extends GitHubBase { }); } + private deferredRawDownload(token: string, filePath: string): stream.Readable { + let active: stream.Readable | undefined; + const download = () => this.downloadFileViaRaw(token, filePath); + const output = stream.Readable.from((async function* () { + active = download(); + for await (const chunk of active) yield chunk; + })(), { objectMode: false }); + output.once("close", () => active?.destroy()); + return output; + } + // Try the blob API, then fall back to the raw URL on statuses where the // path-based endpoint can still succeed. 422 is the blob endpoint's size // cap; 404 can happen with stale/invalid blob SHAs while the path still @@ -135,9 +157,9 @@ export default class GitHubStream extends GitHubBase { // Public raw downloads need no bearer token and do not consume the // unauthenticated REST API quota. GitHub also resolves LFS pointers here. if (!token || githubTokenContext(token)?.publicRepository) { - return Promise.resolve(this.downloadFileViaRaw(token, filePath)); + return Promise.resolve(this.deferredRawDownload(token, filePath)); } - return new Promise((resolve) => { + return new Promise((resolve, reject) => { const blobStream = this.downloadFile(token, sha); let settled = false; @@ -148,7 +170,7 @@ export default class GitHubStream extends GitHubBase { filePath, statusCode, }); - resolve(this.downloadFileViaRaw(token, filePath)); + resolve(this.deferredRawDownload(token, filePath)); }; blobStream.on("error", (err) => { @@ -160,13 +182,10 @@ export default class GitHubStream extends GitHubBase { fallbackToRaw(statusCode); return; } - // Other errors: let the normal pipeline handle them. - // Defer destroy so callers can attach error listeners before - // the error event fires, avoiding an uncaughtException crash. + // Reject before handing a stream to the caller. nextTick runs before + // await continuations, so emitting on a returned stream here can crash. settled = true; - const passthrough = new stream.PassThrough(); - resolve(passthrough); - process.nextTick(() => passthrough.destroy(err)); + reject(err); }); blobStream.on("response", (response) => { @@ -191,9 +210,7 @@ export default class GitHubStream extends GitHubBase { token: string, filePath: string ): stream.Readable { - const out = new stream.PassThrough(); let active = blobStream; - out.once("close", () => active.destroy()); // The generator only reads upstream when the output has capacity. const downloadRaw = () => this.downloadFileViaRaw(token, filePath); async function* resolve() { @@ -225,12 +242,10 @@ export default class GitHubStream extends GitHubBase { } } } - const resolved = stream.Readable.from(resolve()); - out.once("close", () => resolved.destroy()); - resolved.on("error", (error) => out.destroy(error)); - resolved.pipe(out); - - return out; + const resolved = stream.Readable.from(resolve(), { objectMode: false }); + // Keep the generator lazy until the consumer has installed its handlers. + resolved.once("close", () => active.destroy()); + return resolved; } async getFileContentCache( @@ -299,34 +314,38 @@ export default class GitHubStream extends GitHubBase { filling = coordinatedFill(key, cached, async () => { await assertActive(); const token = await this.data.getToken(); - return measureStage("upstream", async () => { - const content = await this.downloadWithFallback(token, expected.sha, filePath); - let count = 0; - const limited = new stream.Transform({ transform(chunk, encoding, callback) { - count += chunk.length; - callback(count > config.MAX_FILE_SIZE ? new AnonymousError("file_too_big", { httpStatus: 413 }) : null, chunk); - } }); - content.on("error", error => { - if (error instanceof AnonymousError) { limited.destroy(error); return; } - const status = (error as { response?: { statusCode?: number }; status?: number; httpStatus?: number }); - const httpStatus = status.response?.statusCode ?? status.status ?? status.httpStatus; - const code = httpStatus === 403 ? "file_not_accessible" - : httpStatus === 404 ? "file_not_found" - : httpStatus === 422 ? "file_too_big" : "upstream_error"; - limited.destroy(new AnonymousError(code, { httpStatus: httpStatus || 502, cause: error, object: filePath })); + 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 }); + try { + return await measureStage("upstream", async () => { + const content = await this.downloadWithFallback(token, expected.sha, filePath) + .catch(error => { throw downloadError(error, filePath); }); + let count = 0; + const limited = new stream.Transform({ transform(chunk, encoding, callback) { + count += chunk.length; + callback(count > config.MAX_FILE_SIZE ? new AnonymousError("file_too_big", { httpStatus: 413 }) : null, chunk); + } }); + content.on("error", error => limited.destroy(downloadError(error, filePath))); + limited.once("close", () => content.destroy()); + limited.on("error", () => {}); + content.pipe(limited); + const deadline = setTimeout(() => limited.destroy(new AnonymousError("upstream_error", { httpStatus: 502 })), 120_000); + try { + await storage.write(repoId, cachePath, limited, this.type, expected.size); + try { await assertActive(); } + catch (error) { await storage.rm(repoId, cachePath); throw error; } + } + finally { clearTimeout(deadline); limited.destroy(); content.destroy(); } + return true; }); - limited.once("close", () => content.destroy()); - limited.on("error", () => {}); - content.pipe(limited); - const deadline = setTimeout(() => limited.destroy(new AnonymousError("upstream_error", { httpStatus: 502 })), 120_000); - try { - await storage.write(repoId, cachePath, limited, this.type, expected.size); - try { await assertActive(); } - catch (error) { await storage.rm(repoId, cachePath); throw error; } + } catch (error) { + if (error instanceof AnonymousError && error.message === "file_not_found" && error.httpStatus === 404) { + missingContent.set(missKey, true, 30_000); } - finally { clearTimeout(deadline); limited.destroy(); content.destroy(); } - return true; - }); + throw error; + } }).then(() => {}); cacheFills.set(key, filling); void filling.finally(() => { if (cacheFills.get(key) === filling) cacheFills.delete(key); }).catch(() => {}); diff --git a/src/server/index.ts b/src/server/index.ts index cd44437..d66a2d1 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -1,3 +1,5 @@ +import { sendRequestRateLimit } from "./request-rate-limit"; +import { installFatalErrorLogging } from "../core/process-monitoring"; import { githubAppRouter, githubAppWebhook } from "./routes/github-app"; import { config as dotenv } from "dotenv"; dotenv(); @@ -47,6 +49,7 @@ import { createReviewConsentPage, isReviewConsentPagePath } from "./review-conse const reviewConsentPage = createReviewConsentPage(); const logger = createLogger("server"); +installFatalErrorLogging("api"); // Lazily build the templated index.html on first request so the server // works even when started before `gulp` finishes. @@ -191,9 +194,7 @@ export default async function start() { keyGenerator: requestRateLimitKey, standardHeaders: "draft-6", legacyHeaders: false, - message: (_request: express.Request, _response: express.Response) => { - return `You can only make ${config.RATE_LIMIT} requests every 15min. Please try again later.`; - }, + handler: sendRequestRateLimit, }); const speedLimiter = slowDown({ windowMs: 15 * 60 * 1000, // 15 minutes diff --git a/src/server/request-rate-limit.ts b/src/server/request-rate-limit.ts new file mode 100644 index 0000000..28e9451 --- /dev/null +++ b/src/server/request-rate-limit.ts @@ -0,0 +1,9 @@ +import { Request, Response } from "express"; + +/** Give clients a retry time instead of an unparseable plain-text failure. */ +export function sendRequestRateLimit(req: Request, res: Response) { + const rate = (req as Request & { rateLimit?: { resetTime?: Date } }).rateLimit; + const resetAt = rate?.resetTime?.getTime() || Date.now() + 15 * 60_000; + res.setHeader("Retry-After", String(Math.max(1, Math.ceil((resetAt - Date.now()) / 1000)))); + res.status(429).json({ error: "rate_limited", resetAt }); +} diff --git a/src/server/routes/repository-public.ts b/src/server/routes/repository-public.ts index 371f09e..ef1c8ad 100644 --- a/src/server/routes/repository-public.ts +++ b/src/server/routes/repository-public.ts @@ -1,3 +1,4 @@ +import { repositoryFailureStatus } from "../../core/repository-failure-status"; import { anonymizePath } from "../../core/anonymize-utils"; import * as express from "express"; import config from "../../config"; @@ -334,7 +335,7 @@ router.get( : "repository_not_accessible", { object: repo, - httpStatus: 500, + httpStatus: repositoryFailureStatus(repo.model.statusMessage), } ); } diff --git a/src/server/routes/route-utils.ts b/src/server/routes/route-utils.ts index 295bb01..cbed9e5 100644 --- a/src/server/routes/route-utils.ts +++ b/src/server/routes/route-utils.ts @@ -233,10 +233,10 @@ export function handleError( errorCode = String(error); } let status = 500; - if (error.httpStatus) { - status = error.httpStatus; - } else if (error.$metadata?.httpStatusCode) { - status = error.$metadata.httpStatusCode; + const explicitStatus = [error?.httpStatus, error?.status, error?.response?.statusCode, + error?.$metadata?.httpStatusCode].find(value => Number.isInteger(value) && value >= 400 && value <= 599); + if (explicitStatus !== undefined) { + status = explicitStatus; } else if ( errorCode && (errorCode.indexOf("not_found") > -1 || diff --git a/src/streamer/index.ts b/src/streamer/index.ts index 28106b0..8e80c7f 100644 --- a/src/streamer/index.ts +++ b/src/streamer/index.ts @@ -1,3 +1,4 @@ +import { installFatalErrorLogging } from "../core/process-monitoring"; import { config as dotenv } from "dotenv"; dotenv(); @@ -16,6 +17,7 @@ import { startPerformanceMonitoring } from "../core/performance-monitoring"; import { getAnonymizationPoolStats } from "../core/anonymization-pool"; const logger = createLogger("streamer"); +installFatalErrorLogging("streamer"); const app = express(); app.use(monitorRequests("streamer")); diff --git a/test/credential-migration.test.js b/test/credential-migration.test.js index 18afd77..08a4848 100644 --- a/test/credential-migration.test.js +++ b/test/credential-migration.test.js @@ -179,13 +179,15 @@ describeMongo("credential access (MongoDB)", function () { await setCredential(owner.id, "old-token"); await Credential.updateOne({ ownerId: owner._id }, { $set: { updatedAt: new Date(0) } }); const originalFetch = global.fetch; + const oauthConfig = [config.GITHUB_OAUTH_ENABLED, config.CLIENT_ID, config.CLIENT_SECRET]; + config.GITHUB_OAUTH_ENABLED = true; config.CLIENT_ID = "fixture-client"; config.CLIENT_SECRET = "fixture-secret"; global.fetch = async (url, options) => { expect(url).to.include("api.github.com/applications/"); expect(JSON.parse(options.body).access_token).to.equal("old-token"); return { ok: true, json: async () => ({ token: "refreshed-token" }) }; }; try { expect(await repo.getToken()).to.equal("refreshed-token"); } - finally { global.fetch = originalFetch; } + finally { global.fetch = originalFetch; [config.GITHUB_OAUTH_ENABLED, config.CLIENT_ID, config.CLIENT_SECRET] = oauthConfig; } expect(await getCredentialToken(owner.id)).to.equal("refreshed-token"); expect((await RepoModel.collection.findOne({ _id: repo.model._id })).source.accessToken).to.equal(undefined); }); diff --git a/test/fixtures/github-stream-failure.js b/test/fixtures/github-stream-failure.js new file mode 100644 index 0000000..faa4b28 --- /dev/null +++ b/test/fixtures/github-stream-failure.js @@ -0,0 +1,44 @@ +const assert = require("node:assert/strict"); +const { Readable } = require("node:stream"); +const { setImmediate } = require("node:timers"); +require("ts-node/register/transpile-only"); +const GitHubStream = require("../../src/core/source/GitHubStream").default; +const failure = Object.assign(new Error("upstream failed"), process.argv[2] === "reset" ? { code: "ECONNRESET" } : { response: { statusCode: 401 } }); +const source = new GitHubStream({ repoId: "fixture", organization: "owner", repoName: "repo", commit: "abc", getToken: () => "token" }); +async function main() { + if (process.argv[2] === "fatal") { + require("../../src/core/process-monitoring").installFatalErrorLogging("streamer"); + await new Promise(() => setImmediate(() => { throw new Error("fixture fatal"); })); + } + if (process.argv[2] === "fallback" || process.argv[2] === "public-raw") { + source.downloadFile = () => { + const input = new Readable({ read() {} }); + process.nextTick(() => input.destroy(Object.assign(new Error("blob unavailable"), { response: { statusCode: 404 } }))); + return input; + }; + source.downloadFileViaRaw = () => { + const input = new Readable({ read() {} }); + process.nextTick(() => input.destroy(failure)); + return input; + }; + const content = await source.downloadWithFallback(process.argv[2] === "public-raw" ? "" : "token", "blob", "file"); + await new Promise(resolve => setImmediate(resolve)); + await assert.rejects(async () => { for await (const chunk of content) void chunk; }, error => error === failure); + } else if (process.argv[2] === "lfs") { + const pointer = "version https://git-lfs.github.com/spec/v1\noid sha256:" + "a".repeat(64) + "\nsize 10\n"; + source.downloadFileViaRaw = () => new Readable({ read() { this.destroy(failure); } }); + const content = source.resolveLfsPointer(Readable.from([pointer]), "token", "file"); + // A producer must not start emitting errors while its consumer is still awaiting. + await new Promise(resolve => setImmediate(resolve)); + await assert.rejects(async () => { for await (const chunk of content) void chunk; }, error => error === failure); + } else { + source.downloadFile = () => { + const content = new Readable({ read() {} }); + process.nextTick(() => content.destroy(failure)); + return content; + }; + await assert.rejects(source.downloadWithFallback("token", "blob", "file"), error => error === failure); + await new Promise(resolve => setImmediate(resolve)); + } +} +main().then(() => { process.stdout.write("failure contained\n"); process.exit(0); }, error => { console.error(error); process.exit(1); }); diff --git a/test/fixtures/performance-streamer.js b/test/fixtures/performance-streamer.js index 3949f34..ee03454 100644 --- a/test/fixtures/performance-streamer.js +++ b/test/fixtures/performance-streamer.js @@ -9,7 +9,18 @@ const maintenance = require(path.join(base, "core/temporary-storage")); const startMaintenance = maintenance.startTemporaryStorageMaintenance; // Exercise the real startup/timer wiring without waiting a production minute. maintenance.startTemporaryStorageMaintenance = () => startMaintenance(25); -GitHubStream.prototype.downloadWithFallback = async function () { +const downloadWithFallback = GitHubStream.prototype.downloadWithFallback; +GitHubStream.prototype.downloadWithFallback = async function (token, sha, filePath) { + if (filePath.startsWith("fail-")) { + this.downloadFile = () => { + const input = new Readable({ read() {} }); + const failure = Object.assign(new Error("fixture upstream failure"), filePath === "fail-reset.txt" + ? { code: "ECONNRESET" } : { response: { statusCode: 401 } }); + process.nextTick(() => input.destroy(failure)); + return input; + }; + return downloadWithFallback.call(this, token, sha, filePath); + } process.send({ download: true }); return Readable.from([Buffer.alloc(300000, "x")]); }; diff --git a/test/performance-regressions.test.js b/test/performance-regressions.test.js index 6c1fab7..2735bb4 100644 --- a/test/performance-regressions.test.js +++ b/test/performance-regressions.test.js @@ -121,22 +121,22 @@ describe("performance regressions", function () { expect(files).to.have.length(20); expect(calls).to.deep.equal([{ sha: "head", recursive: true }]); }); - for (const [upstream, code, status] of [ + for (const [caseId, [upstream, code, status]] of [ [Object.assign(new Error("forbidden"), { response: { statusCode: 403 } }), "file_not_accessible", 403], [Object.assign(new Error("missing"), { response: { statusCode: 404 } }), "file_not_found", 404], [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/AnonymousError").default)("file_not_accessible", { httpStatus: 403 }), "file_not_accessible", 403], - ]) { + ].entries()) { it(`preserves ${code}/${status} for failed downloads (${upstream.message})`, async () => { - const source = new GitHubStream({ repoId: "failed-download", organization: "owner", repoName: "repo", getToken: () => "token" }); + const source = new GitHubStream({ repoId: `failed-download-${caseId}`, organization: "owner", repoName: "repo", getToken: () => "token" }); const input = new PassThrough(); stub(storage, "fileInfo", async () => { throw Error("cache miss"); }); 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; }; - try { await source.getFileContentCache("file.txt", "failed-download", () => ({ sha: "blob", size: 1 })); throw Error("expected failure"); } + 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); if (upstream instanceof require("../src/core/AnonymousError").default) expect(error).to.equal(upstream); diff --git a/test/performance-streamer-integration.test.js b/test/performance-streamer-integration.test.js index 58d9299..ccfc38b 100644 --- a/test/performance-streamer-integration.test.js +++ b/test/performance-streamer-integration.test.js @@ -75,6 +75,22 @@ const uri = process.env.PERFORMANCE_MONGO_URI; repoId: row.repoId, repoFullName: "owner/repo", commit: "abc", ...(legacy ? {} : { cacheGeneration: options.cacheGeneration }), sha: "blob", size: 300000, filePath, anonymizerOptions: options }) }); } + for (const [filePath, expectedStatus] of [["fail-401.txt", 401], ["fail-reset.txt", 502]]) { + it(`contains ${filePath} errors in the actual streamer and serves another request`, async () => { + const runtime = await launch(); + try { + await runtime.ready; + const row = await createRow(filePath); + const failure = await request(runtime, row, filePath); + expect(failure.status).to.equal(expectedStatus); await failure.text(); + expect((await fetch(runtime.url + "/healthcheck")).status).to.equal(200); + const recovered = await request(runtime, row, "working.txt"); + expect(recovered.status).to.equal(200); expect((await recovered.text()).length).to.equal(300000); + expect(runtime.child.exitCode).to.equal(null); + } finally { await stop(runtime); } + }); + } + it("connects before serving and rejects a fill published after expiration in the actual streamer entrypoint", async () => { const runtime = await launch(false, true); let pending; try { diff --git a/test/production-failure-fixes.test.js b/test/production-failure-fixes.test.js new file mode 100644 index 0000000..d1e35a8 --- /dev/null +++ b/test/production-failure-fixes.test.js @@ -0,0 +1,133 @@ +const { expect } = require("chai"); +const { spawnSync } = require("node:child_process"); +const path = require("node:path"); +require("ts-node/register/transpile-only"); +const config = require("../src/config").default; +const credentials = require("../src/core/credentials"); +const { refreshLegacyToken } = require("../src/core/legacy-token-refresh"); +const { ExpiringMap } = require("../src/core/expiring-map"); +const { sendRequestRateLimit } = require("../src/server/request-rate-limit"); +const GitHubStream = require("../src/core/source/GitHubStream").default; +const storage = require("../src/core/storage").default; +const { logSummary, observeLine, containerSummary } = require("../scripts/check-production-logs"); + +describe("production failure fixes", function () { + this.timeout(20000); + const restores = []; + function stub(object, key, value) { const old = object[key]; object[key] = value; restores.push(() => { object[key] = old; }); } + afterEach(() => restores.splice(0).reverse().forEach(restore => restore())); + for (const mode of ["401", "reset", "lfs", "fallback", "public-raw"]) { + it(`contains ${mode} upstream stream errors in a process with no recovery handler`, function () { + const result = spawnSync(process.execPath, [path.join(__dirname, "fixtures/github-stream-failure.js"), mode], { + 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.status, result.stderr).to.equal(0); + expect(result.stdout).to.include("failure contained"); + }); + } + it("records a fatal event without keeping a broken process running", function () { + const result = spawnSync(process.execPath, [path.join(__dirname, "fixtures/github-stream-failure.js"), "fatal"], { + timeout: 15000, encoding: "utf8", env: { ...process.env, NODE_ENV: "test", GITHUB_APP_ENABLED: "false", REDIS_PORT: "1" }, + }); + expect(result.status).to.equal(1); + expect(result.stderr + result.stdout).to.include("process_fatal_error"); + }); + function oauth() { + stub(config, "GITHUB_OAUTH_ENABLED", true); stub(config, "CLIENT_ID", "test-client"); stub(config, "CLIENT_SECRET", "test-secret"); + } + it("coalesces token refreshes and backs off on permanent 404 responses without replacing credentials", async () => { + oauth(); let calls = 0, replacements = 0; + stub(globalThis, "fetch", async () => { calls++; return { ok: false, status: 404, body: { cancel: async () => {} } }; }); + stub(credentials, "replaceCredential", async () => { replacements++; return true; }); + const first = await Promise.all(Array.from({ length: 8 }, () => refreshLegacyToken("owner-failure", "gho_first"))); + expect(first).to.deep.equal(Array(8).fill(null)); + expect(await refreshLegacyToken("owner-failure", "gho_first")).to.equal(null); + expect(calls).to.equal(1); expect(replacements).to.equal(0); + await refreshLegacyToken("owner-failure", "gho_reconnected"); expect(calls).to.equal(2); + config.CLIENT_SECRET = "rotated-secret"; + await refreshLegacyToken("owner-failure", "gho_first"); expect(calls).to.equal(3); + }); + it("skips PAT and GitHub App tokens and disabled OAuth", async () => { + oauth(); stub(globalThis, "fetch", () => { throw Error("unexpected refresh"); }); + for (const token of ["ghp_token", "ghs_token", "ghu_token", "ghr_token", "github_pat_token"]) expect(await refreshLegacyToken("owner", token)).to.equal(null); + config.GITHUB_OAUTH_ENABLED = false; expect(await refreshLegacyToken("owner", "gho_token")).to.equal(null); + }); + it("keeps a newer login when the refresh compare-and-swap loses", async () => { + oauth(); stub(globalThis, "fetch", async () => ({ ok: true, status: 200, json: async () => ({ token: "refreshed" }) })); + stub(credentials, "replaceCredential", async () => false); + stub(credentials, "getCredentialToken", async () => "new-login"); + expect(await refreshLegacyToken("owner-race", "gho_old")).to.equal("new-login"); + }); + it("backs off on network failures while leaving the current token intact", async () => { + oauth(); let calls = 0; + stub(globalThis, "fetch", async () => { calls++; throw Error("network failure"); }); + expect(await refreshLegacyToken("owner-network", "gho_old")).to.equal(null); + expect(await refreshLegacyToken("owner-network", "gho_old")).to.equal(null); + expect(calls).to.equal(1); + }); + it("bounds failure caches and expires entries so repaired content can be retried", () => { + let now = 0; const cache = new ExpiringMap(2, () => now); + cache.set("a", true, 30); cache.set("b", true, 30); cache.set("c", true, 30); + expect(cache.get("a")).to.equal(undefined); expect(cache.get("b")).to.equal(true); + 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 () => { + let token = "credential-a", attempts = 0, now = Date.now(); + stub(Date, "now", () => now); + stub(storage, "fileInfo", async () => { throw Object.assign(Error("missing"), { code: "ENOENT" }); }); + stub(storage, "write", async (_repo, _path, input) => { for await (const chunk of input) void chunk; }); + const source = new GitHubStream({ repoId: "missing-fixture", organization: "owner", repoName: "repo", commit: "first", getToken: () => token }); + source.downloadWithFallback = async () => { attempts++; throw Object.assign(Error("missing"), { response: { statusCode: 404 } }); }; + const read = () => source.getFileContentCache("missing.js", "missing-fixture", () => ({ sha: "sha" })); + for (let i = 0; i < 3; i++) { try { await read(); } catch (error) { expect(error.httpStatus || error.response?.statusCode).to.equal(404); } } + expect(attempts).to.equal(1); + token = "credential-b"; await read().catch(() => {}); expect(attempts).to.equal(2); + source.data.commit = "second"; await read().catch(() => {}); expect(attempts).to.equal(3); + now += 30001; await read().catch(() => {}); expect(attempts).to.equal(4); + source.downloadWithFallback = async () => { attempts++; throw Object.assign(Error("forbidden"), { response: { statusCode: 403 } }); }; + now += 30001; await read().catch(() => {}); await read().catch(() => {}); expect(attempts).to.equal(6); + }); + it("does not cache a storage 404 as a missing upstream file", async () => { + let attempts = 0; + stub(storage, "fileInfo", async () => { throw Object.assign(Error("missing"), { code: "ENOENT" }); }); + stub(storage, "write", async (_repo, _path, input) => { + for await (const chunk of input) void chunk; + throw Object.assign(Error("storage unavailable"), { httpStatus: 404 }); + }); + const source = new GitHubStream({ repoId: "storage-failure-fixture", organization: "owner", repoName: "repo", commit: "first", getToken: () => "token" }); + source.downloadWithFallback = async () => { attempts++; return require("node:stream").Readable.from(["content"]); }; + for (let i = 0; i < 2; i++) { + try { await source.getFileContentCache("file.js", source.data.repoId, () => ({ sha: "sha" })); throw Error("expected storage failure"); } + catch (error) { expect(error.message).to.equal("storage unavailable"); } + } + expect(attempts).to.equal(2); + }); + it("returns a client error from the real options route for an inaccessible source", async () => { + const utils = require("../src/server/routes/route-utils"); + const repo = { assertNotArchived() {}, status: "error", options: {}, + model: { statusDate: new Date(), statusMessage: "repo_not_found", source: {} } }; + stub(utils, "getRepo", async () => repo); stub(utils, "getUser", async () => { throw Error("anonymous"); }); + const router = require("../src/server/routes/repository-public").default; + const handler = router.stack.find(layer => layer.route?.path === "/:repoId/options").route.stack[0].handle; + const res = { header() {}, status(value) { this.statusCode = value; return this; }, json(value) { this.body = value; } }; + await handler({ params: { repoId: "inaccessible" } }, res); + expect(res.statusCode).to.equal(404); expect(res.body).to.deep.equal({ error: "repo_not_found" }); + repo.model.statusMessage = "unexpected_bug"; + await handler({ params: { repoId: "inaccessible" } }, res); expect(res.statusCode).to.equal(500); + }); + it("returns a structured 429 and the actual store retry time", () => { + const headers = {}, res = { setHeader: (key, value) => { headers[key] = value; }, status(value) { this.statusCode = value; return this; }, json(value) { this.body = value; } }; + const resetAt = Date.now() + 10000; + sendRequestRateLimit({ rateLimit: { resetTime: new Date(resetAt) } }, res); + expect(res.statusCode).to.equal(429); expect(res.body).to.deep.equal({ error: "rate_limited", resetAt }); + expect(Number(headers["Retry-After"])).to.be.within(9, 10); + }); + it("flags native unhandled stream errors and restarts without reporting private log fields", () => { + const summary = logSummary(); + observeLine(summary, "throw er; // Unhandled 'error' event"); + observeLine(summary, '2026-10-09T00:00:00Z ERROR [process] process fatal error {"code":"process_fatal_error","message":"private data"}'); + observeLine(summary, '2026-10-09T00:00:00Z INFO [requests] request {"status":502,"outcome":"completed","ms":100,"url":"private"}'); + expect(summary.fatalErrors).to.equal(2); expect(summary.completed5xx).to.equal(1); expect(JSON.stringify(summary)).not.to.include("private"); + expect(containerSummary({ Name: "/streamer", RestartCount: 3, State: { Running: true, Health: { Status: "healthy" }, OOMKilled: false, StartedAt: "now" } }).restartCount).to.equal(3); + }); +}); diff --git a/test/request-monitoring.test.js b/test/request-monitoring.test.js index 4b29ecc..404bae9 100644 --- a/test/request-monitoring.test.js +++ b/test/request-monitoring.test.js @@ -149,7 +149,7 @@ describe("bounded request monitoring", function () { try { const input = await source.getFileContentCache("file.txt", source.data.repoId, () => ({ sha: "blob", size: 7 })); input.destroy(); res.send("ok"); - } catch (error) { res.status(502).send(error === upstreamError ? "original error" : "unexpected error"); } + } catch (error) { res.status(502).send(error.cause === upstreamError ? "original error" : "unexpected error"); } }); try { await listen(); const response = await fetch(server, "/"); diff --git a/test/route-utils-handleError.test.js b/test/route-utils-handleError.test.js index 895d615..b231966 100644 --- a/test/route-utils-handleError.test.js +++ b/test/route-utils-handleError.test.js @@ -38,6 +38,24 @@ after(function () { }); describe("route-utils.handleError", function () { + for (const status of [401, 403, 406, 429, 502]) { + it(`preserves Octokit status ${status} without exposing upstream messages`, function () { + const res = makeRes(); + handleError(Object.assign(new Error("private upstream details"), { status }), res); + expect(res.statusCode).to.equal(status); + expect(res.body).to.deep.equal({ error: "internal_error" }); + }); + } + it("ignores non-error HTTP statuses and rejects invalid status values", function () { + for (const status of [200, "406", 999]) { + const res = makeRes(); handleError(Object.assign(new Error("unknown"), { status }), res); + expect(res.statusCode).to.equal(500); + } + }); + it("preserves got response statusCode", function () { + const res = makeRes(); handleError(Object.assign(new Error("upstream"), { response: { statusCode: 403 } }), res); + expect(res.statusCode).to.equal(403); + }); it("uses error.httpStatus when present", function () { const res = makeRes(); const err = new AnonymousError("boom", { httpStatus: 418 });