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/source/GitHubStream.ts b/src/core/source/GitHubStream.ts index 9059756..7b92778 100644 --- a/src/core/source/GitHubStream.ts +++ b/src/core/source/GitHubStream.ts @@ -149,7 +149,7 @@ export default class GitHubStream extends GitHubBase { } catch (error) { if (!token || githubTokenContext(token)?.publicRepository || (error as { response?: { statusCode?: number } }).response?.statusCode !== 404) throw error; - await waitForTokenGate(token); + 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" }, @@ -186,7 +186,7 @@ export default class GitHubStream extends GitHubBase { if (checked) { yield chunk; continue; } probe = Buffer.concat([probe, Buffer.from(chunk)]); if (probe.length < 150) continue; - 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/")) { throw new AnonymousError("upstream_error", { httpStatus: 502 }); } checked = true; yield probe; @@ -294,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; 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/production-followup-fixes.test.js b/test/production-followup-fixes.test.js index 871fbdd..af62acd 100644 --- a/test/production-followup-fixes.test.js +++ b/test/production-followup-fixes.test.js @@ -7,6 +7,8 @@ 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"); @@ -182,7 +184,7 @@ describe("production follow-up fixes", function () { 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([content]); + 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 }; @@ -203,11 +205,35 @@ describe("production follow-up fixes", function () { expect(error.message).to.equal("upstream_error"); expect(error.httpStatus).to.equal(502); expect(calls).to.have.length(1); }); } - it("does not serve an unresolved LFS pointer as file content", async () => { - const pointer = Buffer.from("version https://git-lfs.github.com/spec/v1\noid sha256:" + "a".repeat(64) + "\nsize 10\n"); - const { stream } = rawFallback("https://raw.githubusercontent.com/owner/repo/pinned/file.bin", pointer); - 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 ["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"); });