From a1ed5a8bad9973e5c546738197b8153853979bf9 Mon Sep 17 00:00:00 2001 From: Desmond Chai <66651820+dezchai@users.noreply.github.com> Date: Fri, 28 Aug 2026 02:29:34 -0700 Subject: [PATCH] Stop PR status polling from rate-limiting your entire GitHub account (#3825) * Batch GitHub PR status polls into aliased GraphQL requests The PR status poller ran one `gh pr view` (GraphQL-backed) plus a per-PR facts query for every workspace on its own timer, so cost scaled linearly with workspace count and exhausted the 5,000 pts/hour GraphQL quota machine-wide past ~50 workspaces (#3587, #2470). Replace the per-target timers with a global scheduler that snaps poll targets to a shared 5s grid and flushes all due targets together: - One aliased discovery query per batch of up to 25 targets (grouped by GH host) resolves PR identity, state, mergeability, and the merge facts that previously required a second GraphQL call per PR. - A second aliased query fetches statusCheckRollup contexts only for the PRs that matched. Checks for merged/closed PRs are frozen and cached by repo#number@headOid, so idle workspaces settle to the discovery query alone. - A failed batch backs off every target in it with one request spent; healthy aliases are recovered from the body gh still prints on stdout when a single alias errors (ForgeCommandError now carries stdout for this). - Targets whose origin slug/host is still resolving, and fork repos whose PR lives in the parent, fall back to the legacy per-target path for that cycle. Unowned lookups filter out other forks' PRs that share a branch name. Fixes #3587. Addresses the API budget in #2470. * fix(server): reserve the shared GitHub polling budget --------- Co-authored-by: Mohamed Boudra --- docs/forge-providers.md | 5 + .../server/src/services/forge-cli-command.ts | 13 + .../src/services/github-service.test.ts | 741 +++++++++++ .../server/src/services/github-service.ts | 1119 ++++++++++++++++- 4 files changed, 1854 insertions(+), 24 deletions(-) diff --git a/docs/forge-providers.md b/docs/forge-providers.md index 128e0f81da..f69c7eb64a 100644 --- a/docs/forge-providers.md +++ b/docs/forge-providers.md @@ -218,3 +218,8 @@ are stale, run `npm run build:server`. (`manual`) from blocking jobs (`manual` plus `action_required`). A failed allowed-failure job carries the `warning` trait, and `canceling` remains active until GitLab reports a terminal status. +- GitHub PR polling owns one account-wide GraphQL budget. Coordinate retained + targets per host, batch their reads, and stop until GitHub's reset time when + the reserve is exhausted. Never add a per-target GitHub request to the poll + path. Resolve fork PRs through their parent repository without abandoning the + shared batch. diff --git a/packages/server/src/services/forge-cli-command.ts b/packages/server/src/services/forge-cli-command.ts index 8ff91cc92b..798f79c656 100644 --- a/packages/server/src/services/forge-cli-command.ts +++ b/packages/server/src/services/forge-cli-command.ts @@ -151,6 +151,12 @@ export interface ForgeCommandFailureParams { cwd: string; exitCode: number | null; stderr: string; + /** + * Whatever the failing command printed to stdout. GraphQL batch callers use + * this to salvage partial `data` when one aliased sub-query errors but the + * others succeeded (the CLI exits non-zero yet still prints the body). + */ + stdout?: string; } export class ForgeCommandError extends Error { @@ -159,6 +165,7 @@ export class ForgeCommandError extends Error { readonly cwd: string; readonly exitCode: number | null; readonly stderr: string; + readonly stdout?: string; constructor(label: { brand: string; binary: string }, params: ForgeCommandFailureParams) { super(`${label.brand} CLI command failed: ${label.binary} ${params.args.join(" ")}`); @@ -166,6 +173,9 @@ export class ForgeCommandError extends Error { this.cwd = params.cwd; this.exitCode = params.exitCode; this.stderr = params.stderr; + if (params.stdout !== undefined) { + this.stdout = params.stdout; + } } } @@ -227,6 +237,7 @@ export function normalizeCliCommandError(options: NormalizeCliCommandErrorOption return options.createMissingError(); } const stderr = bufferOrStringToString(failure.stderr); + const stdout = bufferOrStringToString(failure.stdout); const message = failure.message ?? ""; if (options.isAuthFailureText(stderr) || options.isAuthFailureText(message)) { return options.createAuthError(stderr); @@ -239,6 +250,7 @@ export function normalizeCliCommandError(options: NormalizeCliCommandErrorOption stderr: stderr || `${options.commandName} was terminated before completing (timed out after ${options.timeoutMs}ms or exceeded the output limit)`, + ...(stdout ? { stdout } : {}), }); } return options.createCommandError({ @@ -246,6 +258,7 @@ export function normalizeCliCommandError(options: NormalizeCliCommandErrorOption cwd: options.cwd, exitCode: typeof failure.code === "number" ? failure.code : null, stderr: stderr || message, + ...(stdout ? { stdout } : {}), }); } diff --git a/packages/server/src/services/github-service.test.ts b/packages/server/src/services/github-service.test.ts index d916cea03a..cb7b273615 100644 --- a/packages/server/src/services/github-service.test.ts +++ b/packages/server/src/services/github-service.test.ts @@ -241,6 +241,76 @@ function currentPullRequestGithubFactsJson(overrides: Record = }); } +function batchPollPrNodeJson(overrides: Record = {}): Record { + return { + number: 42, + url: "https://github.com/acme/widgets/pull/42", + title: "Widget PR", + state: "MERGED", + isDraft: false, + baseRefName: "main", + headRefName: "feat-a", + headRefOid: "oid-42", + mergedAt: "2026-08-01T00:00:00Z", + mergeable: "UNKNOWN", + reviewDecision: null, + headRepositoryOwner: { login: "acme" }, + mergeStateStatus: "CLEAN", + autoMergeRequest: null, + viewerCanEnableAutoMerge: false, + viewerCanDisableAutoMerge: false, + viewerCanMergeAsAdmin: false, + viewerCanUpdateBranch: false, + isMergeQueueEnabled: false, + isInMergeQueue: false, + ...overrides, + }; +} + +function batchPollRepositoryJson( + nodes: Array>, + overrides: Record = {}, +): Record { + return { + isFork: false, + autoMergeAllowed: true, + mergeCommitAllowed: true, + squashMergeAllowed: true, + rebaseMergeAllowed: true, + viewerDefaultMergeMethod: "SQUASH", + pullRequests: { nodes }, + ...overrides, + }; +} + +function batchPollChecksAliasJson( + contexts: Array>, +): Record { + return { + pullRequest: { + commits: { nodes: [{ commit: { statusCheckRollup: { contexts: { nodes: contexts } } } }] }, + }, + }; +} + +function batchPollStatusJson(aliases: Record): string { + return JSON.stringify({ data: aliases }); +} + +function batchPollStatusWithRateLimitJson(input: { + aliases: Record; + limit: number; + remaining: number; + resetAt: string; +}): string { + return JSON.stringify({ + data: { + ...input.aliases, + rateLimit: { limit: input.limit, remaining: input.remaining, resetAt: input.resetAt }, + }, + }); +} + function createCurrentPullRequestStatus( overrides: Partial = {}, ): CurrentPullRequestStatus { @@ -976,6 +1046,677 @@ describe("ForgeService", () => { expect(currentPullRequestStatusCalls(runner.calls)).toHaveLength(1); }); + it("batches PR status polls across workspaces into one GraphQL request", async () => { + let now = 0; + const runner = createScriptedRunner([ + batchPollStatusJson({ + t0: batchPollRepositoryJson([ + batchPollPrNodeJson({ + number: 41, + url: "https://github.com/acme/widgets/pull/41", + headRefName: "feat-a", + headRefOid: "oid-a", + }), + ]), + t1: batchPollRepositoryJson([ + batchPollPrNodeJson({ + number: 52, + url: "https://github.com/acme/gadgets/pull/52", + headRefName: "feat-b", + headRefOid: "oid-b", + }), + ]), + }), + batchPollStatusJson({ + t0: batchPollChecksAliasJson([ + { __typename: "StatusContext", context: "ci", state: "SUCCESS" }, + ]), + t1: batchPollChecksAliasJson([ + { __typename: "StatusContext", context: "ci", state: "SUCCESS" }, + ]), + }), + batchPollStatusJson({ + t0: batchPollRepositoryJson([ + batchPollPrNodeJson({ + number: 41, + url: "https://github.com/acme/widgets/pull/41", + headRefName: "feat-a", + headRefOid: "oid-a", + }), + ]), + t1: batchPollRepositoryJson([ + batchPollPrNodeJson({ + number: 52, + url: "https://github.com/acme/gadgets/pull/52", + headRefName: "feat-b", + headRefOid: "oid-b", + }), + ]), + }), + ]); + const slugs: Record = { "/ws-a": "acme/widgets", "/ws-b": "acme/gadgets" }; + const service = createGitHubService({ + ttlMs: 0, + runner: runner.runner, + resolveGhPath: async () => "/usr/bin/gh", + resolveRepoHost: async () => null, + resolveRepoSlug: async (cwd) => slugs[cwd] ?? null, + now: () => now, + }); + const statusesA: Array = []; + const statusesB: Array = []; + + const subscriptionA = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-a", + headRef: "feat-a", + headSha: "oid-a", + onStatus: (status) => statusesA.push(status), + }); + const subscriptionB = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-b", + headRef: "feat-b", + headSha: "oid-b", + onStatus: (status) => statusesB.push(status), + }); + await vi.advanceTimersByTimeAsync(0); + await flushMicrotasks(); + + expect(currentPullRequestStatusCalls(runner.calls)).toHaveLength(0); + expect(runner.calls).toHaveLength(2); + expect(runner.calls[0]?.args.slice(0, 2)).toEqual(["api", "graphql"]); + expect(runner.calls[0]?.args[3]).toContain('t0: repository(owner: "acme", name: "widgets")'); + expect(runner.calls[0]?.args[3]).toContain('t1: repository(owner: "acme", name: "gadgets")'); + expect(runner.calls[0]?.args[3]).toContain('headRefName: "feat-a"'); + expect(runner.calls[0]?.args[3]).toContain('headRefName: "feat-b"'); + expect(runner.calls[1]?.args[3]).toContain("pullRequest(number: 41)"); + expect(runner.calls[1]?.args[3]).toContain("pullRequest(number: 52)"); + expect(statusesA).toEqual([ + expect.objectContaining({ + number: 41, + state: "merged", + isMerged: true, + checksStatus: "success", + forgeSpecific: expect.objectContaining({ forge: "github", mergeStateStatus: "CLEAN" }), + }), + ]); + expect(statusesB).toEqual([ + expect.objectContaining({ number: 52, state: "merged", checksStatus: "success" }), + ]); + + // Merged PR checks are frozen: the second cycle reuses the cached rollup + // and issues only the discovery request. + now = EXPECTED_GITHUB_SLOW_POLL_MS; + await vi.advanceTimersByTimeAsync(EXPECTED_GITHUB_SLOW_POLL_MS); + await flushMicrotasks(); + + expect(runner.calls).toHaveLength(3); + expect(statusesA).toHaveLength(2); + expect(statusesA[1]).toEqual(expect.objectContaining({ number: 41, checksStatus: "success" })); + + subscriptionA?.unsubscribe(); + subscriptionB?.unsubscribe(); + service.dispose?.(); + }); + + it("refetches checks for open PRs on every batched poll at fast cadence", async () => { + let now = 0; + const openNode = batchPollPrNodeJson({ + number: 7, + url: "https://github.com/acme/widgets/pull/7", + state: "OPEN", + mergedAt: null, + headRefName: "feat-a", + headRefOid: "oid-open", + }); + const pendingChecks = batchPollChecksAliasJson([ + { __typename: "StatusContext", context: "ci", state: "PENDING" }, + ]); + const runner = createScriptedRunner([ + batchPollStatusJson({ t0: batchPollRepositoryJson([openNode]) }), + batchPollStatusJson({ t0: pendingChecks }), + batchPollStatusJson({ t0: batchPollRepositoryJson([openNode]) }), + batchPollStatusJson({ t0: pendingChecks }), + ]); + const service = createGitHubService({ + ttlMs: 0, + runner: runner.runner, + resolveGhPath: async () => "/usr/bin/gh", + resolveRepoHost: async () => null, + resolveRepoSlug: async () => "acme/widgets", + now: () => now, + }); + const statuses: Array = []; + + const subscription = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-a", + headRef: "feat-a", + onStatus: (status) => statuses.push(status), + }); + await vi.advanceTimersByTimeAsync(0); + await flushMicrotasks(); + + expect(runner.calls).toHaveLength(2); + expect(statuses).toEqual([ + expect.objectContaining({ number: 7, state: "open", checksStatus: "pending" }), + ]); + + now = EXPECTED_GITHUB_FAST_POLL_MS; + await vi.advanceTimersByTimeAsync(EXPECTED_GITHUB_FAST_POLL_MS); + await flushMicrotasks(); + + expect(runner.calls).toHaveLength(4); + expect(statuses).toHaveLength(2); + + subscription?.unsubscribe(); + service.dispose?.(); + }); + + it("backs off all batched targets on a failed batch without per-target fallback calls", async () => { + let now = 0; + const runner = createScriptedRunner([{ error: new Error("network down") }]); + const slugs: Record = { "/ws-a": "acme/widgets", "/ws-b": "acme/gadgets" }; + const service = createGitHubService({ + ttlMs: 0, + runner: runner.runner, + resolveGhPath: async () => "/usr/bin/gh", + resolveRepoHost: async () => null, + resolveRepoSlug: async (cwd) => slugs[cwd] ?? null, + now: () => now, + }); + const errors: unknown[] = []; + + const subscriptionA = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-a", + headRef: "feat-a", + onError: (error) => errors.push(error), + }); + const subscriptionB = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-b", + headRef: "feat-b", + onError: (error) => errors.push(error), + }); + await vi.advanceTimersByTimeAsync(0); + await flushMicrotasks(); + + expect(runner.calls).toHaveLength(1); + expect(currentPullRequestStatusCalls(runner.calls)).toHaveLength(0); + expect(errors).toHaveLength(2); + + subscriptionA?.unsubscribe(); + subscriptionB?.unsubscribe(); + service.dispose?.(); + }); + + it("falls back to the per-target lookup when a fork repository has no matching PR", async () => { + let now = 0; + const runner = createScriptedRunner([ + batchPollStatusJson({ t0: batchPollRepositoryJson([], { isFork: true }) }), + currentPullRequestJson({ headRefName: "feat-a" }), + currentPullRequestGithubFactsJson(), + ]); + const service = createGitHubService({ + ttlMs: 0, + runner: runner.runner, + resolveGhPath: async () => "/usr/bin/gh", + resolveRepoHost: async () => null, + resolveRepoSlug: async () => "forkowner/widgets", + now: () => now, + }); + const statuses: Array = []; + + const subscription = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-a", + headRef: "feat-a", + onStatus: (status) => statuses.push(status), + }); + await vi.advanceTimersByTimeAsync(0); + await flushMicrotasks(); + + expect(runner.calls[0]?.args.slice(0, 2)).toEqual(["api", "graphql"]); + expect(currentPullRequestStatusCalls(runner.calls)).toHaveLength(1); + expect(statuses).toEqual([expect.objectContaining({ number: 42, state: "open" })]); + + subscription?.unsubscribe(); + service.dispose?.(); + }); + + it("redirects a fork poll to its parent and reuses that batch address", async () => { + let now = 0; + const parent = { owner: { login: "upstream" }, name: "widgets" }; + const forkPr = batchPollPrNodeJson({ + url: "https://github.com/upstream/widgets/pull/42", + headRefName: "feat-a", + headRefOid: "oid-a", + headRepositoryOwner: { login: "forkowner" }, + }); + const runner = createScriptedRunner([ + batchPollStatusJson({ + t0: batchPollRepositoryJson([], { isFork: true, parent }), + }), + batchPollStatusJson({ t0: batchPollRepositoryJson([forkPr]) }), + batchPollStatusJson({ + t0: batchPollChecksAliasJson([ + { __typename: "StatusContext", context: "ci", state: "SUCCESS" }, + ]), + }), + batchPollStatusJson({ t0: batchPollRepositoryJson([forkPr]) }), + ]); + const service = createGitHubService({ + ttlMs: 0, + runner: runner.runner, + resolveGhPath: async () => "/usr/bin/gh", + resolveRepoHost: async () => null, + resolveRepoSlug: async () => "forkowner/widgets", + now: () => now, + }); + const statuses: Array = []; + + const subscription = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-a", + headRef: "feat-a", + headSha: "oid-a", + onStatus: (status) => statuses.push(status), + }); + await vi.advanceTimersByTimeAsync(0); + await flushMicrotasks(); + + expect(currentPullRequestStatusCalls(runner.calls)).toHaveLength(0); + expect(runner.calls[0]?.args[3]).toContain( + 't0: repository(owner: "forkowner", name: "widgets")', + ); + expect(runner.calls[1]?.args[3]).toContain( + 't0: repository(owner: "upstream", name: "widgets")', + ); + expect(statuses).toEqual([ + expect.objectContaining({ number: 42, repoOwner: "upstream", checksStatus: "success" }), + ]); + + now = EXPECTED_GITHUB_SLOW_POLL_MS; + await vi.advanceTimersByTimeAsync(EXPECTED_GITHUB_SLOW_POLL_MS); + await flushMicrotasks(); + + expect(runner.calls[3]?.args[3]).toContain( + 't0: repository(owner: "upstream", name: "widgets")', + ); + + subscription?.unsubscribe(); + service.dispose?.(); + }); + + it("treats a missing pullRequests connection as a failed alias", async () => { + let now = 0; + const runner = createScriptedRunner([ + batchPollStatusJson({ + t0: { isFork: false, autoMergeAllowed: true }, + }), + ]); + const service = createGitHubService({ + ttlMs: 0, + runner: runner.runner, + resolveGhPath: async () => "/usr/bin/gh", + resolveRepoHost: async () => null, + resolveRepoSlug: async () => "acme/widgets", + now: () => now, + }); + const statuses: Array = []; + const errors: unknown[] = []; + + const subscription = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-a", + headRef: "feat-a", + onStatus: (status) => statuses.push(status), + onError: (error) => errors.push(error), + }); + await vi.advanceTimersByTimeAsync(0); + await flushMicrotasks(); + + expect(statuses).toEqual([]); + expect(errors).toHaveLength(1); + expect(errors[0]).toBeInstanceOf(GitHubCommandError); + + subscription?.unsubscribe(); + service.dispose?.(); + }); + + it("treats a malformed checks connection as a failed alias", async () => { + let now = 0; + const runner = createScriptedRunner([ + batchPollStatusJson({ + t0: batchPollRepositoryJson([batchPollPrNodeJson({ state: "OPEN", mergedAt: null })]), + }), + batchPollStatusJson({ t0: { pullRequest: {} } }), + ]); + const service = createGitHubService({ + ttlMs: 0, + runner: runner.runner, + resolveGhPath: async () => "/usr/bin/gh", + resolveRepoHost: async () => null, + resolveRepoSlug: async () => "acme/widgets", + now: () => now, + }); + const statuses: Array = []; + const errors: unknown[] = []; + + const subscription = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-a", + headRef: "feat-a", + onStatus: (status) => statuses.push(status), + onError: (error) => errors.push(error), + }); + await vi.advanceTimersByTimeAsync(0); + await flushMicrotasks(); + + expect(statuses).toEqual([]); + expect(errors).toHaveLength(1); + expect(errors[0]).toBeInstanceOf(GitHubCommandError); + + subscription?.unsubscribe(); + service.dispose?.(); + }); + + it("pauses a host until reset when a batch reaches the GraphQL reserve", async () => { + let now = 0; + const resetAt = new Date(EXPECTED_GITHUB_ERROR_BACKOFF_CAP_MS).toISOString(); + const runner = createScriptedRunner([ + batchPollStatusWithRateLimitJson({ + aliases: { t0: batchPollRepositoryJson([batchPollPrNodeJson()]) }, + limit: 5_000, + remaining: 2_000, + resetAt, + }), + batchPollStatusJson({ t0: batchPollRepositoryJson([]) }), + ]); + const service = createGitHubService({ + ttlMs: 0, + runner: runner.runner, + resolveGhPath: async () => "/usr/bin/gh", + resolveRepoHost: async () => null, + resolveRepoSlug: async () => "acme/widgets", + now: () => now, + }); + const errors: unknown[] = []; + + const subscription = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-a", + headRef: "feat-a", + onError: (error) => errors.push(error), + }); + await vi.advanceTimersByTimeAsync(0); + await flushMicrotasks(); + + expect(runner.calls).toHaveLength(1); + expect(runner.calls[0]?.args[3]).toContain("rateLimit"); + expect(errors).toHaveLength(1); + + now = EXPECTED_GITHUB_ERROR_BACKOFF_CAP_MS; + await vi.advanceTimersByTimeAsync(EXPECTED_GITHUB_ERROR_BACKOFF_CAP_MS); + await flushMicrotasks(); + expect(runner.calls).toHaveLength(1); + + now += 1_000; + await vi.advanceTimersByTimeAsync(1_000); + await flushMicrotasks(); + expect(runner.calls).toHaveLength(2); + + subscription?.unsubscribe(); + service.dispose?.(); + }); + + it("loads the reset time once when GitHub rejects a batch for rate limiting", async () => { + let now = 0; + const rateLimitError = Object.assign(new Error("gh exited 1"), { + code: 1, + stderr: "GraphQL: API rate limit already exceeded for user ID 123", + }); + const runner = createScriptedRunner([ + { error: rateLimitError }, + JSON.stringify({ resources: { graphql: { remaining: 0, reset: 300 } } }), + batchPollStatusJson({ t0: batchPollRepositoryJson([]) }), + ]); + const service = createGitHubService({ + ttlMs: 0, + runner: runner.runner, + resolveGhPath: async () => "/usr/bin/gh", + resolveRepoHost: async () => null, + resolveRepoSlug: async () => "acme/widgets", + now: () => now, + }); + + const subscription = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-a", + headRef: "feat-a", + }); + await vi.advanceTimersByTimeAsync(0); + await flushMicrotasks(); + + expect(runner.calls.map((call) => call.args.slice(0, 2))).toEqual([ + ["api", "graphql"], + ["api", "rate_limit"], + ]); + + now = EXPECTED_GITHUB_ERROR_BACKOFF_CAP_MS; + await vi.advanceTimersByTimeAsync(EXPECTED_GITHUB_ERROR_BACKOFF_CAP_MS); + await flushMicrotasks(); + expect(runner.calls).toHaveLength(2); + + now += 1_000; + await vi.advanceTimersByTimeAsync(1_000); + await flushMicrotasks(); + expect(runner.calls).toHaveLength(3); + + subscription?.unsubscribe(); + service.dispose?.(); + }); + + it("keeps refetching checks for a closed PR while its checks are still pending", async () => { + let now = 0; + const closedNode = batchPollPrNodeJson({ + number: 9, + url: "https://github.com/acme/widgets/pull/9", + state: "CLOSED", + mergedAt: null, + headRefName: "feat-a", + headRefOid: "oid-closed", + }); + const runner = createScriptedRunner([ + batchPollStatusJson({ t0: batchPollRepositoryJson([closedNode]) }), + batchPollStatusJson({ + t0: batchPollChecksAliasJson([ + { __typename: "StatusContext", context: "ci", state: "PENDING" }, + ]), + }), + batchPollStatusJson({ t0: batchPollRepositoryJson([closedNode]) }), + batchPollStatusJson({ + t0: batchPollChecksAliasJson([ + { __typename: "StatusContext", context: "ci", state: "SUCCESS" }, + ]), + }), + ]); + const service = createGitHubService({ + ttlMs: 0, + runner: runner.runner, + resolveGhPath: async () => "/usr/bin/gh", + resolveRepoHost: async () => null, + resolveRepoSlug: async () => "acme/widgets", + now: () => now, + }); + const statuses: Array = []; + + const subscription = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-a", + headRef: "feat-a", + headSha: "oid-closed", + onStatus: (status) => statuses.push(status), + }); + await vi.advanceTimersByTimeAsync(0); + await flushMicrotasks(); + + expect(runner.calls).toHaveLength(2); + expect(statuses).toEqual([expect.objectContaining({ number: 9, checksStatus: "pending" })]); + + // Pending checks are never frozen, so the next cycle re-runs the checks + // query and observes the completed run. + now = EXPECTED_GITHUB_FAST_POLL_MS; + await vi.advanceTimersByTimeAsync(EXPECTED_GITHUB_FAST_POLL_MS); + await flushMicrotasks(); + + expect(runner.calls).toHaveLength(4); + expect(statuses[1]).toEqual(expect.objectContaining({ number: 9, checksStatus: "success" })); + + subscription?.unsubscribe(); + service.dispose?.(); + }); + + it("falls back to the per-target lookup when the batch candidate page is full", async () => { + let now = 0; + const crowdedNodes = Array.from({ length: 10 }, (_, index) => + batchPollPrNodeJson({ + number: 100 + index, + url: `https://github.com/acme/widgets/pull/${100 + index}`, + state: "OPEN", + mergedAt: null, + headRefName: "feat-a", + headRepositoryOwner: { login: "stranger" }, + }), + ); + const runner = createScriptedRunner([ + batchPollStatusJson({ t0: batchPollRepositoryJson(crowdedNodes) }), + currentPullRequestJson({ headRefName: "feat-a" }), + currentPullRequestGithubFactsJson(), + ]); + const service = createGitHubService({ + ttlMs: 0, + runner: runner.runner, + resolveGhPath: async () => "/usr/bin/gh", + resolveRepoHost: async () => null, + resolveRepoSlug: async () => "acme/widgets", + now: () => now, + }); + const statuses: Array = []; + + const subscription = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-a", + headRef: "feat-a", + onStatus: (status) => statuses.push(status), + }); + await vi.advanceTimersByTimeAsync(0); + await flushMicrotasks(); + + // Ten same-named fork PRs filled the page, so absence of a self-owned PR + // is inconclusive and the target resolves through the legacy path. + expect(currentPullRequestStatusCalls(runner.calls)).toHaveLength(1); + expect(statuses).toEqual([expect.objectContaining({ number: 42, state: "open" })]); + + subscription?.unsubscribe(); + service.dispose?.(); + }); + + it("ignores another fork owner's same-named branch in batched polls without an owner hint", async () => { + let now = 0; + const runner = createScriptedRunner([ + batchPollStatusJson({ + t0: batchPollRepositoryJson([ + batchPollPrNodeJson({ + number: 900, + url: "https://github.com/acme/widgets/pull/900", + state: "OPEN", + mergedAt: null, + headRefName: "main", + headRepositoryOwner: { login: "stranger" }, + }), + ]), + }), + ]); + const service = createGitHubService({ + ttlMs: 0, + runner: runner.runner, + resolveGhPath: async () => "/usr/bin/gh", + resolveRepoHost: async () => null, + resolveRepoSlug: async () => "acme/widgets", + now: () => now, + }); + const statuses: Array = []; + + const subscription = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-a", + headRef: "main", + onStatus: (status) => statuses.push(status), + }); + await vi.advanceTimersByTimeAsync(0); + await flushMicrotasks(); + + expect(runner.calls).toHaveLength(1); + expect(statuses).toEqual([null]); + + subscription?.unsubscribe(); + service.dispose?.(); + }); + + it("recovers healthy aliases from a partially failed batch response", async () => { + let now = 0; + const partialBody = batchPollStatusJson({ + t0: batchPollRepositoryJson([ + batchPollPrNodeJson({ + number: 41, + url: "https://github.com/acme/widgets/pull/41", + headRefName: "feat-a", + headRefOid: "oid-a", + }), + ]), + t1: null, + }); + const runner = createScriptedRunner([ + { + error: Object.assign(new Error("gh exited 1"), { + code: 1, + stdout: partialBody, + stderr: "GraphQL: Could not resolve to a Repository", + }), + }, + batchPollStatusJson({ + t0: batchPollChecksAliasJson([ + { __typename: "StatusContext", context: "ci", state: "SUCCESS" }, + ]), + }), + ]); + const slugs: Record = { "/ws-a": "acme/widgets", "/ws-b": "acme/missing" }; + const service = createGitHubService({ + ttlMs: 0, + runner: runner.runner, + resolveGhPath: async () => "/usr/bin/gh", + resolveRepoHost: async () => null, + resolveRepoSlug: async (cwd) => slugs[cwd] ?? null, + now: () => now, + }); + const statuses: Array = []; + const errors: unknown[] = []; + + const subscriptionA = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-a", + headRef: "feat-a", + headSha: "oid-a", + onStatus: (status) => statuses.push(status), + }); + const subscriptionB = service.retainCurrentPullRequestStatusPoll?.({ + cwd: "/ws-b", + headRef: "feat-b", + onError: (error) => errors.push(error), + }); + await vi.advanceTimersByTimeAsync(0); + await flushMicrotasks(); + + expect(runner.calls).toHaveLength(2); + expect(statuses).toEqual([ + expect.objectContaining({ number: 41, state: "merged", checksStatus: "success" }), + ]); + expect(errors).toHaveLength(1); + expect(errors[0]).toBeInstanceOf(GitHubCommandError); + + subscriptionA?.unsubscribe(); + subscriptionB?.unsubscribe(); + service.dispose?.(); + }); + it("fetches PR reviews and issue comments with one GraphQL call sorted chronologically", async () => { const runner = createRunner([pullRequestTimelineJson()]); const service = createGitHubService({ diff --git a/packages/server/src/services/github-service.ts b/packages/server/src/services/github-service.ts index e7eb70a98f..124216674e 100644 --- a/packages/server/src/services/github-service.ts +++ b/packages/server/src/services/github-service.ts @@ -107,6 +107,17 @@ const FAILED_CHECK_JOB_LIMIT = 5; export const GITHUB_POLL_FAST_INTERVAL_MS = 20_000; export const GITHUB_POLL_SLOW_INTERVAL_MS = 120_000; export const GITHUB_POLL_ERROR_BACKOFF_CAP_MS = 300_000; +// The PR status poller batches every due target into aliased GraphQL queries +// instead of one `gh pr view` + facts query per workspace. Cost scales with the +// GraphQL point budget (5,000/hour), so the batch size and the shared poll time +// grid below exist to keep a many-workspace daemon from exhausting the quota +// machine-wide (issues #3587 / #2470). +export const GITHUB_POLL_BATCH_MAX = 25; +export const GITHUB_POLL_ALIGNMENT_MS = 5_000; +const GITHUB_GRAPHQL_RESERVE_RATIO = 0.4; +const GITHUB_GRAPHQL_RESET_GRACE_MS = 1_000; +const BATCH_PR_CANDIDATE_LIMIT = 10; +const FROZEN_PR_CHECKS_CACHE_MAX_ENTRIES = 512; const GITHUB_ENV = { GIT_TERMINAL_PROMPT: "0", } as const; @@ -601,6 +612,308 @@ query PullRequestTimeline($owner: String!, $name: String!, $number: Int!) { } }`; +const BatchPollPrNodeSchema = z.object({ + number: z.number(), + url: z.string().catch(""), + title: z.string().catch(""), + state: z.string().catch(""), + isDraft: z.boolean().optional().catch(false), + baseRefName: z.string().catch(""), + headRefName: z.string().catch(""), + headRefOid: z.string().catch(""), + mergedAt: z.string().nullable().optional(), + mergeable: PullRequestMergeableSchema.optional().default("UNKNOWN"), + reviewDecision: z.unknown().optional(), + headRepositoryOwner: HeadRepositoryOwnerSchema, + mergeStateStatus: z.string().nullable().optional().catch(null), + autoMergeRequest: GitHubAutoMergeRequestSchema, + viewerCanEnableAutoMerge: z.boolean().optional().catch(false), + viewerCanDisableAutoMerge: z.boolean().optional().catch(false), + viewerCanMergeAsAdmin: z.boolean().optional().catch(false), + viewerCanUpdateBranch: z.boolean().optional().catch(false), + isMergeQueueEnabled: z.boolean().optional().catch(false), + isInMergeQueue: z.boolean().optional().catch(false), +}); + +const BatchPollRepositorySchema = z.object({ + owner: z.object({ login: z.string().optional() }).nullable().optional().catch(null), + isFork: z.boolean().optional().catch(false), + parent: z + .object({ owner: z.object({ login: z.string() }), name: z.string() }) + .nullable() + .optional() + .catch(null), + autoMergeAllowed: z.boolean().optional().catch(false), + mergeCommitAllowed: z.boolean().optional().catch(false), + squashMergeAllowed: z.boolean().optional().catch(false), + rebaseMergeAllowed: z.boolean().optional().catch(false), + viewerDefaultMergeMethod: z.string().nullable().optional().catch(null), + pullRequests: z.object({ nodes: z.array(BatchPollPrNodeSchema) }), +}); + +const GitHubGraphqlRateLimitSchema = z.object({ + limit: z.number(), + remaining: z.number(), + resetAt: z.string(), +}); + +const GitHubRateLimitResponseSchema = z.object({ + resources: z.object({ + graphql: z.object({ + remaining: z.number(), + reset: z.number(), + }), + }), +}); + +const BatchChecksCommitSchema = z.object({ + commit: z.object({ + statusCheckRollup: z + .object({ + contexts: z.object({ nodes: z.array(z.unknown()) }), + }) + .nullable() + .optional(), + }), +}); + +const BatchChecksRepositorySchema = z.object({ + pullRequest: z + .object({ + commits: z.object({ nodes: z.array(BatchChecksCommitSchema) }), + }) + .nullable(), +}); + +// GraphQL CheckRun nodes carry the workflow name behind +// checkSuite.workflowRun.workflow.name; the gh CLI flattens it to workflowName. +// Batch results are remapped to the CLI shape so parseStatusCheckRollup stays +// the single rollup parser for both paths. +const BatchCheckRunContextSchema = z.object({ + __typename: z.literal("CheckRun"), + databaseId: z.number().nullable().optional(), + name: z.string(), + status: z.string().nullable().optional(), + conclusion: z.string().nullable().optional(), + detailsUrl: z.string().nullable().optional(), + startedAt: z.string().nullable().optional(), + completedAt: z.string().nullable().optional(), + checkSuite: z + .object({ + workflowRun: z + .object({ + databaseId: z.number().nullable().optional(), + workflow: z.object({ name: z.string().nullable().optional() }).nullable().optional(), + }) + .nullable() + .optional(), + }) + .nullable() + .optional(), +}); + +type BatchPollPrNode = z.infer; +type BatchPollRepository = z.infer; + +const BATCH_PR_STATUS_FRAGMENT = ` +fragment PaseoPollPullRequest on PullRequest { + number + url + title + state + isDraft + baseRefName + headRefName + headRefOid + mergedAt + mergeable + reviewDecision + headRepositoryOwner { + login + } + mergeStateStatus + autoMergeRequest { + enabledAt + mergeMethod + enabledBy { + login + } + } + viewerCanEnableAutoMerge + viewerCanDisableAutoMerge + viewerCanMergeAsAdmin + viewerCanUpdateBranch + isMergeQueueEnabled + isInMergeQueue +}`; + +function batchAlias(index: number): string { + return `t${index}`; +} + +// GraphQL string literals share JSON's escaping rules. +function graphqlString(value: string): string { + return JSON.stringify(value); +} + +function buildBatchPullRequestStatusQuery( + entries: Array<{ owner: string; name: string; headRef: string }>, +): string { + const aliases = entries.map( + (entry, index) => ` ${batchAlias(index)}: repository(owner: ${graphqlString( + entry.owner, + )}, name: ${graphqlString(entry.name)}) { + owner { + login + } + isFork + parent { + owner { + login + } + name + } + autoMergeAllowed + mergeCommitAllowed + squashMergeAllowed + rebaseMergeAllowed + viewerDefaultMergeMethod + pullRequests(headRefName: ${graphqlString( + entry.headRef, + )}, first: ${BATCH_PR_CANDIDATE_LIMIT}, orderBy: {field: CREATED_AT, direction: DESC}) { + nodes { + ...PaseoPollPullRequest + } + } + }`, + ); + return `query PaseoBatchPullRequestStatus { + rateLimit { + limit + remaining + resetAt + } +${aliases.join("\n")} +}\n${BATCH_PR_STATUS_FRAGMENT}`; +} + +function buildBatchPullRequestChecksQuery( + entries: Array<{ owner: string; name: string; number: number }>, +): string { + const aliases = entries.map( + (entry, index) => ` ${batchAlias(index)}: repository(owner: ${graphqlString( + entry.owner, + )}, name: ${graphqlString(entry.name)}) { + pullRequest(number: ${entry.number}) { + commits(last: 1) { + nodes { + commit { + statusCheckRollup { + contexts(first: 100) { + nodes { + __typename + ... on CheckRun { + databaseId + name + status + conclusion + detailsUrl + startedAt + completedAt + checkSuite { + workflowRun { + databaseId + workflow { + name + } + } + } + } + ... on StatusContext { + context + state + targetUrl + createdAt + } + } + } + } + } + } + } + } + }`, + ); + return `query PaseoBatchPullRequestChecks { + rateLimit { + limit + remaining + resetAt + } +${aliases.join("\n")} +}`; +} + +function toBatchStatusCheckRollupNode(node: unknown): unknown { + const parsed = BatchCheckRunContextSchema.safeParse(node); + if (!parsed.success) { + // StatusContext nodes already match the CLI rollup shape. + return node; + } + const { checkSuite, ...rest } = parsed.data; + return { + ...rest, + workflowName: checkSuite?.workflowRun?.workflow?.name ?? null, + checkSuite: { workflowRun: { databaseId: checkSuite?.workflowRun?.databaseId ?? null } }, + }; +} + +function extractGraphqlBatchAliases(body: unknown): Record | null { + if (!body || typeof body !== "object") { + return null; + } + const data = (body as Record).data; + if (!data || typeof data !== "object") { + return null; + } + return data as Record; +} + +function parseGraphqlBatchAliases(stdout: string): Record { + let parsed: unknown; + try { + parsed = JSON.parse(stdout || "{}"); + } catch { + return {}; + } + return extractGraphqlBatchAliases(parsed) ?? {}; +} + +// gh exits non-zero when any aliased sub-query errors but still prints the +// response body, so the healthy aliases in a batch are recoverable. +function extractGraphqlBatchAliasesFromError(error: unknown): Record | null { + if (!(error instanceof GitHubCommandError) || !error.stdout) { + return null; + } + try { + return extractGraphqlBatchAliases(JSON.parse(error.stdout)); + } catch { + return null; + } +} + +function parseGraphqlBatchRateLimit( + stdout: string, +): z.infer | null { + try { + const aliases = extractGraphqlBatchAliases(JSON.parse(stdout)); + const parsed = GitHubGraphqlRateLimitSchema.safeParse(aliases?.rateLimit); + return parsed.success ? parsed.data : null; + } catch { + return null; + } +} + interface CacheEntry { value: unknown; expiresAt: number; @@ -617,6 +930,13 @@ interface GitHubServiceDependencies { * call routes to the workspace's instance instead of github.com. */ resolveRepoHost: (cwd: string) => Promise; + /** + * owner/name slug for a workspace, resolved offline from the origin remote. + * Batched PR status polls need the repository identity up front (GraphQL + * repository(owner:, name:) aliases), unlike `gh pr view` which resolves it + * per invocation. + */ + resolveRepoSlug: (cwd: string) => Promise; } export interface GitHubCommandRunnerOptions { @@ -677,6 +997,18 @@ export class GitHubCommandError extends ForgeCommandError { } } +class GitHubGraphqlPollPausedError extends Error { + readonly remaining: number; + readonly resetAt: string; + + constructor(params: { remaining: number; resetAt: string }) { + super(`GitHub GraphQL polling paused until ${params.resetAt}`); + this.name = "GitHubGraphqlPollPausedError"; + this.remaining = params.remaining; + this.resetAt = params.resetAt; + } +} + export class GitHubEnterpriseHostProbeError extends Error { readonly host: string; override readonly cause: Error; @@ -695,6 +1027,7 @@ interface CreateGitHubServiceOptions { resolveGhPath?: () => Promise; now?: () => number; resolveRepoHost?: (cwd: string) => Promise; + resolveRepoSlug?: (cwd: string) => Promise; } type PullRequestCheckRunNode = z.infer; @@ -720,13 +1053,48 @@ interface GitHubPollTarget { headSha?: string; headRepositoryOwner?: string; retainCount: number; - timer: NodeJS.Timeout | null; + /** Absolute due time on the shared poll grid; null while a poll is in flight. */ + nextDueAt: number | null; + /** + * When the in-flight poll cycle became due. The next cycle is anchored here + * rather than at completion time so slow requests don't drift the poll grid. + */ + pollCycleStartedAt: number | null; latestStatus: CurrentPullRequestStatus | null; consecutiveErrors: number; callbacks: Set<(status: CurrentPullRequestStatus | null) => void>; errorCallbacks: Set<(error: unknown) => void>; } +interface GitHubBatchPollEntry { + target: GitHubPollTarget; + host: string; + originOwner: string; + originName: string; + owner: string; + name: string; + headRepositoryOwner?: string; +} + +interface PendingBatchPullRequestChecks { + entry: GitHubBatchPollEntry; + node: BatchPollPrNode; + repository: BatchPollRepository; + /** Frozen-checks cache key for non-open PRs, or null when checks may still change. */ + cacheKey: string | null; +} + +interface GitHubBatchQueryResponse { + aliases: Record; + stdout: string; +} + +interface GitHubBatchDistribution { + pendingChecks: PendingBatchPullRequestChecks[]; + legacyFallback: GitHubPollTarget[]; + redirected: GitHubBatchPollEntry[]; +} + interface ResolvedPullRequestCandidate { status: CurrentPullRequestStatus; headSha?: string; @@ -740,18 +1108,47 @@ export function createGitHubService(options: CreateGitHubServiceOptions = {}): G resolveGhPath: options.resolveGhPath ?? resolveGhPath, now: options.now ?? Date.now, resolveRepoHost: options.resolveRepoHost ?? resolveGitHubEnterpriseHost, + resolveRepoSlug: options.resolveRepoSlug ?? resolveGitHubSlugFromOrigin, }; // A resolved enterprise host is cached permanently; a null resolution (no // host, or the auth probe said no) expires so `gh auth login --hostname` - // run after the first probe is picked up without a daemon restart. + // run after the first probe is picked up without a daemon restart. The + // `resolved` marker survives a TTL refresh so poll batching can keep using + // the last known host synchronously while the refresh is in flight. const repoHostByCwd = new Map< string, - { promise: Promise; expiresAt: number | null } + { + promise: Promise; + expiresAt: number | null; + resolved?: { value: string | null }; + } + >(); + // owner/name from the origin remote, cached until invalidate() (which fires + // on remote changes) so batched polls can group targets synchronously. Null + // resolutions expire like the host cache's so transient git failures don't + // permanently degrade a cwd to per-target polling. + const repoSlugByCwd = new Map< + string, + { + promise: Promise; + expiresAt: number | null; + resolved?: { value: string | null }; + } >(); const cache = new Map(); const inFlight = new Map(); const pollTargets = new Map(); const checkLogTailCache = new Map(); + // Rollup nodes for merged/closed PRs keyed by repo#number@headOid: their + // checks can no longer change, so each is fetched at most once per daemon run. + const frozenPullRequestChecksCache = new Map(); + const pollRepositoryRedirects = new Map< + string, + { owner: string; name: string; headRepositoryOwner: string } + >(); + const pollPausedUntilByHost = new Map(); + const rateLimitResetLoads = new Map>(); + let githubPollTimer: NodeJS.Timeout | null = null; let api!: GitHubService; async function cached(params: { @@ -815,8 +1212,11 @@ export function createGitHubService(options: CreateGitHubServiceOptions = {}): G .resolveRepoHost(cwd) .then((host) => { const current = repoHostByCwd.get(cwd); - if (host === null && current?.promise === pending) { - current.expiresAt = deps.now() + REPO_HOST_NULL_TTL_MS; + if (current?.promise === pending) { + current.resolved = { value: host }; + if (host === null) { + current.expiresAt = deps.now() + REPO_HOST_NULL_TTL_MS; + } } return host; }) @@ -826,10 +1226,75 @@ export function createGitHubService(options: CreateGitHubServiceOptions = {}): G } throw error; }); - repoHostByCwd.set(cwd, { promise: pending, expiresAt: null }); + repoHostByCwd.set(cwd, { + promise: pending, + expiresAt: null, + ...(cachedHost?.resolved ? { resolved: cachedHost.resolved } : {}), + }); return pending; } + function peekRepoHost(cwd: string): { settled: boolean; value: string | null } { + const entry = repoHostByCwd.get(cwd); + if (!entry) { + void resolveRepoHostCached(cwd).catch(() => {}); + return { settled: false, value: null }; + } + return entry.resolved + ? { settled: true, value: entry.resolved.value } + : { settled: false, value: null }; + } + + function resolveRepoSlugCached(cwd: string): Promise { + const cachedSlug = repoSlugByCwd.get(cwd); + if (cachedSlug && (cachedSlug.expiresAt === null || deps.now() < cachedSlug.expiresAt)) { + return cachedSlug.promise; + } + // A null slug expires: resolveRepoSlug also returns null for transient git + // failures (lock contention, timeout), and a permanent null would silently + // pin this cwd to the legacy per-target poll path. The `resolved` marker + // survives the refresh so batching keeps the last known slug meanwhile. + const entry: { + promise: Promise; + expiresAt: number | null; + resolved?: { value: string | null }; + } = { + promise: deps + .resolveRepoSlug(cwd) + .then((slug) => { + if (repoSlugByCwd.get(cwd) === entry) { + entry.resolved = { value: slug }; + if (slug === null) { + entry.expiresAt = deps.now() + REPO_HOST_NULL_TTL_MS; + } + } + return slug; + }) + .catch(() => { + if (repoSlugByCwd.get(cwd) === entry) { + entry.resolved = { value: null }; + entry.expiresAt = deps.now() + REPO_HOST_NULL_TTL_MS; + } + return null; + }), + expiresAt: null, + ...(cachedSlug?.resolved ? { resolved: cachedSlug.resolved } : {}), + }; + repoSlugByCwd.set(cwd, entry); + return entry.promise; + } + + function peekRepoSlug(cwd: string): { settled: boolean; value: string | null } { + const entry = repoSlugByCwd.get(cwd); + if (!entry) { + void resolveRepoSlugCached(cwd); + return { settled: false, value: null }; + } + return entry.resolved + ? { settled: true, value: entry.resolved.value } + : { settled: false, value: null }; + } + async function run(args: string[], runOptions: GitHubCommandRunnerOptions): Promise { const ghPath = await deps.resolveGhPath(); if (!ghPath) { @@ -923,14 +1388,569 @@ export function createGitHubService(options: CreateGitHubServiceOptions = {}): G if (target.retainCount <= 0) { return; } - if (target.timer) { - clearTimeout(target.timer); + const now = deps.now(); + const baseline = target.pollCycleStartedAt ?? now; + target.pollCycleStartedAt = null; + const pauseUntil = getPollPauseUntil(target.cwd); + if (delayMs <= 0) { + target.nextDueAt = Math.max(now, pauseUntil); + armGitHubPollTimer(); + return; + } + // Non-immediate polls snap to a shared time grid so targets converge on the + // same tick and one batched GraphQL request covers all of them. The next + // cycle anchors at the previous cycle's due time so slow requests don't + // drift the grid — but when the cycle overran its slot entirely, re-anchor + // at completion: scheduling in the past would poll back-to-back precisely + // when GitHub is slow or erroring. + const dueAt = alignGitHubPollTime(baseline + delayMs); + const nextDueAt = dueAt >= now ? dueAt : alignGitHubPollTime(now + delayMs); + target.nextDueAt = Math.max(nextDueAt, pauseUntil); + armGitHubPollTimer(); + } + + function getPollPauseUntil(cwd: string): number { + const host = peekRepoHost(cwd); + if (!host.settled) { + return 0; + } + const key = host.value ?? ""; + const pauseUntil = pollPausedUntilByHost.get(key) ?? 0; + if (pauseUntil <= deps.now()) { + pollPausedUntilByHost.delete(key); + return 0; + } + return pauseUntil; + } + + function alignGitHubPollTime(dueAt: number): number { + return Math.ceil(dueAt / GITHUB_POLL_ALIGNMENT_MS) * GITHUB_POLL_ALIGNMENT_MS; + } + + function armGitHubPollTimer(): void { + let earliest: number | null = null; + for (const target of pollTargets.values()) { + if (target.retainCount <= 0 || target.nextDueAt === null) { + continue; + } + if (earliest === null || target.nextDueAt < earliest) { + earliest = target.nextDueAt; + } + } + if (githubPollTimer) { + clearTimeout(githubPollTimer); + githubPollTimer = null; + } + if (earliest === null) { + return; + } + githubPollTimer = setTimeout( + () => { + githubPollTimer = null; + void flushDueGitHubPolls(); + }, + Math.max(0, earliest - deps.now()), + ); + } + + async function flushDueGitHubPolls(): Promise { + const now = deps.now(); + const due: GitHubPollTarget[] = []; + for (const target of pollTargets.values()) { + if (target.retainCount > 0 && target.nextDueAt !== null && target.nextDueAt <= now) { + target.nextDueAt = null; + target.pollCycleStartedAt = now; + due.push(target); + } + } + armGitHubPollTimer(); + if (due.length === 0) { + return; } - target.timer = setTimeout(() => { - target.timer = null; - void runGitHubPoll(target); - }, delayMs); + const legacyTargets: GitHubPollTarget[] = []; + const batchGroups = new Map(); + for (const target of due) { + const host = peekRepoHost(target.cwd); + const slug = peekRepoSlug(target.cwd); + const [owner, name, extra] = slug.value?.split("/") ?? []; + // A cwd whose host or slug is still resolving (or has no origin slug at + // all) polls through the legacy per-target path this cycle; batch + // grouping needs both facts synchronously. + if (!host.settled || !slug.settled || !owner || !name || extra !== undefined) { + legacyTargets.push(target); + continue; + } + const key = host.value ?? ""; + const group = batchGroups.get(key) ?? []; + group.push({ + target, + host: key, + originOwner: owner, + originName: name, + owner, + name, + }); + batchGroups.set(key, group); + } + + await Promise.all([ + ...legacyTargets.map((target) => runGitHubPoll(target)), + ...[...batchGroups.values()].map((group) => runGitHubPollGroup(group, now)), + ]); + } + + async function runGitHubPollGroup( + entries: GitHubBatchPollEntry[], + startedAt: number, + ): Promise { + for (let start = 0; start < entries.length; start += GITHUB_POLL_BATCH_MAX) { + const chunk = entries.slice(start, start + GITHUB_POLL_BATCH_MAX); + if ((pollPausedUntilByHost.get(chunk[0]?.host ?? "") ?? 0) > deps.now()) { + return; + } + await runGitHubPollBatch(chunk, startedAt); + } + } + + function pauseGitHubPollsAtReserve( + host: string, + rateLimit: z.infer, + ): boolean { + if (rateLimit.remaining > rateLimit.limit * GITHUB_GRAPHQL_RESERVE_RATIO) { + return false; + } + const resetAt = Date.parse(rateLimit.resetAt); + if (!Number.isFinite(resetAt)) { + return false; + } + pauseGitHubPollHostUntil(host, resetAt + GITHUB_GRAPHQL_RESET_GRACE_MS); + return true; + } + + function pauseGitHubPollHostUntil(host: string, pauseUntil: number): void { + if (pauseUntil <= deps.now()) { + return; + } + const currentPause = pollPausedUntilByHost.get(host) ?? 0; + pollPausedUntilByHost.set(host, Math.max(currentPause, pauseUntil)); + for (const target of pollTargets.values()) { + const targetHost = peekRepoHost(target.cwd); + if (!targetHost.settled || (targetHost.value ?? "") !== host) { + continue; + } + target.pollCycleStartedAt = null; + target.nextDueAt = Math.max(target.nextDueAt ?? 0, pauseUntil); + } + armGitHubPollTimer(); + } + + async function pauseGitHubPollsAfterRateLimit( + entry: GitHubBatchPollEntry, + error: unknown, + ): Promise { + if (!isGitHubRateLimitError(error)) { + return false; + } + const resetAt = await loadGitHubGraphqlResetAt(entry); + if (resetAt === null) { + return false; + } + pauseGitHubPollHostUntil(entry.host, resetAt + GITHUB_GRAPHQL_RESET_GRACE_MS); + return true; + } + + function loadGitHubGraphqlResetAt(entry: GitHubBatchPollEntry): Promise { + const existing = rateLimitResetLoads.get(entry.host); + if (existing) { + return existing; + } + const pending = runGhJson( + ["api", "rate_limit"], + { cwd: entry.target.cwd }, + GitHubRateLimitResponseSchema, + "{}", + ) + .then((response) => response.resources.graphql.reset * 1_000) + .catch(() => null) + .finally(() => { + if (rateLimitResetLoads.get(entry.host) === pending) { + rateLimitResetLoads.delete(entry.host); + } + }); + rateLimitResetLoads.set(entry.host, pending); + return pending; + } + + async function runGitHubPollBatch( + entries: GitHubBatchPollEntry[], + startedAt: number, + ): Promise { + const addressedEntries = entries.map(addressGitHubBatchEntry); + const firstEntry = addressedEntries[0]; + if (!firstEntry) { + return; + } + const query = buildBatchPullRequestStatusQuery( + addressedEntries.map((entry) => ({ + owner: entry.owner, + name: entry.name, + headRef: entry.target.headRef, + })), + ); + const args = ["api", "graphql", "-f", `query=${query}`]; + const response = await loadGitHubPollBatchResponse({ + entries: addressedEntries, + firstEntry, + args, + }); + if (!response) { + return; + } + const rateLimit = parseGraphqlBatchRateLimit(response.stdout); + if (rateLimit && pauseGitHubPollsAtReserve(firstEntry.host, rateLimit)) { + const error = new GitHubGraphqlPollPausedError(rateLimit); + reportGitHubBatchError(addressedEntries, error); + return; + } + + const distribution = distributeGitHubPollBatch({ + entries: addressedEntries, + aliases: response.aliases, + args, + startedAt, + }); + if (distribution.pendingChecks.length > 0) { + await loadBatchPullRequestChecks(distribution.pendingChecks, startedAt); + } + await Promise.all(distribution.legacyFallback.map((target) => runGitHubPoll(target))); + if (distribution.redirected.length > 0) { + await runGitHubPollBatch(distribution.redirected, startedAt); + } + } + + async function loadGitHubPollBatchResponse(input: { + entries: GitHubBatchPollEntry[]; + firstEntry: GitHubBatchPollEntry; + args: string[]; + }): Promise { + try { + const stdout = await run(input.args, { cwd: input.firstEntry.target.cwd }); + return { aliases: parseGraphqlBatchAliases(stdout), stdout }; + } catch (error) { + const isPaused = await pauseGitHubPollsAfterRateLimit(input.firstEntry, error); + const partial = extractGraphqlBatchAliasesFromError(error); + if (isPaused || !partial) { + reportGitHubBatchError(input.entries, error); + return null; + } + const stdout = error instanceof GitHubCommandError ? (error.stdout ?? "") : ""; + return { aliases: partial, stdout }; + } + } + + function reportGitHubBatchError(entries: GitHubBatchPollEntry[], error: unknown): void { + for (const entry of entries) { + updatePollTargetAfterError(entry.target, error); + } + } + + function distributeGitHubPollBatch(input: { + entries: GitHubBatchPollEntry[]; + aliases: Record; + args: string[]; + startedAt: number; + }): GitHubBatchDistribution { + const distribution: GitHubBatchDistribution = { + pendingChecks: [], + legacyFallback: [], + redirected: [], + }; + for (const [index, entry] of input.entries.entries()) { + const aliasValue = input.aliases[batchAlias(index)]; + const repository = aliasValue ? BatchPollRepositorySchema.safeParse(aliasValue) : null; + if (!repository?.success) { + updatePollTargetAfterError( + entry.target, + new GitHubCommandError({ + args: input.args, + cwd: entry.target.cwd, + exitCode: null, + stderr: `GitHub batch PR status response is missing repository ${entry.owner}/${entry.name}`, + }), + ); + continue; + } + if (repository.data.isFork && repository.data.parent) { + const forkOwner = repository.data.owner?.login ?? entry.owner; + pollRepositoryRedirects.set(batchRepositoryRedirectKey(entry), { + owner: repository.data.parent.owner.login, + name: repository.data.parent.name, + headRepositoryOwner: forkOwner, + }); + distribution.redirected.push(entry); + continue; + } + const node = selectBatchPollNode(entry, repository.data); + if (!node) { + // Legacy fallback covers the two ways "no match" can be wrong: a fork + // whose PR lives in the parent repository, and a full candidate page — + // ten newer same-named fork PRs can crowd the checkout's own PR out of + // the first:N window, so absence is only conclusive on a partial page. + if ( + repository.data.isFork || + repository.data.pullRequests.nodes.length >= BATCH_PR_CANDIDATE_LIMIT + ) { + distribution.legacyFallback.push(entry.target); + } else { + finalizeBatchPollTarget({ + entry, + node: null, + repository: repository.data, + rollup: null, + startedAt: input.startedAt, + }); + } + continue; + } + const frozenKey = + node.state !== "OPEN" && node.headRefOid.length > 0 + ? `${entry.owner}/${entry.name}#${node.number}@${node.headRefOid}` + : null; + const cachedRollup = frozenKey ? frozenPullRequestChecksCache.get(frozenKey) : undefined; + if (frozenKey && cachedRollup) { + rememberFrozenPullRequestChecks(frozenKey, cachedRollup); + finalizeBatchPollTarget({ + entry, + node, + repository: repository.data, + rollup: cachedRollup, + startedAt: input.startedAt, + }); + } else { + distribution.pendingChecks.push({ + entry, + node, + repository: repository.data, + cacheKey: frozenKey, + }); + } + } + return distribution; + } + + function addressGitHubBatchEntry(entry: GitHubBatchPollEntry): GitHubBatchPollEntry { + const redirect = pollRepositoryRedirects.get(batchRepositoryRedirectKey(entry)); + return redirect + ? { + ...entry, + owner: redirect.owner, + name: redirect.name, + headRepositoryOwner: redirect.headRepositoryOwner, + } + : entry; + } + + function batchRepositoryRedirectKey(entry: GitHubBatchPollEntry): string { + return `${entry.host}\n${entry.originOwner}/${entry.originName}`; + } + + function selectBatchPollNode( + entry: GitHubBatchPollEntry, + repository: BatchPollRepository, + ): BatchPollPrNode | null { + // Prefer the canonical owner from the response: the origin slug can carry + // stale casing or a pre-transfer owner that repository() redirects from. + const selfOwner = (repository.owner?.login ?? entry.owner).toLowerCase(); + const headRepositoryOwner = entry.target.headRepositoryOwner ?? entry.headRepositoryOwner; + const nodeByCandidate = new Map(); + const candidates: ResolvedPullRequestCandidate[] = []; + for (const node of repository.pullRequests.nodes) { + const candidate = toCurrentPullRequestCandidate( + toBatchCurrentPullRequestItem(node, undefined), + entry.target.headRef, + ); + if (!candidate) { + continue; + } + // pullRequests(headRefName:) also matches forks whose branch shares the + // name. Without a fork-owner hint the workspace branch lives in this + // repository, so only self-owned heads are real candidates (mirrors + // `gh pr view` resolving the current branch's own PR). + if (!headRepositoryOwner && candidate.headRepositoryOwner?.toLowerCase() !== selfOwner) { + continue; + } + nodeByCandidate.set(candidate, node); + candidates.push(candidate); + } + const match = pickPullRequestCandidate({ + candidates, + headRef: entry.target.headRef, + headSha: entry.target.headSha, + headRepositoryOwner, + }); + return match ? (nodeByCandidate.get(match) ?? null) : null; + } + + async function loadBatchPullRequestChecks( + pending: PendingBatchPullRequestChecks[], + startedAt: number, + ): Promise { + const firstItem = pending[0]; + if (!firstItem) { + return; + } + const query = buildBatchPullRequestChecksQuery( + pending.map((item) => ({ + owner: item.entry.owner, + name: item.entry.name, + number: item.node.number, + })), + ); + const args = ["api", "graphql", "-f", `query=${query}`]; + let aliasData: Record; + let responseStdout: string; + try { + responseStdout = await run(args, { cwd: firstItem.entry.target.cwd }); + aliasData = parseGraphqlBatchAliases(responseStdout); + } catch (error) { + const isPaused = await pauseGitHubPollsAfterRateLimit(firstItem.entry, error); + if (isPaused) { + for (const item of pending) { + updatePollTargetAfterError(item.entry.target, error); + } + return; + } + const partial = extractGraphqlBatchAliasesFromError(error); + if (!partial) { + for (const item of pending) { + updatePollTargetAfterError(item.entry.target, error); + } + return; + } + aliasData = partial; + responseStdout = error instanceof GitHubCommandError ? (error.stdout ?? "") : ""; + } + + const rateLimit = parseGraphqlBatchRateLimit(responseStdout); + if (rateLimit) { + pauseGitHubPollsAtReserve(firstItem.entry.host, rateLimit); + } + + for (const [index, item] of pending.entries()) { + const aliasValue = aliasData[batchAlias(index)]; + const parsed = aliasValue ? BatchChecksRepositorySchema.safeParse(aliasValue) : null; + if (!parsed?.success || !parsed.data.pullRequest) { + updatePollTargetAfterError( + item.entry.target, + new GitHubCommandError({ + args, + cwd: item.entry.target.cwd, + exitCode: null, + stderr: `GitHub batch PR checks response is missing ${item.entry.owner}/${item.entry.name}#${item.node.number}`, + }), + ); + continue; + } + const contexts = + parsed.data.pullRequest.commits.nodes[0]?.commit.statusCheckRollup?.contexts.nodes ?? []; + const rollup = contexts.map(toBatchStatusCheckRollupNode); + // Freeze only fully terminal rollups: a PR closed while CI is running + // would otherwise pin "pending" checks (and the fast poll cadence) for + // the daemon's lifetime. Reruns that flip a terminal check back to + // pending on the same SHA are deliberately not tracked — stale terminal + // checks on a closed/merged PR are cosmetic. + const hasPendingChecks = parseStatusCheckRollup(rollup).some( + (check) => check.status === "pending", + ); + if (item.cacheKey && !hasPendingChecks) { + rememberFrozenPullRequestChecks(item.cacheKey, rollup); + } + finalizeBatchPollTarget({ + entry: item.entry, + node: item.node, + repository: item.repository, + rollup, + startedAt, + }); + } + } + + function rememberFrozenPullRequestChecks(key: string, rollup: unknown[]): void { + frozenPullRequestChecksCache.delete(key); + frozenPullRequestChecksCache.set(key, rollup); + while (frozenPullRequestChecksCache.size > FROZEN_PR_CHECKS_CACHE_MAX_ENTRIES) { + const oldestKey = frozenPullRequestChecksCache.keys().next().value; + if (!oldestKey) { + break; + } + frozenPullRequestChecksCache.delete(oldestKey); + } + } + + function finalizeBatchPollTarget(input: { + entry: GitHubBatchPollEntry; + node: BatchPollPrNode | null; + repository: BatchPollRepository; + rollup: unknown[] | null; + /** When this poll cycle became due; cache writes newer than this win. */ + startedAt: number; + }): void { + const { entry, node, repository } = input; + let status: CurrentPullRequestStatus | null = null; + if (node) { + const facts: GitHubPullRequestStatusFacts = { + mergeStateStatus: node.mergeStateStatus ?? null, + autoMergeRequest: toGitHubAutoMergeRequest(node.autoMergeRequest), + viewerCanEnableAutoMerge: node.viewerCanEnableAutoMerge ?? false, + viewerCanDisableAutoMerge: node.viewerCanDisableAutoMerge ?? false, + viewerCanMergeAsAdmin: node.viewerCanMergeAsAdmin ?? false, + viewerCanUpdateBranch: node.viewerCanUpdateBranch ?? false, + repository: toGitHubRepositoryMergePolicy(repository), + isMergeQueueEnabled: node.isMergeQueueEnabled ?? false, + isInMergeQueue: node.isInMergeQueue ?? false, + }; + const built = toCurrentPullRequestStatus( + toBatchCurrentPullRequestItem(node, input.rollup ?? undefined), + entry.target.headRef, + ); + status = built ? { ...built, forgeSpecific: { forge: "github", ...facts } } : null; + } + const cacheKey = buildCacheKey({ + cwd: entry.target.cwd, + method: "getCurrentPullRequestStatus", + args: { + headRef: entry.target.headRef, + headSha: entry.target.headSha, + headRepositoryOwner: entry.target.headRepositoryOwner, + }, + }); + // A direct or forced read can finish between this batch becoming due and + // its checks query returning. Its cache write is newer than this cycle, so + // converge on it instead of overwriting fresher data with the batch's. + const existing = cache.get(cacheKey); + if (existing && existing.expiresAt - ttlMs > input.startedAt) { + updatePollTargetAfterSuccess({ + cwd: entry.target.cwd, + headRef: entry.target.headRef, + headSha: entry.target.headSha, + headRepositoryOwner: entry.target.headRepositoryOwner, + status: existing.value as CurrentPullRequestStatus | null, + notify: true, + }); + return; + } + // Mirror the cached() write of getCurrentPullRequestStatus so direct reads + // between polls reuse the batch result instead of spawning gh again. + cache.set(cacheKey, { value: status, cwd: entry.target.cwd, expiresAt: deps.now() + ttlMs }); + updatePollTargetAfterSuccess({ + cwd: entry.target.cwd, + headRef: entry.target.headRef, + headSha: entry.target.headSha, + headRepositoryOwner: entry.target.headRepositoryOwner, + status, + notify: true, + }); } async function runGitHubPoll(target: GitHubPollTarget): Promise { @@ -943,19 +1963,24 @@ export function createGitHubService(options: CreateGitHubServiceOptions = {}): G reason: "self-heal-github", }); } catch (error) { - target.consecutiveErrors += 1; - for (const callback of target.errorCallbacks) { - callback(error); - } - scheduleGitHubPoll(target); + updatePollTargetAfterError(target, error); } } - function closeGitHubPollTarget(target: GitHubPollTarget): void { - if (target.timer) { - clearTimeout(target.timer); - target.timer = null; + function updatePollTargetAfterError(target: GitHubPollTarget, error: unknown): void { + if (target.retainCount <= 0) { + return; } + target.consecutiveErrors += 1; + for (const callback of target.errorCallbacks) { + callback(error); + } + scheduleGitHubPoll(target); + } + + function closeGitHubPollTarget(target: GitHubPollTarget): void { + target.nextDueAt = null; + target.pollCycleStartedAt = null; target.retainCount = 0; target.callbacks.clear(); target.errorCallbacks.clear(); @@ -1518,13 +2543,18 @@ export function createGitHubService(options: CreateGitHubServiceOptions = {}): G headSha: input.headSha, headRepositoryOwner: input.headRepositoryOwner, retainCount: 0, - timer: null, + nextDueAt: null, + pollCycleStartedAt: null, latestStatus: null, consecutiveErrors: 0, callbacks: new Set(), errorCallbacks: new Set(), }; pollTargets.set(key, target); + // Warm the host and slug caches so the first poll flush can already + // batch this target instead of taking the legacy per-target path. + void resolveRepoHostCached(input.cwd).catch(() => {}); + void resolveRepoSlugCached(input.cwd); } const isNewlyRetained = target.retainCount === 0; @@ -1560,6 +2590,7 @@ export function createGitHubService(options: CreateGitHubServiceOptions = {}): G } closeGitHubPollTarget(target); pollTargets.delete(key); + armGitHubPollTimer(); }, }; }, @@ -1577,16 +2608,24 @@ export function createGitHubService(options: CreateGitHubServiceOptions = {}): G inFlight.delete(key); } } - // Drop the cached host so a changed remote re-resolves GH_HOST instead of - // routing later gh calls to the previous instance. + // Drop the cached host and slug so a changed remote re-resolves GH_HOST + // and the batch-poll repository identity instead of reusing stale ones. repoHostByCwd.delete(input.cwd); + repoSlugByCwd.delete(input.cwd); }, dispose() { + if (githubPollTimer) { + clearTimeout(githubPollTimer); + githubPollTimer = null; + } for (const target of pollTargets.values()) { closeGitHubPollTarget(target); } pollTargets.clear(); + pollRepositoryRedirects.clear(); + pollPausedUntilByHost.clear(); + rateLimitResetLoads.clear(); }, }; @@ -1880,6 +2919,15 @@ function isStatusCheckRollupPermissionError(error: unknown): boolean { return error.stderr.toLowerCase().includes("statuscheckrollup"); } +function isGitHubRateLimitError(error: unknown): boolean { + return ( + error instanceof GitHubCommandError && + /(?:rate limit|rate_limit).*(?:exceed|exhaust)|(?:exceed|exhaust).*rate limit/i.test( + error.stderr, + ) + ); +} + async function resolveCurrentPullRequestView(options: { cwd: string; headRef: string; @@ -2182,6 +3230,29 @@ function toGitHubRepositoryMergePolicy( }; } +// Reshape a batched GraphQL PR node into the gh CLI JSON shape so candidate +// picking and status construction stay shared with the per-target path. +function toBatchCurrentPullRequestItem( + node: BatchPollPrNode, + statusCheckRollup: unknown[] | undefined, +): CurrentPullRequestStatusItem { + return { + number: node.number, + url: node.url, + title: node.title, + state: node.state, + isDraft: node.isDraft ?? false, + baseRefName: node.baseRefName, + headRefName: node.headRefName, + ...(node.headRefOid ? { headRefOid: node.headRefOid } : {}), + mergedAt: node.mergedAt ?? null, + ...(statusCheckRollup ? { statusCheckRollup } : {}), + reviewDecision: node.reviewDecision, + mergeable: node.mergeable, + headRepositoryOwner: node.headRepositoryOwner, + }; +} + function toCurrentPullRequestCandidate( item: CurrentPullRequestStatusItem, fallbackHeadRefName: string,