fix: cancel token-gate waits and detect complete LFS prefixes

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<void> {
+function abortableGateLookup(lookup: Promise<number>, signal: AbortSignal): Promise<number> {
+  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<void> {
+  signal?.throwIfAborted();
   const key = githubQuotaKey(token);
   const localGate = tokenGates.get(key);
   let waitMs = 0;
@@ -96,7 +108,9 @@ async function waitForTokenGate(token: string): Promise<void> {
     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<void> {
     waitMs,
     resetAt: new Date(resetAt).toISOString(),
   });
-  await new Promise((resolve) => setTimeout(resolve, waitMs));
+  await delay(waitMs, undefined, { signal });
   if (localGate) tokenGates.delete(key);
 }
 

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;

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.

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"); });