diff --git a/migrations/0087_github_rate_limit_admission_key.sql b/migrations/0087_github_rate_limit_admission_key.sql new file mode 100644 index 0000000000..7657726802 --- /dev/null +++ b/migrations/0087_github_rate_limit_admission_key.sql @@ -0,0 +1,5 @@ +ALTER TABLE github_rate_limit_observations + ADD COLUMN admission_key TEXT; + +CREATE INDEX IF NOT EXISTS github_rate_limit_observations_admission_observed_idx + ON github_rate_limit_observations (admission_key, observed_at); diff --git a/src/auth/github-oauth.ts b/src/auth/github-oauth.ts index 943fe027c8..283842d37f 100644 --- a/src/auth/github-oauth.ts +++ b/src/auth/github-oauth.ts @@ -5,6 +5,7 @@ import { timingSafeEqual, } from "./security"; import { recordAuditEvent } from "../db/repositories"; +import { timeoutFetch } from "../github/client"; import type { JsonValue } from "../types"; type GitHubDeviceCodeResponse = { @@ -178,7 +179,7 @@ export async function createSessionFromGitHubToken( }); throw new Error("github_token_audience_invalid"); } - const response = await fetch("https://api.github.com/user", { + const response = await timeoutFetch("https://api.github.com/user", { headers: { accept: "application/vnd.github+json", authorization: `Bearer ${githubToken}`, @@ -206,7 +207,7 @@ export async function createSessionFromGitHubToken( // isn't configured, the token can't be vouched for, so it is rejected. async function verifyTokenBelongsToApp(env: Env, githubToken: string): Promise { if (!env.GITHUB_OAUTH_CLIENT_ID || !env.GITHUB_OAUTH_CLIENT_SECRET) return false; - const response = await fetch(`https://api.github.com/applications/${env.GITHUB_OAUTH_CLIENT_ID}/token`, { + const response = await timeoutFetch(`https://api.github.com/applications/${env.GITHUB_OAUTH_CLIENT_ID}/token`, { method: "POST", headers: { accept: "application/vnd.github+json", diff --git a/src/db/repositories.ts b/src/db/repositories.ts index 501a25edd5..6ca42a6f95 100644 --- a/src/db/repositories.ts +++ b/src/db/repositories.ts @@ -992,6 +992,7 @@ export async function recordGitHubRateLimitObservation(env: Env, observation: Gi await db.insert(githubRateLimitObservations).values({ id: observation.id ?? crypto.randomUUID(), repoFullName: observation.repoFullName, + admissionKey: observation.admissionKey, resource: observation.resource, path: observation.path, statusCode: observation.statusCode, @@ -4059,6 +4060,7 @@ function toGitHubRateLimitObservationRecord(row: typeof githubRateLimitObservati return { id: row.id, repoFullName: row.repoFullName, + admissionKey: row.admissionKey, resource: row.resource === "graphql" ? "graphql" : "rest", path: row.path, statusCode: row.statusCode, diff --git a/src/db/schema.ts b/src/db/schema.ts index 1ef7944ecf..aae9020190 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -168,6 +168,7 @@ export const githubRateLimitObservations = sqliteTable( { id: text("id").primaryKey(), repoFullName: text("repo_full_name"), + admissionKey: text("admission_key"), resource: text("resource").notNull().default("rest"), path: text("path").notNull(), statusCode: integer("status_code").notNull(), @@ -177,6 +178,7 @@ export const githubRateLimitObservations = sqliteTable( observedAt: text("observed_at").notNull().$defaultFn(() => nowIso()), }, (table) => ({ + admissionObserved: index("github_rate_limit_observations_admission_observed_idx").on(table.admissionKey, table.observedAt), repoObserved: index("github_rate_limit_observations_repo_observed_idx").on(table.repoFullName, table.observedAt), reset: index("github_rate_limit_observations_reset_idx").on(table.resetAt), }), diff --git a/src/github/app.ts b/src/github/app.ts index 2913142265..70bf258ba6 100644 --- a/src/github/app.ts +++ b/src/github/app.ts @@ -5,6 +5,7 @@ import { } from "../orb/broker-client"; import { clearGitHubResponseCacheForTest, + githubRateLimitAdmissionKeyForInstallation, makeInstallationOctokit, timeoutFetch, } from "./client"; @@ -337,7 +338,11 @@ export async function getRepositoryCollaboratorPermission( const token = await createInstallationToken(env, installationId); const response = await timeoutFetch( `https://api.github.com/repos/${encodeURIComponent(owner)}/${encodeURIComponent(name)}/collaborators/${encodeURIComponent(login)}/permission`, - { headers: githubHeaders(`Bearer ${token}`) }, + { + headers: githubHeaders(`Bearer ${token}`), + githubRateLimitAdmission: true, + githubRateLimitAdmissionKey: githubRateLimitAdmissionKeyForInstallation(installationId), + }, ); if (response.status === 404) return null; if (!response.ok) { @@ -579,7 +584,7 @@ async function createOrUpdateNamedCheckRun( return await withInstallationTokenRetry(env, installationId, async (token) => { // makeInstallationOctokit injects the shared per-request timeout (a stalled PATCH can never orphan the // in_progress check) AND suppresses the check-run writes under a non-live mode (dry-run / pause / freeze). - const octokit = makeInstallationOctokit(env, token, check.mode); + const octokit = makeInstallationOctokit(env, token, check.mode, githubRateLimitAdmissionKeyForInstallation(installationId)); // Point the merge-box "Details" link at the repo's Gittensory maintainer panel instead of GitHub's generic // check page. Spread conditionally so a URL-construction failure (null) just omits it. (#audit-details-url) const detailsUrl = maintainerControlPanelUrl(env, repoFullName); diff --git a/src/github/backfill.ts b/src/github/backfill.ts index 04c9a7b263..9544ca50fa 100644 --- a/src/github/backfill.ts +++ b/src/github/backfill.ts @@ -2779,7 +2779,7 @@ function githubRestHeaders(token?: string): HeadersInit { } async function githubGraphQl(env: Env, query: string, token: string): Promise { - const response = await fetch("https://api.github.com/graphql", { + const response = await timeoutFetch("https://api.github.com/graphql", { method: "POST", headers: { accept: "application/vnd.github+json", diff --git a/src/github/client.ts b/src/github/client.ts index b7dc0f4955..890a498513 100644 --- a/src/github/client.ts +++ b/src/github/client.ts @@ -45,6 +45,23 @@ export function setGitHubResponseCache(cache: GitHubResponseCache | null): void export type GitHubCacheClass = "branch_protection" | "metadata"; type EnvLookup = Record; +export type GitHubTimeoutFetchInit = RequestInit & { + /** Opt in to using this response's REST bucket headers for self-host queue admission control. */ + githubRateLimitAdmission?: boolean; + /** Stable actor key for admission control. Installation-token reads should use the installation id. */ + githubRateLimitAdmissionKey?: string; +}; +export type GitHubRateLimitAdmissionKey = string; +export type LocalGitHubRestRateLimitObservation = { + remaining: number; + resetAt: string; + observedAtMs: number; +}; +const latestRestRateLimitObservations = new Map(); + +export function githubRateLimitAdmissionKeyForInstallation(installationId: number): GitHubRateLimitAdmissionKey { + return `installation:${Math.trunc(installationId)}`; +} /** Only cache explicitly stable GitHub REST reads. PR/issue/comment/label/event/check/status reads are mutable * review inputs and must always reflect the current GitHub state. Exported for tests. */ @@ -86,6 +103,13 @@ export function githubResponseCacheTtlSeconds(cls: GitHubCacheClass, env: EnvLoo return positiveEnvSeconds(env, "GITHUB_METADATA_CACHE_TTL_SECONDS", DEFAULT_METADATA_TTL_SECONDS); } +function isCacheableGithubResponseStatus(cls: GitHubCacheClass, status: number): boolean { + if (status === 200) return true; + // Branch-protection permissions are repo/base-branch metadata. Cache stable negative answers too, + // otherwise a missing permission can burn the REST bucket on every PR pass. + return cls === "branch_protection" && (status === 403 || status === 404); +} + function hasConditionalRequestHeader(headers: Headers): boolean { return headers.has("if-none-match") || headers.has("if-modified-since") || headers.has("if-match") || headers.has("if-unmodified-since"); } @@ -102,6 +126,30 @@ function recordGitHubCacheMetric(result: "hit" | "miss" | "set" | "coalesced" | incr(GITHUB_RESPONSE_CACHE_METRIC, { result, class: cls }); } +function parseRateLimitInt(value: string | null): number | null { + if (value === null) return null; + const parsed = Number(value); + return Number.isFinite(parsed) ? parsed : null; +} + +function observeGitHubRestRateLimit(url: string, response: Response, admissionKey: GitHubRateLimitAdmissionKey): void { + if (!url.startsWith(`${GITHUB_API_PREFIX}/`)) return; + const resource = response.headers.get("x-ratelimit-resource"); + if (resource !== null && resource !== "core") return; + const remaining = parseRateLimitInt(response.headers.get("x-ratelimit-remaining")); + const reset = parseRateLimitInt(response.headers.get("x-ratelimit-reset")); + if (remaining === null || reset === null) return; + latestRestRateLimitObservations.set(admissionKey, { + remaining, + resetAt: new Date(reset * 1000).toISOString(), + observedAtMs: Date.now(), + }); +} + +export function latestGitHubRestRateLimitObservation(admissionKey: GitHubRateLimitAdmissionKey): LocalGitHubRestRateLimitObservation | null { + return latestRestRateLimitObservations.get(admissionKey) ?? null; +} + async function sha256Short(value: string): Promise { const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(value)); return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join("").slice(0, 16); @@ -114,6 +162,25 @@ async function responseCacheKey(url: string, headers: Headers): Promise return `v2:${authHash}:${accept}:${apiVersion}:${url}`; } +type VolatileSingleFlightScope = { requestKey: string; authorization: string }; + +function volatileSingleFlightScope(url: string, headers: Headers): VolatileSingleFlightScope { + const accept = encodeURIComponent(headers.get("accept") || ""); + const apiVersion = encodeURIComponent(headers.get("x-github-api-version") || ""); + return { requestKey: `volatile:${accept}:${apiVersion}:${url}`, authorization: headers.get("authorization") || "" }; +} + +function isVolatileSingleFlightEligibleGithubUrl(url: string, headers: Headers): boolean { + if (!url.startsWith(`${GITHUB_API_PREFIX}/`)) return false; + const accept = (headers.get("accept") ?? "").toLowerCase(); + if (accept.includes("raw") || accept.includes("text/plain")) return false; + const path = githubApiPath(url); + return ( + !/^\/repos\/[^/]+\/[^/]+\/contents(?:\/|$|[?#])/.test(path) && + !/^\/repos\/[^/]+\/[^/]+\/git\/(?:trees|blobs)\//.test(path) + ); +} + function requestHeaders(input: RequestInfo | URL, init: RequestInit | undefined): Headers { const headers = new Headers(typeof Request !== "undefined" && input instanceof Request ? input.headers : undefined); new Headers(init?.headers).forEach((value, key) => headers.set(key, value)); @@ -128,6 +195,22 @@ function requestUrl(input: RequestInfo | URL): string { return typeof Request !== "undefined" && input instanceof Request ? input.url : String(input); } +function requestSignal(input: RequestInfo | URL, init: GitHubTimeoutFetchInit | undefined): AbortSignal | undefined { + return init?.signal ?? (typeof Request !== "undefined" && input instanceof Request ? input.signal : undefined); +} + +function rateLimitAdmissionKey(init: GitHubTimeoutFetchInit | undefined): GitHubRateLimitAdmissionKey | null { + if (init?.githubRateLimitAdmission !== true) return null; + const key = init.githubRateLimitAdmissionKey?.trim(); + return key ? key : null; +} + +function requestInitForFetch(init: GitHubTimeoutFetchInit | undefined): RequestInit | undefined { + if (!init || (!("githubRateLimitAdmission" in init) && !("githubRateLimitAdmissionKey" in init))) return init; + const { githubRateLimitAdmission: _omitted, githubRateLimitAdmissionKey: _omittedKey, ...rest } = init; + return rest; +} + export function isGitHubResponseCacheReplay(response: Response): boolean { return response.headers.get(GITHUB_RESPONSE_CACHE_REPLAY_HEADER) !== null; } @@ -179,15 +262,29 @@ function responseFromCached(hit: CachedGitHubResponse, replayKind: "hit" | "coal }); } -async function fetchWithGitHubRetry(input: RequestInfo | URL, init?: RequestInit): Promise { +async function replayableResponse(response: Response): Promise { + return { + status: response.status, + body: await response.clone().text(), + contentType: response.headers.get("content-type") ?? "application/json", + ...(response.headers.get("link") ? { link: response.headers.get("link")! } : {}), + ...(response.headers.get("etag") ? { etag: response.headers.get("etag")! } : {}), + ...(response.headers.get("last-modified") ? { lastModified: response.headers.get("last-modified")! } : {}), + }; +} + +async function fetchWithGitHubRetry(input: RequestInfo | URL, init?: GitHubTimeoutFetchInit): Promise { let response: Response; + const fetchInit = requestInitForFetch(init); + const admissionKey = rateLimitAdmissionKey(init); for (let attempt = 0; ; attempt += 1) { - response = init?.signal - ? await fetch(input, init) + response = fetchInit?.signal + ? await fetch(input, fetchInit) : await fetch(input, { - ...(init ?? {}), + ...(fetchInit ?? {}), signal: AbortSignal.timeout(GITHUB_FETCH_TIMEOUT_MS), }); + if (admissionKey) observeGitHubRestRateLimit(requestUrl(input), response, admissionKey); // Retry a transient rate-limit (with backoff) instead of surfacing it; stop once exhausted or it's not a limit. if (attempt >= GITHUB_RATE_LIMIT_MAX_RETRIES || !(await isRateLimitedResponse(response))) break; await sleep(rateLimitRetryMs(response, attempt)); @@ -197,22 +294,16 @@ async function fetchWithGitHubRetry(input: RequestInfo | URL, init?: RequestInit async function fetchAndMaybeCacheGitHubGet( input: RequestInfo | URL, - init: RequestInit | undefined, + init: GitHubTimeoutFetchInit | undefined, url: string, cacheKey: string, cls: GitHubCacheClass, ): Promise<{ response: Response; cached: CachedGitHubResponse | null }> { const response = await fetchWithGitHubRetry(input, init); - if (response.status !== 200) return { response, cached: null }; + if (!isCacheableGithubResponseStatus(cls, response.status)) return { response, cached: null }; + if (await isRateLimitedResponse(response)) return { response, cached: null }; try { - const cached = { - status: 200, - body: await response.clone().text(), - contentType: response.headers.get("content-type") ?? "application/json", - ...(response.headers.get("link") ? { link: response.headers.get("link")! } : {}), - ...(response.headers.get("etag") ? { etag: response.headers.get("etag")! } : {}), - ...(response.headers.get("last-modified") ? { lastModified: response.headers.get("last-modified")! } : {}), - }; + const cached = await replayableResponse(response); await responseCache!.set(cacheKey, cached, githubResponseCacheTtlSeconds(cls)); recordGitHubCacheMetric("set", cls); return { response, cached }; @@ -225,15 +316,76 @@ async function fetchAndMaybeCacheGitHubGet( // Single-flight cacheable GETs inside one isolate: a webhook burst often asks for the same metadata // before Redis has been populated. Join those cold misses so GitHub sees one request, then replay the cached body. const inFlightCacheableGets = new Map>(); +// Mutable GitHub GETs are not persisted in Redis, but simultaneous identical reads in one burst can still share the +// leader's response. This dedupes review fan-out without replaying stale CI, PR, label, comment, or event data later. +const inFlightVolatileGets = new Map>>(); + +async function fetchWithVolatileSingleFlight( + input: RequestInfo | URL, + init: GitHubTimeoutFetchInit | undefined, + scope: VolatileSingleFlightScope, +): Promise { + const existing = inFlightVolatileGets.get(scope.requestKey)?.get(scope.authorization); + if (existing) { + recordGitHubCacheMetric("coalesced", "sensitive"); + const replay = await waitForVolatileReplay(existing, requestSignal(input, init)); + if (replay) return responseFromCached(replay, "coalesced"); + } + let resolveShared!: (value: CachedGitHubResponse | null) => void; + const shared = new Promise((resolve) => { + resolveShared = resolve; + }); + let bucket = inFlightVolatileGets.get(scope.requestKey); + if (!bucket) { + bucket = new Map(); + inFlightVolatileGets.set(scope.requestKey, bucket); + } + const sharedWithCleanup = shared.finally(() => { + const current = inFlightVolatileGets.get(scope.requestKey); + current?.delete(scope.authorization); + if (current?.size === 0) inFlightVolatileGets.delete(scope.requestKey); + }); + bucket.set(scope.authorization, sharedWithCleanup); + recordGitHubCacheMetric("bypassed", "sensitive"); + try { + const response = await fetchWithGitHubRetry(input, init); + try { + resolveShared(await replayableResponse(response)); + } catch { + resolveShared(null); + } + return response; + } catch (error) { + resolveShared(null); + throw error; + } +} + +function abortSignalError(signal: AbortSignal): Error { + return signal.reason instanceof Error ? signal.reason : new Error("The operation was aborted."); +} + +function waitForVolatileReplay(shared: Promise, signal: AbortSignal | undefined): Promise { + if (!signal) return shared; + if (signal.aborted) return Promise.reject(abortSignalError(signal)); + return new Promise((resolve, reject) => { + const onAbort = () => reject(abortSignalError(signal)); + signal.addEventListener("abort", onAbort, { once: true }); + shared.then(resolve, reject).finally(() => signal.removeEventListener("abort", onAbort)); + }); +} // A 12s hard cap on every GitHub request. Centralised here so the app token/installation raw fetches plus comment / // label / check-run / pr-action Octokit helpers all inherit the cache boundary, retry, and timeout behavior. -export async function timeoutFetch(input: RequestInfo | URL, init?: RequestInit): Promise { +export async function timeoutFetch(input: RequestInfo | URL, init?: GitHubTimeoutFetchInit): Promise { const method = requestMethod(input, init); const url = requestUrl(input); const headers = requestHeaders(input, init); const conditional = hasConditionalRequestHeader(headers); const cls = method === "GET" && !conditional ? githubCacheClassForUrl(url) : null; + if (method === "GET" && !conditional && cls === null && isVolatileSingleFlightEligibleGithubUrl(url, headers)) { + return fetchWithVolatileSingleFlight(input, init, volatileSingleFlightScope(url, headers)); + } const useCache = responseCache !== null && cls !== null; if (!useCache) { recordGitHubCacheMetric("bypassed", cacheBypassClass(method, url, headers)); @@ -276,6 +428,8 @@ export async function timeoutFetch(input: RequestInfo | URL, init?: RequestInit) export function clearGitHubResponseCacheForTest(): void { responseCache = null; inFlightCacheableGets.clear(); + inFlightVolatileGets.clear(); + latestRestRateLimitObservations.clear(); } const WRITE_METHODS = new Set(["POST", "PATCH", "PUT", "DELETE"]); @@ -330,8 +484,17 @@ export function forcedSelfhostMode(env: { SELFHOST_DEPLOYMENT_MODE?: string | un * the executor are not double-denied; surface callers (check-run / comment / label) pass the resolved repo mode. * A SELFHOST_DEPLOYMENT_MODE override beats the per-call mode so the whole instance can be forced non-actuating. */ -export function makeInstallationOctokit(env: Env, token: string, mode: AgentActionMode = "live"): Octokit { - const octokit = new Octokit({ auth: token, request: { fetch: timeoutFetch } }); +export function makeInstallationOctokit(env: Env, token: string, mode: AgentActionMode = "live", admissionKey?: GitHubRateLimitAdmissionKey | undefined): Octokit { + const octokit = new Octokit({ + auth: token, + request: { + fetch: (input: RequestInfo | URL, init?: RequestInit) => { + const fetchInit: GitHubTimeoutFetchInit = Object.assign({ githubRateLimitAdmission: admissionKey !== undefined }, init); + if (admissionKey) fetchInit.githubRateLimitAdmissionKey = admissionKey; + return timeoutFetch(input, fetchInit); + }, + }, + }); const effectiveMode = forcedSelfhostMode(env) ?? mode; if (effectiveMode !== "live") { octokit.hook.wrap("request", async (request, options) => { diff --git a/src/github/comments.ts b/src/github/comments.ts index 06a83fd0a6..2b519f1b74 100644 --- a/src/github/comments.ts +++ b/src/github/comments.ts @@ -1,5 +1,5 @@ import { withInstallationTokenRetry } from "./app"; -import { makeInstallationOctokit } from "./client"; +import { githubRateLimitAdmissionKeyForInstallation, makeInstallationOctokit } from "./client"; import type { AgentActionMode } from "../settings/agent-execution"; export const PR_PANEL_COMMENT_MARKER = ""; @@ -55,7 +55,7 @@ async function createOrUpdateIssueCommentWithMarker( return await withInstallationTokenRetry(env, installationId, async (token) => { // Non-live mode suppresses the comment create/update writes; the GET marker-search probe below still runs. - const octokit = makeInstallationOctokit(env, token, options.mode ?? "live"); + const octokit = makeInstallationOctokit(env, token, options.mode ?? "live", githubRateLimitAdmissionKeyForInstallation(installationId)); const botLogin = `${env.GITHUB_APP_SLUG}[bot]`; const markers = markerAliases(marker); const existing: IssueComment[] = []; diff --git a/src/github/labels.ts b/src/github/labels.ts index bad47a685c..89ee07737d 100644 --- a/src/github/labels.ts +++ b/src/github/labels.ts @@ -1,5 +1,5 @@ import { createInstallationToken } from "./app"; -import { makeInstallationOctokit } from "./client"; +import { githubRateLimitAdmissionKeyForInstallation, makeInstallationOctokit } from "./client"; import type { AgentActionMode } from "../settings/agent-execution"; type GitHubLabel = { @@ -19,7 +19,7 @@ export async function ensurePullRequestLabel( const token = await createInstallationToken(env, installationId); // Non-live mode suppresses the label create + apply writes; the GET dedup probe below still runs. - const octokit = makeInstallationOctokit(env, token, options.mode ?? "live"); + const octokit = makeInstallationOctokit(env, token, options.mode ?? "live", githubRateLimitAdmissionKeyForInstallation(installationId)); const existing = await octokit.request("GET /repos/{owner}/{repo}/issues/{issue_number}/labels", { owner, repo, @@ -64,7 +64,7 @@ export async function removePullRequestLabel(env: Env, installationId: number, r const [owner, repo] = repoFullName.split("/"); if (!owner || !repo) return; const token = await createInstallationToken(env, installationId); - const octokit = makeInstallationOctokit(env, token, mode); + const octokit = makeInstallationOctokit(env, token, mode, githubRateLimitAdmissionKeyForInstallation(installationId)); await octokit .request("DELETE /repos/{owner}/{repo}/issues/{issue_number}/labels/{name}", { owner, repo, issue_number: pullNumber, name: labelName }) .catch(() => undefined); diff --git a/src/github/pr-actions.ts b/src/github/pr-actions.ts index 1c604aeb4e..f465a85714 100644 --- a/src/github/pr-actions.ts +++ b/src/github/pr-actions.ts @@ -1,5 +1,5 @@ import { createInstallationToken } from "./app"; -import { makeInstallationOctokit } from "./client"; +import { githubRateLimitAdmissionKeyForInstallation, makeInstallationOctokit } from "./client"; import type { AgentActionMode } from "../settings/agent-execution"; import type { AutoMergeMethod } from "../types"; @@ -30,7 +30,7 @@ export async function createPullRequestReview( ): Promise<{ id: number }> { const { owner, repo } = splitRepo(repoFullName); const token = await createInstallationToken(env, installationId); - const octokit = makeInstallationOctokit(env, token); + const octokit = makeInstallationOctokit(env, token, "live", githubRateLimitAdmissionKeyForInstallation(installationId)); const response = await octokit.request("POST /repos/{owner}/{repo}/pulls/{pull_number}/reviews", { owner, repo, @@ -57,7 +57,7 @@ export async function createPullRequestReviewComments( ): Promise<{ id: number }> { const { owner, repo } = splitRepo(repoFullName); const token = await createInstallationToken(env, installationId); - const octokit = makeInstallationOctokit(env, token, mode); + const octokit = makeInstallationOctokit(env, token, mode, githubRateLimitAdmissionKeyForInstallation(installationId)); const response = await octokit.request("POST /repos/{owner}/{repo}/pulls/{pull_number}/reviews", { owner, repo, @@ -80,7 +80,7 @@ export async function mergePullRequest( ): Promise<{ merged: boolean; sha: string | null }> { const { owner, repo } = splitRepo(repoFullName); const token = await createInstallationToken(env, installationId); - const octokit = makeInstallationOctokit(env, token); + const octokit = makeInstallationOctokit(env, token, "live", githubRateLimitAdmissionKeyForInstallation(installationId)); const response = await octokit.request("PUT /repos/{owner}/{repo}/pulls/{pull_number}/merge", { owner, repo, @@ -106,7 +106,7 @@ export async function updatePullRequestBranch( ): Promise { const { owner, repo } = splitRepo(repoFullName); const token = await createInstallationToken(env, installationId); - const octokit = makeInstallationOctokit(env, token); + const octokit = makeInstallationOctokit(env, token, "live", githubRateLimitAdmissionKeyForInstallation(installationId)); await octokit.request("PUT /repos/{owner}/{repo}/pulls/{pull_number}/update-branch", { owner, repo, @@ -119,7 +119,7 @@ export async function updatePullRequestBranch( export async function createIssueComment(env: Env, installationId: number, repoFullName: string, issueNumber: number, body: string): Promise<{ id: number }> { const { owner, repo } = splitRepo(repoFullName); const token = await createInstallationToken(env, installationId); - const octokit = makeInstallationOctokit(env, token); + const octokit = makeInstallationOctokit(env, token, "live", githubRateLimitAdmissionKeyForInstallation(installationId)); const response = await octokit.request("POST /repos/{owner}/{repo}/issues/{issue_number}/comments", { owner, repo, @@ -133,7 +133,7 @@ export async function createIssueComment(env: Env, installationId: number, repoF export async function closePullRequest(env: Env, installationId: number, repoFullName: string, pullNumber: number): Promise<{ state: string }> { const { owner, repo } = splitRepo(repoFullName); const token = await createInstallationToken(env, installationId); - const octokit = makeInstallationOctokit(env, token); + const octokit = makeInstallationOctokit(env, token, "live", githubRateLimitAdmissionKeyForInstallation(installationId)); const response = await octokit.request("PATCH /repos/{owner}/{repo}/pulls/{pull_number}", { owner, repo, @@ -156,7 +156,7 @@ export async function getLastCloserLogin(env: Env, installationId: number, repoF try { const { owner, repo } = splitRepo(repoFullName); const token = await createInstallationToken(env, installationId); - const octokit = makeInstallationOctokit(env, token); + const octokit = makeInstallationOctokit(env, token, "live", githubRateLimitAdmissionKeyForInstallation(installationId)); const requestPage = (page: number) => octokit.request("GET /repos/{owner}/{repo}/issues/{issue_number}/events", { owner, repo, issue_number: issueNumber, per_page: ISSUE_EVENTS_PAGE_SIZE, page }); const firstResponse = await requestPage(1); diff --git a/src/github/public.ts b/src/github/public.ts index 372ac77eeb..26e80beb79 100644 --- a/src/github/public.ts +++ b/src/github/public.ts @@ -1,3 +1,5 @@ +import { timeoutFetch } from "./client"; + export type PublicContributorProfile = { login: string; name?: string | null | undefined; @@ -75,15 +77,15 @@ export async function fetchPublicContributorProfile(login: string, env?: Pick, repoFullName: string, nowMs: number): Promise { const [owner, repo] = repoFullName.split("/") as [string, string]; - const response = await fetch(`https://api.github.com/repos/${encodeURIComponent(owner)}/${encodeURIComponent(repo)}`, { + const response = await timeoutFetch(`https://api.github.com/repos/${encodeURIComponent(owner)}/${encodeURIComponent(repo)}`, { headers: { accept: "application/vnd.github+json", "user-agent": "gittensory/0.1", diff --git a/src/github/rate-limit.ts b/src/github/rate-limit.ts index 3f37b343d6..b97fc14d21 100644 --- a/src/github/rate-limit.ts +++ b/src/github/rate-limit.ts @@ -4,8 +4,10 @@ import { listLatestGitHubRateLimitObservations } from "../db/repositories"; // from draining the budget real webhook traffic needs, maintenance yields while there is still headroom: // - backfill yields at LOW_REST_RATE_LIMIT_REMAINING; // - the re-gate sweep + its per-PR jobs yield EARLIER, at MAINTENANCE_RESERVED_HEADROOM, reserving the budget -// between the two floors for webhooks. Webhooks never pre-yield — they run, and the queue retries them after -// the reset if the bucket is exhausted (the surviving event-loss path). (#audit-rate-headroom) +// between the two floors for webhooks. +// Self-host queues also use the latest persisted observation for admission control, so a known-exhausted bucket +// delays webhook jobs before they start and avoids burning the first live delivery just to discover the limit. +// (#audit-rate-headroom) export const LOW_REST_RATE_LIMIT_REMAINING = 75; export const MAINTENANCE_RESERVED_HEADROOM = 150; diff --git a/src/orb/oauth.ts b/src/orb/oauth.ts index 6846e13bfa..1fba42ad53 100644 --- a/src/orb/oauth.ts +++ b/src/orb/oauth.ts @@ -13,6 +13,7 @@ // (read back at token-exchange, never from a request). No request input is echoed into the markup (no injection // surface). import type { Context } from "hono"; +import { timeoutFetch } from "../github/client"; import { isOrbBrokerEnabled, issueOrbEnrollment } from "./broker"; type GitHubUser = { login: string; id?: number }; @@ -32,7 +33,7 @@ export async function exchangeOrbOAuthCode(env: Env, code: string, fetchImpl: ty } /** Identify the authenticated maintainer (GET /user with their token). Null on any non-OK / loginless response. */ -export async function fetchOrbOAuthUser(token: string, fetchImpl: typeof fetch = fetch): Promise { +export async function fetchOrbOAuthUser(token: string, fetchImpl: typeof fetch = timeoutFetch): Promise { const res = await fetchImpl("https://api.github.com/user", { headers: { authorization: `Bearer ${token}`, accept: "application/vnd.github+json", "user-agent": "gittensory/0.1" }, }); @@ -51,7 +52,7 @@ export async function verifyInstallationAdmin( accountLogin: string | null, accountType: string | null, accountId: number | null, - fetchImpl: typeof fetch = fetch, + fetchImpl: typeof fetch = timeoutFetch, ): Promise { if (!accountLogin || accountId === null) return false; if (accountType !== "Organization") { diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 24492d633d..85e9dfe3ef 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -134,7 +134,7 @@ import { ensurePullRequestLabel, removePullRequestLabel, } from "../github/labels"; -import { resolveRepoActionMode } from "../github/client"; +import { githubRateLimitAdmissionKeyForInstallation, resolveRepoActionMode } from "../github/client"; import { ALL_TYPE_LABELS, resolvePrTypeLabel } from "../settings/pr-type-label"; import { fetchPublicContributorProfile } from "../github/public"; import { refreshRegistry } from "../registry/sync"; @@ -5341,6 +5341,7 @@ async function maybePublishPrPublicSurface( previewFromChecks: true, }, visualFiles, + githubRateLimitAdmissionKeyForInstallation(installationId), ); beforeAfter = capture.routes; // Visual self-poll: the FIRST capture returns a "loading" placeholder for the AFTER shot when the diff --git a/src/review/grounding-wire.ts b/src/review/grounding-wire.ts index f8d5daf8dd..67e7271de1 100644 --- a/src/review/grounding-wire.ts +++ b/src/review/grounding-wire.ts @@ -12,6 +12,7 @@ // fail-safe: any missing CI data / fetch error degrades to "no grounding" and the review proceeds on the diff. import { createInstallationToken } from "../github/app"; +import { githubRateLimitAdmissionKeyForInstallation, timeoutFetch, type GitHubRateLimitAdmissionKey } from "../github/client"; import type { CheckSummaryRecord, PullRequestFileRecord } from "../types"; import { repoParts } from "../utils/json"; import { isConvergenceRepoAllowed } from "./cutover-gate"; @@ -114,8 +115,10 @@ function toGroundingFiles(files: PullRequestFileRecord[]): PullRequestFile[] { export async function makeGithubFileFetcher(env: Env, repoFullName: string, installationId: number | null | undefined): Promise { // Resolve the token once (best-effort): installation token > public token > none. let token: string | undefined; + let admissionKey: GitHubRateLimitAdmissionKey | undefined; if (installationId) { token = await createInstallationToken(env, installationId).catch(() => undefined); + admissionKey = token !== undefined ? githubRateLimitAdmissionKeyForInstallation(installationId) : undefined; } token = token ?? env.GITHUB_PUBLIC_TOKEN; const { owner, name } = repoParts(repoFullName); @@ -129,8 +132,10 @@ export async function makeGithubFileFetcher(env: Env, repoFullName: string, inst const controller = new AbortController(); const timeout = setTimeout(() => controller.abort(), 10_000); try { - const response = await fetch(url, { + const response = await timeoutFetch(url, { signal: controller.signal, + githubRateLimitAdmission: admissionKey !== undefined, + ...(admissionKey ? { githubRateLimitAdmissionKey: admissionKey } : {}), headers: { // raw media type returns the file body directly (no base64 envelope to decode). accept: "application/vnd.github.raw+json", diff --git a/src/review/rag-index.ts b/src/review/rag-index.ts index b5155c2ef6..13aafb33d7 100644 --- a/src/review/rag-index.ts +++ b/src/review/rag-index.ts @@ -23,6 +23,7 @@ // GitHub call, and does no adapter use — the deploy is byte-identical to today. import { createInstallationToken } from "../github/app"; +import { githubRateLimitAdmissionKeyForInstallation, timeoutFetch, type GitHubRateLimitAdmissionKey } from "../github/client"; import { repoParts } from "../utils/json"; import { createReviewAdapters } from "./adapters"; import { @@ -51,10 +52,12 @@ const GITHUB_FETCH_TIMEOUT_MS = 10_000; /** Resolve the read token once for a repo: installation token (private-repo read) → public token → none. * Best-effort — a token failure degrades to the next fallback, never throws. (Mirrors makeGithubFileFetcher.) */ -async function resolveReadToken(env: Env, installationId: number | null | undefined): Promise { - let token: string | undefined; - if (installationId) token = await createInstallationToken(env, installationId).catch(() => undefined); - return token ?? env.GITHUB_PUBLIC_TOKEN; +async function resolveReadToken(env: Env, installationId: number | null | undefined): Promise<{ token: string | undefined; admissionKey?: GitHubRateLimitAdmissionKey | undefined }> { + if (installationId) { + const token = await createInstallationToken(env, installationId).catch(() => undefined); + if (token) return { token, admissionKey: githubRateLimitAdmissionKeyForInstallation(installationId) }; + } + return { token: env.GITHUB_PUBLIC_TOKEN }; } /** Shared GitHub headers for the read calls (raw media type returns file bodies directly). */ @@ -73,11 +76,16 @@ function ghHeaders(token: string | undefined, accept: string): Record { +async function fetchRepoTree(env: Env, repoFullName: string, ref: string, token: string | undefined, admissionKey: GitHubRateLimitAdmissionKey | undefined): Promise { try { const { owner, name } = repoParts(repoFullName); const url = `https://api.github.com/repos/${encodeURIComponent(owner)}/${encodeURIComponent(name)}/git/trees/${encodeURIComponent(ref)}?recursive=1`; - const response = await fetch(url, { headers: ghHeaders(token, "application/vnd.github+json"), signal: AbortSignal.timeout(GITHUB_FETCH_TIMEOUT_MS) }); + const response = await timeoutFetch(url, { + headers: ghHeaders(token, "application/vnd.github+json"), + signal: AbortSignal.timeout(GITHUB_FETCH_TIMEOUT_MS), + githubRateLimitAdmission: admissionKey !== undefined, + ...(admissionKey ? { githubRateLimitAdmissionKey: admissionKey } : {}), + }); if (!response.ok) return null; const body = (await response.json()) as { tree?: Array<{ path?: string; type?: string; size?: number }> } | null; const entries: TreeEntry[] = []; @@ -133,6 +141,7 @@ async function fetchFileText( path: string, ref: string, token: string | undefined, + admissionKey: GitHubRateLimitAdmissionKey | undefined, maxBytes = MAX_FILE_BYTES, ): Promise { try { @@ -141,7 +150,12 @@ async function fetchFileText( .split("/") .map(encodeURIComponent) .join("/")}?ref=${encodeURIComponent(ref)}`; - const response = await fetch(url, { headers: ghHeaders(token, "application/vnd.github.raw+json"), signal: AbortSignal.timeout(GITHUB_FETCH_TIMEOUT_MS) }); + const response = await timeoutFetch(url, { + headers: ghHeaders(token, "application/vnd.github.raw+json"), + signal: AbortSignal.timeout(GITHUB_FETCH_TIMEOUT_MS), + githubRateLimitAdmission: admissionKey !== undefined, + ...(admissionKey ? { githubRateLimitAdmissionKey: admissionKey } : {}), + }); if (!response.ok) return null; return await readTextCapped(response, maxBytes); } catch { @@ -237,13 +251,13 @@ export async function indexRepo( const repoFullName = repo.fullName; const [, repoName] = splitRepo(repoFullName); const namespace = ragNamespace(project, repoName); - const token = await resolveReadToken(env, repo.installationId); + const { token, admissionKey } = await resolveReadToken(env, repo.installationId); const ref = indexRef(repo.defaultBranch); // 1. Fetch the tree, filter to indexable code/docs, and prune retained chunks for files that disappeared // or moved to a non-indexable path. If the tree fetch fails (null), skip pruning to avoid deleting good // chunks during a transient GitHub/API failure. - const rawTree = await fetchRepoTree(env, repoFullName, ref, token); + const rawTree = await fetchRepoTree(env, repoFullName, ref, token, admissionKey); if (rawTree === null) return empty; const tree = rawTree .filter((entry) => isIndexablePath(entry.path, entry.size)) @@ -261,7 +275,7 @@ export async function indexRepo( capped = true; break; } - const text = await fetchFileText(env, repoFullName, entry.path, ref, token); + const text = await fetchFileText(env, repoFullName, entry.path, ref, token, admissionKey); if (text === null) continue; const chunks = chunkFile(entry.path, text, namespace); if (chunks.length === 0) continue; @@ -313,7 +327,7 @@ export async function reindexChangedPaths( // 2. Re-index the ones that are still indexable code/docs at the default branch. const indexable = unique.filter((path) => isIndexablePath(path)); if (indexable.length === 0) return { indexed: 0, files: 0, capped: false }; - const token = await resolveReadToken(env, repo.installationId); + const { token, admissionKey } = await resolveReadToken(env, repo.installationId); const ref = indexRef(repo.defaultBranch); let stored = await countRepoChunks(infra.storage, project, repoName); let upserted = 0; @@ -324,7 +338,7 @@ export async function reindexChangedPaths( capped = true; break; } - const text = await fetchFileText(env, repoFullName, path, ref, token); + const text = await fetchFileText(env, repoFullName, path, ref, token, admissionKey); if (text === null) continue; // file deleted at head, oversized, or unreadable — already removed above, leave it gone const chunks = chunkFile(path, text, namespace); if (chunks.length === 0) continue; diff --git a/src/review/visual/capture.ts b/src/review/visual/capture.ts index bc90c40b2e..92485f70c4 100644 --- a/src/review/visual/capture.ts +++ b/src/review/visual/capture.ts @@ -11,6 +11,7 @@ // preview session, and explicit-route override are intentionally dropped here — gittensory's UI uses the // default TanStack route convention; those hooks can return if a per-repo visual config is added. import { sha256Hex } from "../../utils/crypto"; +import type { GitHubRateLimitAdmissionKey } from "../../github/client"; import { findPreviewUrlFromChecks, findPreviewUrlFromPrComments, @@ -138,7 +139,7 @@ async function capturePage( * collapsible). Fully fail-safe — a missing preview / failed render degrades to placeholders or dashes; this * NEVER throws (the caller also wraps it in try/catch so a capture failure can't sink a review). */ -export async function buildCapture(env: Env, token: string, target: CaptureTarget, visualFiles: string[]): Promise { +export async function buildCapture(env: Env, token: string, target: CaptureTarget, visualFiles: string[], rateLimitAdmissionKey?: GitHubRateLimitAdmissionKey | undefined): Promise { const repo = parseRepo(target.repoFullName); const apiVersion = "2022-11-28"; // before = production (PUBLIC_SITE_ORIGIN, e.g. https://gittensory.aethereal.dev). @@ -153,19 +154,19 @@ export async function buildCapture(env: Env, token: string, target: CaptureTarge let previewPending = false; if (!previewBase && !previewFailed) { try { - const status = await getLatestDeploymentStatus({ token, repo, sha: target.headSha, ref: target.headRef, apiVersion }); + const status = await getLatestDeploymentStatus({ token, repo, sha: target.headSha, ref: target.headRef, apiVersion, rateLimitAdmissionKey }); previewBase = status.url ?? ""; previewFailed = status.failed; } catch { previewBase = ""; } if (!previewBase && !previewFailed && target.previewFromChecks && target.headSha) { - previewBase = (await findPreviewUrlFromChecks({ token, repo, sha: target.headSha, apiVersion })) ?? ""; + previewBase = (await findPreviewUrlFromChecks({ token, repo, sha: target.headSha, apiVersion, rateLimitAdmissionKey })) ?? ""; if (!previewBase && target.prNumber) { - previewBase = (await findPreviewUrlFromPrComments({ token, repo, prNumber: target.prNumber, apiVersion })) ?? ""; + previewBase = (await findPreviewUrlFromPrComments({ token, repo, prNumber: target.prNumber, apiVersion, rateLimitAdmissionKey })) ?? ""; } if (!previewBase && target.headSha) { - const buildState = await getPreviewBuildState({ token, repo, sha: target.headSha, apiVersion }); + const buildState = await getPreviewBuildState({ token, repo, sha: target.headSha, apiVersion, rateLimitAdmissionKey }); if (buildState === "failed") previewFailed = true; else if (buildState === "building" || buildState === "succeeded") previewPending = true; } diff --git a/src/review/visual/preview-url.ts b/src/review/visual/preview-url.ts index 8b5f80a6e2..f49bcc520b 100644 --- a/src/review/visual/preview-url.ts +++ b/src/review/visual/preview-url.ts @@ -15,6 +15,8 @@ // (resolved via createInstallationToken). Every helper degrades to null/absent on failure — preview // discovery must NEVER sink a review. +import { timeoutFetch, type GitHubRateLimitAdmissionKey } from "../../github/client"; + const DEFAULT_GITHUB_TIMEOUT_MS = 20_000; export type GitHubRepo = { owner: string; repo: string }; @@ -38,13 +40,21 @@ class PreviewGitHubError extends Error { /** Minimal fetch→JSON helper (mirrors reviewbot's core/github.ts githubJson). Throws PreviewGitHubError on a * non-2xx so callers can distinguish a 404 ("no deployments") from a transient outage. */ -async function githubJson(url: string, init: { token?: string | undefined; apiVersion?: string | undefined } = {}): Promise { +async function githubJson( + url: string, + init: { token?: string | undefined; apiVersion?: string | undefined; rateLimitAdmissionKey?: GitHubRateLimitAdmissionKey | undefined } = {}, +): Promise { const headers = new Headers(); headers.set("accept", "application/vnd.github+json"); headers.set("user-agent", "gittensory/0.1"); headers.set("x-github-api-version", init.apiVersion || "2022-11-28"); if (init.token) headers.set("authorization", `Bearer ${init.token}`); - const response = await fetch(url, { headers, signal: AbortSignal.timeout(DEFAULT_GITHUB_TIMEOUT_MS) }); + const response = await timeoutFetch(url, { + headers, + signal: AbortSignal.timeout(DEFAULT_GITHUB_TIMEOUT_MS), + githubRateLimitAdmission: init.rateLimitAdmissionKey !== undefined, + ...(init.rateLimitAdmissionKey ? { githubRateLimitAdmissionKey: init.rateLimitAdmissionKey } : {}), + }); const text = await response.text(); let payload: unknown = null; if (text) { @@ -76,6 +86,7 @@ export async function getLatestDeploymentStatus(params: { sha?: string | undefined; ref?: string | undefined; apiVersion?: string | undefined; + rateLimitAdmissionKey?: GitHubRateLimitAdmissionKey | undefined; }): Promise { const base = `https://api.github.com/repos/${params.repo.owner}/${params.repo.repo}`; const selector = params.sha @@ -89,6 +100,7 @@ export async function getLatestDeploymentStatus(params: { deployments = await githubJson>(`${base}/deployments?${selector}&per_page=10`, { token: params.token, apiVersion: params.apiVersion, + rateLimitAdmissionKey: params.rateLimitAdmissionKey, }); } catch (error) { // 404 → the ref genuinely has no deployments. Any other failure (403 missing scope, rate limit, 5xx) is @@ -103,6 +115,7 @@ export async function getLatestDeploymentStatus(params: { githubJson>(`${base}/deployments/${id}/statuses?per_page=10`, { token: params.token, apiVersion: params.apiVersion, + rateLimitAdmissionKey: params.rateLimitAdmissionKey, }).catch((error) => { console.log(JSON.stringify({ ev: "deployment_status_error", deployment: id, message: String(error).slice(0, 200) })); return [] as Array<{ state?: string; environment_url?: string }>; @@ -158,9 +171,10 @@ export async function findPreviewUrlFromChecks(params: { repo: GitHubRepo; sha: string; apiVersion?: string | undefined; + rateLimitAdmissionKey?: GitHubRateLimitAdmissionKey | undefined; }): Promise { const base = `https://api.github.com/repos/${params.repo.owner}/${params.repo.repo}`; - const opts = { token: params.token, apiVersion: params.apiVersion }; + const opts = { token: params.token, apiVersion: params.apiVersion, rateLimitAdmissionKey: params.rateLimitAdmissionKey }; try { const combined = await githubJson<{ statuses?: Array<{ state?: string; target_url?: string }> }>( `${base}/commits/${encodeURIComponent(params.sha)}/status`, @@ -198,12 +212,13 @@ export async function findPreviewUrlFromPrComments(params: { repo: GitHubRepo; prNumber: number; apiVersion?: string | undefined; + rateLimitAdmissionKey?: GitHubRateLimitAdmissionKey | undefined; }): Promise { const base = `https://api.github.com/repos/${params.repo.owner}/${params.repo.repo}`; try { const comments = await githubJson>( `${base}/issues/${params.prNumber}/comments?per_page=100`, - { token: params.token, apiVersion: params.apiVersion }, + { token: params.token, apiVersion: params.apiVersion, rateLimitAdmissionKey: params.rateLimitAdmissionKey }, ).catch(() => null); if (!Array.isArray(comments)) return null; // Newest first (the bot edits one comment in place). @@ -229,12 +244,13 @@ export async function getPreviewBuildState(params: { repo: GitHubRepo; sha: string; apiVersion?: string | undefined; + rateLimitAdmissionKey?: GitHubRateLimitAdmissionKey | undefined; }): Promise<"building" | "succeeded" | "failed" | "absent"> { const base = `https://api.github.com/repos/${params.repo.owner}/${params.repo.repo}`; try { const checks = await githubJson<{ check_runs?: Array<{ name?: string; status?: string; conclusion?: string }> }>( `${base}/commits/${encodeURIComponent(params.sha)}/check-runs?per_page=100`, - { token: params.token, apiVersion: params.apiVersion }, + { token: params.token, apiVersion: params.apiVersion, rateLimitAdmissionKey: params.rateLimitAdmissionKey }, ).catch(() => null); const build = (checks?.check_runs ?? []).find((r) => /workers builds|cloudflare/i.test(r.name ?? "")); if (!build) return "absent"; diff --git a/src/scoring/model.ts b/src/scoring/model.ts index 8786eb4815..c25005eba8 100644 --- a/src/scoring/model.ts +++ b/src/scoring/model.ts @@ -2,6 +2,7 @@ import { getLatestScoringModelSnapshot, persistScoringModelSnapshot, } from "../db/repositories"; +import { timeoutFetch } from "../github/client"; import { getLatestRegistrySnapshot } from "../registry/sync"; import { syncUnmodeledScoringConstantDrift } from "../upstream/unmodeled-scoring-drift"; import type { JsonValue, ScoringModelSnapshotRecord } from "../types"; @@ -80,7 +81,7 @@ function upstreamRawUrl(config: { repo: string; ref: string }, path: string): st // missing administration token must never block the constants refresh itself. async function fetchUpstreamRefSha(upstream: { repo: string; ref: string }, token: string | undefined): Promise { try { - const response = await fetch(`https://api.github.com/repos/${upstream.repo}/commits/${encodeURIComponent(upstream.ref)}`, { headers: githubHeaders(token, "application/vnd.github+json") }); + const response = await timeoutFetch(`https://api.github.com/repos/${upstream.repo}/commits/${encodeURIComponent(upstream.ref)}`, { headers: githubHeaders(token, "application/vnd.github+json") }); if (!response.ok) return null; const data = (await response.json()) as { sha?: string }; return typeof data.sha === "string" && data.sha.length > 0 ? data.sha : null; @@ -313,7 +314,7 @@ function hasDensityConstants(constants: Record): boolean { async function fetchText(url: string, token?: string): Promise<{ ok: true; value: string } | { ok: false; error: string }> { try { - const response = await fetch(url, { headers: githubHeaders(token, "text/plain") }); + const response = await timeoutFetch(url, { headers: githubHeaders(token, "text/plain") }); if (!response.ok) return { ok: false, error: `${response.status} ${response.statusText}` }; return { ok: true, value: await response.text() }; } catch (error) { @@ -323,7 +324,7 @@ async function fetchText(url: string, token?: string): Promise<{ ok: true; value async function fetchJson(url: string, token?: string): Promise<{ ok: true; value: Record } | { ok: false; error: string }> { try { - const response = await fetch(url, { headers: githubHeaders(token, "application/json") }); + const response = await timeoutFetch(url, { headers: githubHeaders(token, "application/json") }); if (!response.ok) return { ok: false, error: `${response.status} ${response.statusText}` }; return { ok: true, value: (await response.json()) as Record }; } catch (error) { diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index 1aa8ee4491..a467a1ca20 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -11,7 +11,9 @@ import { consumingRetryDelayMs, deterministicJitterMs, FOREGROUND_QUEUE_PRIORITY_FLOOR, - githubBackgroundRateLimitDelayMs, + githubRateLimitAdmissionDelayMs, + githubRateLimitAdmissionKeyForJob, + githubRateLimitAdmissionRemainingFloor, githubRateLimitRetryDelayMs, isGitHubBudgetBackgroundJob, jobCoalesceKey, @@ -342,25 +344,24 @@ export function createPgQueue( return true; } const jobTraceParent = message.type === "github-webhook" ? message.traceParent : undefined; - const backgroundRateLimitDelay = isGitHubBudgetBackgroundJob(message) - ? await backgroundRateLimitDelayMs() - : null; - if (backgroundRateLimitDelay !== null) { + const rateLimitAdmission = await rateLimitAdmissionDelayMs(message); + if (rateLimitAdmission !== null) { const now = Date.now(); const retryAfter = now + rateLimitRetryDelayWithJitter( - backgroundRateLimitDelay, + rateLimitAdmission.delayMs, `${job.job_key ?? ""}:${job.id}:${job.payload}`, ); + const lastError = `github rate-limit ${rateLimitAdmission.kind} admission`; const update = await pool.query( `UPDATE ${TABLE} SET status='pending', run_after=GREATEST(run_after, $1), last_error=COALESCE(last_error, $2) WHERE id=$3`, - [retryAfter, "github rate-limit background admission", job.id], + [retryAfter, lastError, job.id], ); if (update.rowCount) { await recordQueueMetric("gittensory_jobs_rate_limit_deferred_total"); console.warn( JSON.stringify({ level: "warn", - event: "selfhost_queue_background_admission_deferred", + event: `selfhost_queue_${rateLimitAdmission.kind}_admission_deferred`, jobType: message.type, retry_after_ms: Math.max(0, retryAfter - now), }), @@ -605,15 +606,29 @@ export function createPgQueue( return changed; } - async function backgroundRateLimitDelayMs(): Promise { + async function rateLimitAdmissionDelayMs(message: JobMessage): Promise<{ kind: "background" | "webhook"; delayMs: number } | null> { + const kind = + message.type === "github-webhook" + ? "webhook" + : isGitHubBudgetBackgroundJob(message) + ? "background" + : null; + if (kind === null) return null; + const admissionKey = githubRateLimitAdmissionKeyForJob(message); + const remainingFloor = githubRateLimitAdmissionRemainingFloor(kind); const res = await pool.query( - `SELECT remaining, reset_at FROM github_rate_limit_observations - WHERE resource='rest' AND remaining IS NOT NULL - ORDER BY observed_at DESC - LIMIT 1`, + `SELECT remaining, reset_at, observed_at FROM github_rate_limit_observations + WHERE resource='rest' AND remaining IS NOT NULL AND ( + ($1::text IS NOT NULL AND admission_key=$1) + OR admission_key IS NULL + ) + ORDER BY CASE WHEN remaining <= $2 THEN 1 ELSE 0 END DESC, observed_at DESC + LIMIT 16`, + [admissionKey, remainingFloor], ); - const row = res.rows[0] as { remaining?: number | string | null; reset_at?: string | null } | undefined; - return githubBackgroundRateLimitDelayMs(row); + const rows = res.rows as Array<{ remaining?: number | string | null; reset_at?: string | null; observed_at?: string | null }>; + const delayMs = githubRateLimitAdmissionDelayMs(kind, admissionKey, rows); + return delayMs === null ? null : { kind, delayMs }; } async function mergeRescheduledJobIntoPending( diff --git a/src/selfhost/queue-common.ts b/src/selfhost/queue-common.ts index fc5aa8d668..c28e48b677 100644 --- a/src/selfhost/queue-common.ts +++ b/src/selfhost/queue-common.ts @@ -1,5 +1,13 @@ import { retryableJobDelayMs } from "../queue/retryable"; -import { MAINTENANCE_RESERVED_HEADROOM } from "../github/rate-limit"; +import { + LOW_REST_RATE_LIMIT_REMAINING, + MAINTENANCE_RESERVED_HEADROOM, +} from "../github/rate-limit"; +import { + githubRateLimitAdmissionKeyForInstallation, + latestGitHubRestRateLimitObservation, + type GitHubRateLimitAdmissionKey, +} from "../github/client"; import { githubWebhookCoalesceKey } from "../github/webhook-coalesce"; import type { GitHubWebhookPayload, JobMessage } from "../types"; import { extractPayloadType } from "./audit"; @@ -87,11 +95,12 @@ export function isGitHubBudgetBackgroundJob(message: JobMessage): boolean { return GITHUB_BUDGET_BACKGROUND_TYPES.has(message.type); } -export function githubBackgroundRateLimitDelayMs( +function githubObservedRateLimitDelayMs( observation: | { remaining?: unknown; reset_at?: unknown; resetAt?: unknown } | null | undefined, + floor: number, nowMs = Date.now(), ): number | null { const rawRemaining = observation?.remaining; @@ -108,12 +117,110 @@ export function githubBackgroundRateLimitDelayMs( ? observation.resetAt : null; if (remaining === null || !resetAt) return null; - if (remaining > MAINTENANCE_RESERVED_HEADROOM) return null; + if (remaining > floor) return null; const ms = Date.parse(resetAt) - nowMs; if (!Number.isFinite(ms) || ms <= 0) return null; return Math.max(30_000, Math.min(900_000, (Math.ceil(ms / 1000) + 15) * 1000)); } +function observationMs( + observation: + | { observed_at?: unknown; observedAt?: unknown; observedAtMs?: unknown } + | null + | undefined, +): number | null { + if (typeof observation?.observedAtMs === "number" && Number.isFinite(observation.observedAtMs)) { + return observation.observedAtMs; + } + const raw = + typeof observation?.observed_at === "string" + ? observation.observed_at + : typeof observation?.observedAt === "string" + ? observation.observedAt + : null; + if (!raw) return null; + const parsed = Date.parse(raw); + return Number.isFinite(parsed) ? parsed : null; +} + +type AdmissionObservation = { + remaining?: unknown; + reset_at?: unknown; + resetAt?: unknown; + observed_at?: unknown; + observedAt?: unknown; + observedAtMs?: unknown; +}; + +function newestRateLimitObservation( + admissionKey: GitHubRateLimitAdmissionKey | null | undefined, + persisted: AdmissionObservation | null | undefined, +): + | AdmissionObservation + | null + | undefined { + const local = admissionKey ? latestGitHubRestRateLimitObservation(admissionKey) : null; + if (!local) return persisted; + if (!persisted) return local; + const persistedMs = observationMs(persisted); + return persistedMs !== null && persistedMs > local.observedAtMs ? persisted : local; +} + +export function githubRateLimitAdmissionKeyForJob(message: JobMessage): GitHubRateLimitAdmissionKey | null { + const installationId = + message.type === "github-webhook" + ? message.payload?.installation?.id + : "installationId" in message + ? message.installationId + : null; + return typeof installationId === "number" && Number.isFinite(installationId) + ? githubRateLimitAdmissionKeyForInstallation(installationId) + : null; +} + +export function githubRateLimitAdmissionDelayMs( + kind: "background" | "webhook", + admissionKey: GitHubRateLimitAdmissionKey | null | undefined, + persisted: AdmissionObservation | readonly AdmissionObservation[] | null | undefined, + nowMs = Date.now(), +): number | null { + const candidates = Array.isArray(persisted) ? persisted : [persisted]; + let maxDelay: number | null = null; + for (const candidate of candidates.length > 0 ? candidates : [undefined]) { + const observation = newestRateLimitObservation(admissionKey, candidate); + const delay = + kind === "webhook" + ? githubWebhookRateLimitDelayMs(observation, nowMs) + : githubBackgroundRateLimitDelayMs(observation, nowMs); + if (delay !== null) maxDelay = Math.max(maxDelay ?? 0, delay); + } + return maxDelay; +} + +export function githubBackgroundRateLimitDelayMs( + observation: + | { remaining?: unknown; reset_at?: unknown; resetAt?: unknown } + | null + | undefined, + nowMs = Date.now(), +): number | null { + return githubObservedRateLimitDelayMs(observation, MAINTENANCE_RESERVED_HEADROOM, nowMs); +} + +export function githubWebhookRateLimitDelayMs( + observation: + | { remaining?: unknown; reset_at?: unknown; resetAt?: unknown } + | null + | undefined, + nowMs = Date.now(), +): number | null { + return githubObservedRateLimitDelayMs(observation, LOW_REST_RATE_LIMIT_REMAINING, nowMs); +} + +export function githubRateLimitAdmissionRemainingFloor(kind: "background" | "webhook"): number { + return kind === "webhook" ? LOW_REST_RATE_LIMIT_REMAINING : MAINTENANCE_RESERVED_HEADROOM; +} + function githubWebhookPriority(payload: string): number { try { const message = JSON.parse(payload) as { diff --git a/src/selfhost/setup-wizard.ts b/src/selfhost/setup-wizard.ts index 4e67961b89..da37092a9b 100644 --- a/src/selfhost/setup-wizard.ts +++ b/src/selfhost/setup-wizard.ts @@ -4,6 +4,7 @@ // code for the App's credentials and writes them to a file the operator loads (then restarts). The routes are // disabled once an App is configured (server.ts gates on GITHUB_APP_ID), so this can't rebind a live install. import { createHmac, timingSafeEqual } from "node:crypto"; +import { timeoutFetch } from "../github/client"; export const SETUP_TOKEN_FORM_MAX_BYTES = 4096; @@ -124,7 +125,7 @@ ${error}
} /** Exchange the temporary manifest code for the App's credentials (id, slug, webhook secret, private key). */ -export async function exchangeManifestCode(code: string, fetchImpl: typeof fetch = fetch): Promise { +export async function exchangeManifestCode(code: string, fetchImpl: typeof fetch = timeoutFetch): Promise { const res = await fetchImpl(`https://api.github.com/app-manifests/${encodeURIComponent(code)}/conversions`, { method: "POST", headers: { accept: "application/vnd.github+json", "user-agent": "gittensory-selfhost" }, diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index 0a20dcb699..008c10b818 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -12,7 +12,9 @@ import { consumingRetryDelayMs, deterministicJitterMs, FOREGROUND_QUEUE_PRIORITY_FLOOR, - githubBackgroundRateLimitDelayMs, + githubRateLimitAdmissionDelayMs, + githubRateLimitAdmissionKeyForJob, + githubRateLimitAdmissionRemainingFloor, githubRateLimitRetryDelayMs, isGitHubBudgetBackgroundJob, jobCoalesceKey, @@ -285,25 +287,24 @@ export function createSqliteQueue( return true; } const jobTraceParent = message.type === "github-webhook" ? message.traceParent : undefined; - const backgroundRateLimitDelay = isGitHubBudgetBackgroundJob(message) - ? backgroundRateLimitDelayMs(driver) - : null; - if (backgroundRateLimitDelay !== null) { + const rateLimitAdmission = rateLimitAdmissionDelayMs(driver, message); + if (rateLimitAdmission !== null) { const now = Date.now(); const retryAfter = now + rateLimitRetryDelayWithJitter( - backgroundRateLimitDelay, + rateLimitAdmission.delayMs, `${job.job_key ?? ""}:${job.id}:${job.payload}`, ); + const lastError = `github rate-limit ${rateLimitAdmission.kind} admission`; const { changes } = driver.query( `UPDATE ${TABLE} SET status='pending', run_after=max(run_after, ?), last_error=coalesce(last_error, ?) WHERE id=?`, - [retryAfter, "github rate-limit background admission", job.id], + [retryAfter, lastError, job.id], ); if (changes) { recordQueueMetric(driver, "gittensory_jobs_rate_limit_deferred_total"); console.warn( JSON.stringify({ level: "warn", - event: "selfhost_queue_background_admission_deferred", + event: `selfhost_queue_${rateLimitAdmission.kind}_admission_deferred`, jobType: message.type, retry_after_ms: Math.max(0, retryAfter - now), }), @@ -602,16 +603,32 @@ function deferPendingJobsForRateLimit( return changed; } -function backgroundRateLimitDelayMs(driver: SqliteDriver): number | null { +function rateLimitAdmissionDelayMs( + driver: SqliteDriver, + message: JobMessage, +): { kind: "background" | "webhook"; delayMs: number } | null { + const kind = + message.type === "github-webhook" + ? "webhook" + : isGitHubBudgetBackgroundJob(message) + ? "background" + : null; + if (kind === null) return null; try { - const row = driver.query( - `SELECT remaining, reset_at FROM github_rate_limit_observations - WHERE resource='rest' AND remaining IS NOT NULL - ORDER BY observed_at DESC - LIMIT 1`, - [], - ).rows[0] as { remaining?: number | null; reset_at?: string | null } | undefined; - return githubBackgroundRateLimitDelayMs(row); + const admissionKey = githubRateLimitAdmissionKeyForJob(message); + const remainingFloor = githubRateLimitAdmissionRemainingFloor(kind); + const rows = driver.query( + `SELECT remaining, reset_at, observed_at FROM github_rate_limit_observations + WHERE resource='rest' AND remaining IS NOT NULL AND ( + (? IS NOT NULL AND admission_key=?) + OR admission_key IS NULL + ) + ORDER BY CASE WHEN remaining <= ? THEN 1 ELSE 0 END DESC, observed_at DESC + LIMIT 16`, + [admissionKey, admissionKey, remainingFloor], + ).rows as Array<{ remaining?: number | null; reset_at?: string | null; observed_at?: string | null }>; + const delayMs = githubRateLimitAdmissionDelayMs(kind, admissionKey, rows); + return delayMs === null ? null : { kind, delayMs }; } catch { return null; } diff --git a/src/services/contributor-issue-draft.ts b/src/services/contributor-issue-draft.ts index bd3866f920..425b79289c 100644 --- a/src/services/contributor-issue-draft.ts +++ b/src/services/contributor-issue-draft.ts @@ -17,6 +17,7 @@ import { import type { IssueRecord, RepositoryRecord, RepositorySettings } from "../types"; import { isGlobalAgentPause } from "../settings/agent-execution"; import { isMaintainerAssociation } from "../github/commands"; +import { timeoutFetch } from "../github/client"; import { sha256Hex } from "../utils/crypto"; import { jsonString, nowIso, repoParts } from "../utils/json"; import { @@ -550,7 +551,7 @@ async function createGitHubContributorIssue(env: Env, repoFullName: string, draf if (!token) return null; const { owner, name } = repoParts(repoFullName); if (!owner || !name) return null; - const response = await fetch(`https://api.github.com/repos/${owner}/${name}/issues`, { + const response = await timeoutFetch(`https://api.github.com/repos/${owner}/${name}/issues`, { method: "POST", headers: githubHeaders(token), body: jsonString({ diff --git a/src/services/draft.ts b/src/services/draft.ts index b1da91f781..1136ffdd48 100644 --- a/src/services/draft.ts +++ b/src/services/draft.ts @@ -10,6 +10,7 @@ // reviewbot is collapsed into module constants + env vars. The flow is gated by GITTENSORY_REVIEW_DRAFT; when // the flag is off the router never mounts these handlers (callers see 404). import { decryptDraftToken, encryptDraftToken, newDraftId, randomDraftToken, sha256Hex, timingSafeEqualHex } from "../utils/crypto"; +import { timeoutFetch } from "../github/client"; const REDACT_KEYS = /(email|phone|address|contact|zip|postcode|name)/i; const TOKEN_TTL_SECONDS = 900; @@ -352,7 +353,7 @@ async function githubUserJson(url: string, init: RequestInit & { token: strin headers.set("x-github-api-version", GITHUB_API_VERSION); /* v8 ignore next -- token-absent arm is unreachable: every caller passes a decrypted user token; the { token: "" } default only guards the type. */ if (init.token) headers.set("authorization", `Bearer ${init.token}`); - const response = await fetch(url, { ...init, headers, signal: AbortSignal.timeout(GITHUB_FETCH_TIMEOUT_MS) }); + const response = await timeoutFetch(url, { ...init, headers, signal: AbortSignal.timeout(GITHUB_FETCH_TIMEOUT_MS) }); const body = await response.text(); let payload: unknown = null; if (body) { diff --git a/src/signals/focus-manifest-loader.ts b/src/signals/focus-manifest-loader.ts index 191860e395..0ae40c66ec 100644 --- a/src/signals/focus-manifest-loader.ts +++ b/src/signals/focus-manifest-loader.ts @@ -5,6 +5,7 @@ import { featuresConfigToJson, gateConfigToJson, MAX_FOCUS_MANIFEST_BYTES, parse import { GITTENSORY_REPO_FOCUS_MANIFEST_YAML, resolveGittensorySelfRepoFullName } from "../config/gittensory-repo-focus-manifest"; export const REPO_FOCUS_MANIFEST_SIGNAL = "repo-focus-manifest"; +export const REPO_PUBLIC_FOCUS_MANIFEST_SIGNAL = "repo-public-focus-manifest"; export const REPO_FOCUS_MANIFEST_MAX_AGE_MS = 6 * 60 * 60 * 1000; export const REPO_FOCUS_MANIFEST_MAX_CONCURRENT_LOADS = 4; @@ -142,7 +143,9 @@ async function loadRepoFocusManifestWithCachePolicy( } catch { manifest = parseFocusManifest(null); } - if (!cachePolicy.publicOnly) { + if (cachePolicy.publicOnly) { + await persistRepoFocusManifest(env, repoFullName, manifest, REPO_PUBLIC_FOCUS_MANIFEST_SIGNAL); + } else { // Persist even an ABSENT manifest (negative cache): effective settings are resolved from // `.gittensory.yml` on every webhook, so a repo without one must not re-fetch the raw file each time. // The TTL still refreshes it, so a newly-added manifest is picked up on the next window. @@ -218,23 +221,46 @@ export async function upsertRepoFocusManifest(env: Env, repoFullName: string, ra } async function readCachedManifest(env: Env, repoFullName: string, maxAgeMs: number, options: { publicOnly?: boolean } = {}): Promise { - const [latest] = await listSignalSnapshots(env, REPO_FOCUS_MANIFEST_SIGNAL, repoFullName); + if (options.publicOnly) { + return ( + (await readCachedManifestSnapshot(env, REPO_PUBLIC_FOCUS_MANIFEST_SIGNAL, repoFullName, maxAgeMs, options)) ?? + // Back-compat: public previews may reuse old repo-file snapshots written before the dedicated public cache + // existed, but must still ignore maintainer/API-backed records. + (await readCachedManifestSnapshot(env, REPO_FOCUS_MANIFEST_SIGNAL, repoFullName, maxAgeMs, { ...options, requireRepoFileSource: true })) + ); + } + return readCachedManifestSnapshot(env, REPO_FOCUS_MANIFEST_SIGNAL, repoFullName, maxAgeMs, options); +} + +async function readCachedManifestSnapshot( + env: Env, + signalType: string, + repoFullName: string, + maxAgeMs: number, + options: { publicOnly?: boolean; requireRepoFileSource?: boolean } = {}, +): Promise { + const [latest] = await listSignalSnapshots(env, signalType, repoFullName); if (!latest) return null; const manifest = parseFocusManifest(latest.payload); const explicitSource = latest.payload !== null && typeof latest.payload === "object" && !Array.isArray(latest.payload) ? (latest.payload as Record).source : undefined; - if (options.publicOnly && explicitSource !== "repo_file") return null; + if (options.requireRepoFileSource) { + if (explicitSource !== "repo_file") return null; + } + if (options.publicOnly) { + if (explicitSource === "api_record") return null; + } if (explicitSource === "api_record") return manifest; if (snapshotAgeMs(latest.generatedAt) > maxAgeMs) return null; return manifest; } -async function persistRepoFocusManifest(env: Env, repoFullName: string, manifest: FocusManifest): Promise { +async function persistRepoFocusManifest(env: Env, repoFullName: string, manifest: FocusManifest, signalType = REPO_FOCUS_MANIFEST_SIGNAL): Promise { await persistSignalSnapshot(env, { id: crypto.randomUUID(), - signalType: REPO_FOCUS_MANIFEST_SIGNAL, + signalType, targetKey: repoFullName, repoFullName, payload: manifestToJson(manifest), diff --git a/src/types.ts b/src/types.ts index 392fc7de30..73e419d53d 100644 --- a/src/types.ts +++ b/src/types.ts @@ -798,6 +798,7 @@ export type PullRequestDetailSyncStateRecord = { export type GitHubRateLimitObservationRecord = { id?: string | undefined; repoFullName?: string | null | undefined; + admissionKey?: string | null | undefined; resource: "rest" | "graphql"; path: string; statusCode: number; diff --git a/src/upstream/ruleset.ts b/src/upstream/ruleset.ts index d315f5e128..18c668f05b 100644 --- a/src/upstream/ruleset.ts +++ b/src/upstream/ruleset.ts @@ -10,6 +10,7 @@ import { updateUpstreamDriftReportIssue, upsertUpstreamDriftReport, } from "../db/repositories"; +import { timeoutFetch } from "../github/client"; import { isGlobalAgentPause } from "../settings/agent-execution"; import { normalizeRegistryPayload } from "../registry/normalize"; import { DEFAULT_GITTENSOR_UPSTREAM_REF, DEFAULT_GITTENSOR_UPSTREAM_REPO, detectActiveModel, findUnmodeledConstantKeys, parsePythonNumberConstants } from "../scoring/model"; @@ -414,7 +415,7 @@ async function latestSourcesByKey(env: Env): Promise { const url = `https://api.github.com/repos/${config.repo}/commits/${encodeURIComponent(config.ref)}`; try { - const response = await fetch(url, { headers: githubHeaders(env.GITHUB_PUBLIC_TOKEN, "application/vnd.github+json") }); + const response = await timeoutFetch(url, { headers: githubHeaders(env.GITHUB_PUBLIC_TOKEN, "application/vnd.github+json") }); if (!response.ok) return null; const payload = (await response.json()) as { sha?: string }; return payload.sha ?? null; @@ -434,7 +435,7 @@ async function fetchTrackedSource( const apiUrl = `https://api.github.com/repos/${config.repo}/contents/${source.path}?ref=${encodeURIComponent(config.ref)}`; const warnings: string[] = []; try { - const response = await fetch(apiUrl, { + const response = await timeoutFetch(apiUrl, { headers: { ...githubHeaders(env.GITHUB_PUBLIC_TOKEN, "application/vnd.github+json"), ...(previous?.etag ? { "if-none-match": previous.etag } : {}), @@ -1029,7 +1030,7 @@ async function findGitHubIssueForFingerprint(repo: string, token: string, finger try { for (let page = 1; ; page += 1) { const url = `https://api.github.com/repos/${owner}/${name}/issues?state=open&labels=signals&per_page=100&page=${page}`; - const response = await fetch(url, { headers: githubHeaders(token, "application/vnd.github+json") }); + const response = await timeoutFetch(url, { headers: githubHeaders(token, "application/vnd.github+json") }); if (!response.ok) return null; const issues = (await response.json()) as Array<{ number?: number; html_url?: string; body?: string | null }>; const match = issues.find((issue) => issue.body?.includes(`gittensory-upstream-drift:${fingerprint}`)); @@ -1044,7 +1045,7 @@ async function findGitHubIssueForFingerprint(repo: string, token: string, finger async function createGitHubDriftIssue(repo: string, token: string, report: UpstreamDriftReportRecord, assignees: string[]): Promise<{ number: number; url: string } | null> { const [owner, name] = repo.split("/"); if (!owner || !name) return null; - const response = await fetch(`https://api.github.com/repos/${owner}/${name}/issues`, { + const response = await timeoutFetch(`https://api.github.com/repos/${owner}/${name}/issues`, { method: "POST", headers: githubHeaders(token, "application/vnd.github+json"), body: jsonString(githubDriftIssuePayload(report, assignees)), @@ -1057,7 +1058,7 @@ async function createGitHubDriftIssue(repo: string, token: string, report: Upstr async function updateGitHubDriftIssue(repo: string, token: string, issueNumber: number, report: UpstreamDriftReportRecord, assignees: string[]): Promise<{ number: number; url: string } | null> { const [owner, name] = repo.split("/"); if (!owner || !name || !Number.isInteger(issueNumber) || issueNumber <= 0) return null; - const response = await fetch(`https://api.github.com/repos/${owner}/${name}/issues/${issueNumber}`, { + const response = await timeoutFetch(`https://api.github.com/repos/${owner}/${name}/issues/${issueNumber}`, { method: "PATCH", headers: githubHeaders(token, "application/vnd.github+json"), body: jsonString(githubDriftIssuePayload(report, assignees)), @@ -1074,7 +1075,7 @@ async function validateRecordedGitHubIssue(repo: string, token: string, report: if (!owner || !name || !parsedUrl || parsedUrl.number !== report.issueNumber) return null; if (parsedUrl.owner.toLowerCase() !== owner.toLowerCase() || parsedUrl.name.toLowerCase() !== name.toLowerCase()) return null; try { - const response = await fetch(`https://api.github.com/repos/${owner}/${name}/issues/${report.issueNumber}`, { headers: githubHeaders(token, "application/vnd.github+json") }); + const response = await timeoutFetch(`https://api.github.com/repos/${owner}/${name}/issues/${report.issueNumber}`, { headers: githubHeaders(token, "application/vnd.github+json") }); if (!response.ok) return null; const issue = (await response.json()) as { number?: number; html_url?: string; state?: string; body?: string | null; labels?: Array }; if (issue.number !== report.issueNumber || !issue.html_url || issue.state !== "open") return null; diff --git a/test/unit/focus-manifest-loader.test.ts b/test/unit/focus-manifest-loader.test.ts index 5658ee2461..9bc3805851 100644 --- a/test/unit/focus-manifest-loader.test.ts +++ b/test/unit/focus-manifest-loader.test.ts @@ -10,6 +10,7 @@ import { upsertRepoFocusManifest, REPO_FOCUS_MANIFEST_MAX_AGE_MS, REPO_FOCUS_MANIFEST_MAX_CONCURRENT_LOADS, + REPO_PUBLIC_FOCUS_MANIFEST_SIGNAL, } from "../../src/signals/focus-manifest-loader"; import { MAX_FOCUS_MANIFEST_BYTES, parseFocusManifestContent } from "../../src/signals/focus-manifest"; @@ -154,6 +155,110 @@ describe("focus-manifest loader", () => { expect(privateManifest.gate.linkedIssue).toBe("block"); }); + it("caches public-only repo-file manifests without touching private/API-backed records", async () => { + const env = createTestEnv(); + let fetches = 0; + const first = await loadPublicRepoFocusManifest(env, "owner/public-cache", { + fetcher: async () => { + fetches += 1; + return JSON.stringify({ wantedPaths: ["public/"] }); + }, + }); + const second = await loadPublicRepoFocusManifest(env, "owner/public-cache", { + fetcher: async () => { + throw new Error("should use the public-only cache"); + }, + }); + + expect(fetches).toBe(1); + expect(first.source).toBe("repo_file"); + expect(second.wantedPaths).toEqual(["public/"]); + }); + + it("negative-caches absent public-only manifests in the public cache stream", async () => { + const env = createTestEnv(); + await loadPublicRepoFocusManifest(env, "owner/no-public-cache", { fetcher: async () => null }); + const { listSignalSnapshots } = await import("../../src/db/repositories"); + const snapshots = await listSignalSnapshots(env, REPO_PUBLIC_FOCUS_MANIFEST_SIGNAL, "owner/no-public-cache"); + expect(snapshots).toHaveLength(1); + + let fetches = 0; + const cached = await loadPublicRepoFocusManifest(env, "owner/no-public-cache", { + fetcher: async () => { + fetches += 1; + return JSON.stringify({ wantedPaths: ["unexpected/"] }); + }, + }); + expect(fetches).toBe(0); + expect(cached.present).toBe(false); + expect(cached.source).toBe("none"); + }); + + it("ignores unknown-source legacy snapshots on public-only manifest loads", async () => { + const env = createTestEnv(); + const { persistSignalSnapshot } = await import("../../src/db/repositories"); + const { REPO_FOCUS_MANIFEST_SIGNAL } = await import("../../src/signals/focus-manifest-loader"); + await persistSignalSnapshot(env, { + id: crypto.randomUUID(), + signalType: REPO_FOCUS_MANIFEST_SIGNAL, + targetKey: "owner/legacy-unknown", + repoFullName: "owner/legacy-unknown", + payload: { wantedPaths: ["unknown-source/"] }, + generatedAt: new Date().toISOString(), + }); + + const manifest = await loadPublicRepoFocusManifest(env, "owner/legacy-unknown", { + fetcher: async () => JSON.stringify({ wantedPaths: ["repo-file/"] }), + }); + + expect(manifest.source).toBe("repo_file"); + expect(manifest.wantedPaths).toEqual(["repo-file/"]); + }); + + it("ignores API-backed snapshots in the public cache stream", async () => { + const env = createTestEnv(); + const { persistSignalSnapshot } = await import("../../src/db/repositories"); + await persistSignalSnapshot(env, { + id: crypto.randomUUID(), + signalType: REPO_PUBLIC_FOCUS_MANIFEST_SIGNAL, + targetKey: "owner/public-api-cache", + repoFullName: "owner/public-api-cache", + payload: { source: "api_record", wantedPaths: ["private/"], gate: { linkedIssue: "block", readinessMinScore: 99 } }, + generatedAt: new Date().toISOString(), + }); + + const manifest = await loadPublicRepoFocusManifest(env, "owner/public-api-cache", { + fetcher: async () => JSON.stringify({ wantedPaths: ["repo-file/"], gate: { linkedIssue: "advisory" } }), + }); + + expect(manifest.source).toBe("repo_file"); + expect(manifest.wantedPaths).toEqual(["repo-file/"]); + expect(manifest.gate.linkedIssue).toBe("advisory"); + }); + + it("accepts explicit repo-file legacy snapshots on public-only manifest loads", async () => { + const env = createTestEnv(); + const { persistSignalSnapshot } = await import("../../src/db/repositories"); + const { REPO_FOCUS_MANIFEST_SIGNAL } = await import("../../src/signals/focus-manifest-loader"); + await persistSignalSnapshot(env, { + id: crypto.randomUUID(), + signalType: REPO_FOCUS_MANIFEST_SIGNAL, + targetKey: "owner/legacy-repo-file", + repoFullName: "owner/legacy-repo-file", + payload: { source: "repo_file", wantedPaths: ["legacy-repo-file/"] }, + generatedAt: new Date().toISOString(), + }); + + const manifest = await loadPublicRepoFocusManifest(env, "owner/legacy-repo-file", { + fetcher: async () => { + throw new Error("explicit repo-file legacy snapshot should be reused"); + }, + }); + + expect(manifest.source).toBe("repo_file"); + expect(manifest.wantedPaths).toEqual(["legacy-repo-file/"]); + }); + it("bulk-loads manifests for many repos with a concurrency cap", async () => { const env = createTestEnv(); let active = 0; diff --git a/test/unit/github-client.test.ts b/test/unit/github-client.test.ts index 6a757e7106..42b38fb49e 100644 --- a/test/unit/github-client.test.ts +++ b/test/unit/github-client.test.ts @@ -2,10 +2,12 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { clearGitHubResponseCacheForTest, forcedSelfhostMode, + githubRateLimitAdmissionKeyForInstallation, GITHUB_RESPONSE_CACHE_REPLAY_HEADER, githubResponseCacheTtlSeconds, isCacheableGithubUrl, isRateLimitedResponse, + latestGitHubRestRateLimitObservation, makeInstallationOctokit, resolveRepoActionMode, setGitHubResponseCache, @@ -39,6 +41,7 @@ afterEach(() => { resetMetrics(); vi.unstubAllEnvs(); vi.unstubAllGlobals(); + vi.useRealTimers(); }); describe("makeInstallationOctokit", () => { @@ -169,6 +172,95 @@ describe("timeoutFetch", () => { expect(injected).toBeInstanceOf(AbortSignal); }); + it("records only live GitHub core REST rate-limit observations", async () => { + const now = Date.parse("2026-06-24T12:00:00.000Z"); + const key = githubRateLimitAdmissionKeyForInstallation(123); + const otherKey = githubRateLimitAdmissionKeyForInstallation(456); + vi.useFakeTimers(); + vi.setSystemTime(now); + let headers = new Headers({ + "x-ratelimit-resource": "core", + "x-ratelimit-remaining": "42", + "x-ratelimit-reset": String(Math.floor(Date.parse("2026-06-24T12:10:00.000Z") / 1000)), + }); + vi.stubGlobal("fetch", async () => new Response("ok", { headers })); + + await timeoutFetch("https://api.github.com/repos/o/r/issues", { githubRateLimitAdmission: true }); + expect(latestGitHubRestRateLimitObservation(key)).toBeNull(); + + await timeoutFetch("https://api.github.com/repos/o/r/issues", { githubRateLimitAdmission: true, githubRateLimitAdmissionKey: key }); + expect(latestGitHubRestRateLimitObservation(key)).toEqual({ + remaining: 42, + resetAt: "2026-06-24T12:10:00.000Z", + observedAtMs: now, + }); + expect(latestGitHubRestRateLimitObservation(otherKey)).toBeNull(); + + headers = new Headers({ + "x-ratelimit-resource": "search", + "x-ratelimit-remaining": "0", + "x-ratelimit-reset": String(Math.floor(Date.parse("2026-06-24T12:30:00.000Z") / 1000)), + }); + await timeoutFetch("https://api.github.com/search/issues?q=repo:o/r", { githubRateLimitAdmission: true, githubRateLimitAdmissionKey: key }); + await timeoutFetch("https://example.test/health"); + headers = new Headers({ + "x-ratelimit-resource": "core", + "x-ratelimit-remaining": "0", + "x-ratelimit-reset": String(Math.floor(Date.parse("2026-06-24T12:40:00.000Z") / 1000)), + }); + await timeoutFetch("https://api.github.com/repos/o/r/pulls"); + headers = new Headers(); + await timeoutFetch("https://api.github.com/repos/o/r/comments", { githubRateLimitAdmission: true, githubRateLimitAdmissionKey: key }); + headers = new Headers({ + "x-ratelimit-resource": "core", + "x-ratelimit-remaining": "not-a-number", + "x-ratelimit-reset": String(Math.floor(Date.parse("2026-06-24T12:40:00.000Z") / 1000)), + }); + await timeoutFetch("https://api.github.com/repos/o/r/pulls", { githubRateLimitAdmission: true, githubRateLimitAdmissionKey: key }); + headers = new Headers({ + "x-ratelimit-resource": "core", + "x-ratelimit-remaining": "0", + "x-ratelimit-reset": String(Math.floor(Date.parse("2026-06-24T12:50:00.000Z") / 1000)), + }); + await timeoutFetch("https://example.test/health", { githubRateLimitAdmission: true, githubRateLimitAdmissionKey: key }); + + expect(latestGitHubRestRateLimitObservation(key)).toEqual({ + remaining: 42, + resetAt: "2026-06-24T12:10:00.000Z", + observedAtMs: now, + }); + }); + + it("records keyed REST admission telemetry from installation Octokit reads", async () => { + const now = Date.parse("2026-06-24T12:00:00.000Z"); + const key = githubRateLimitAdmissionKeyForInstallation(789); + vi.useFakeTimers(); + vi.setSystemTime(now); + vi.stubGlobal( + "fetch", + async () => + Response.json( + [{ id: 1 }], + { + headers: { + "x-ratelimit-resource": "core", + "x-ratelimit-remaining": "12", + "x-ratelimit-reset": String(Math.floor(Date.parse("2026-06-24T12:10:00.000Z") / 1000)), + }, + }, + ), + ); + const octokit = makeInstallationOctokit(createTestEnv(), "tok", "live", key); + + await octokit.request("GET /repos/{owner}/{repo}/issues/{issue_number}/comments", { owner: "o", repo: "r", issue_number: 1 }); + + expect(latestGitHubRestRateLimitObservation(key)).toEqual({ + remaining: 12, + resetAt: "2026-06-24T12:10:00.000Z", + observedAtMs: now, + }); + }); + it("serves stable installation Octokit metadata GETs from the shared GitHub response cache", async () => { const store = installMemoryResponseCache(); let getFetches = 0; @@ -290,6 +382,99 @@ describe("timeoutFetch", () => { expect(getFetches).toBe(1); }); + it("negative-caches stable branch-protection permission and missing-resource responses", async () => { + installMemoryResponseCache(); + let getFetches = 0; + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + getFetches += 1; + const branch = String(input).includes("/branches/dev/"); + return Response.json( + { message: branch ? "Branch not found" : "Resource not accessible by integration" }, + { status: branch ? 404 : 403 }, + ); + }); + + const forbiddenUrl = "https://api.github.com/repos/o/r/branches/main/protection/required_status_checks"; + const missingUrl = "https://api.github.com/repos/o/r/branches/dev/protection/required_status_checks"; + const forbidden = await timeoutFetch(forbiddenUrl); + const forbiddenReplay = await timeoutFetch(forbiddenUrl); + const missing = await timeoutFetch(missingUrl); + const missingReplay = await timeoutFetch(missingUrl); + + expect(forbidden.status).toBe(403); + expect(forbiddenReplay.status).toBe(403); + expect(forbiddenReplay.headers.get(GITHUB_RESPONSE_CACHE_REPLAY_HEADER)).toBe("hit"); + expect(await forbiddenReplay.json()).toEqual({ message: "Resource not accessible by integration" }); + expect(missing.status).toBe(404); + expect(missingReplay.status).toBe(404); + expect(missingReplay.headers.get(GITHUB_RESPONSE_CACHE_REPLAY_HEADER)).toBe("hit"); + expect(await missingReplay.json()).toEqual({ message: "Branch not found" }); + expect(getFetches).toBe(2); + }); + + it("does not cache GitHub rate-limit 403 responses as branch-protection metadata", async () => { + installMemoryResponseCache(); + let getFetches = 0; + vi.stubGlobal("fetch", async () => { + getFetches += 1; + return Response.json( + { message: getFetches === 1 ? "API rate limit exceeded" : "Resource not accessible by integration" }, + getFetches === 1 + ? { status: 403, headers: { "x-ratelimit-remaining": "0" } } + : { status: 403 }, + ); + }); + + const url = "https://api.github.com/repos/o/r/branches/main/protection/required_status_checks"; + const first = await timeoutFetch(url); + const second = await timeoutFetch(url); + const replay = await timeoutFetch(url); + + expect(first.status).toBe(403); + expect(second.status).toBe(403); + expect(replay.status).toBe(403); + expect(replay.headers.get(GITHUB_RESPONSE_CACHE_REPLAY_HEADER)).toBe("hit"); + expect(await replay.json()).toEqual({ message: "Resource not accessible by integration" }); + expect(getFetches).toBe(2); + }); + + it("does not cache a branch-protection 403 while every inline retry is still rate-limited", async () => { + const set = vi.fn(async () => undefined); + setGitHubResponseCache({ + get: async () => null, + set, + }); + let getFetches = 0; + vi.stubGlobal("fetch", async () => { + getFetches += 1; + return Response.json( + { message: "API rate limit exceeded" }, + { status: 403, headers: { "retry-after": "0", "x-ratelimit-remaining": "0" } }, + ); + }); + + const response = await timeoutFetch("https://api.github.com/repos/o/r/branches/main/protection/required_status_checks"); + + expect(response.status).toBe(403); + expect(await response.json()).toEqual({ message: "API rate limit exceeded" }); + expect(getFetches).toBe(4); + expect(set).not.toHaveBeenCalled(); + }); + + it("does not negative-cache stable metadata denials outside branch protection", async () => { + installMemoryResponseCache(); + let getFetches = 0; + vi.stubGlobal("fetch", async () => { + getFetches += 1; + return Response.json({ message: `denied-${getFetches}` }, { status: 403 }); + }); + + const url = "https://api.github.com/repos/o/r"; + expect(await (await timeoutFetch(url)).json()).toEqual({ message: "denied-1" }); + expect(await (await timeoutFetch(url)).json()).toEqual({ message: "denied-2" }); + expect(getFetches).toBe(2); + }); + it("defaults cached replays to application/json when GitHub omits content-type", async () => { const store = installMemoryResponseCache(); vi.stubGlobal("fetch", async () => new Response(new TextEncoder().encode('{"ok":true}'), { status: 200 })); @@ -426,6 +611,214 @@ describe("timeoutFetch", () => { expect(await renderMetrics()).toContain(`gittensory_github_response_cache_total{class="sensitive",result="bypassed"} ${mutableCases.length * 2}`); }); + it("single-flights concurrent mutable GitHub GETs without persisting them in Redis", async () => { + const cacheGet = vi.fn(async () => ({ + status: 200, + body: JSON.stringify({ state: "stale" }), + contentType: "application/json", + })); + const cacheSet = vi.fn(async () => undefined); + setGitHubResponseCache({ get: cacheGet, set: cacheSet }); + let releaseFetch!: () => void; + const fetchGate = new Promise((resolve) => { + releaseFetch = resolve; + }); + let markFetchStarted!: () => void; + const fetchStarted = new Promise((resolve) => { + markFetchStarted = resolve; + }); + let getFetches = 0; + vi.stubGlobal("fetch", async () => { + getFetches += 1; + if (getFetches === 1) { + markFetchStarted(); + await fetchGate; + } + return Response.json([{ state: `live-${getFetches}` }]); + }); + + const url = "https://api.github.com/repos/o/r/pulls/7/reviews?per_page=100&page=1"; + const init = { headers: { authorization: "Bearer volatile-token" } }; + const first = timeoutFetch(url, init); + await fetchStarted; + await Promise.resolve(); + const second = timeoutFetch(url, init); + releaseFetch(); + const [firstResponse, secondResponse] = await Promise.all([first, second]); + const later = await timeoutFetch(url, init); + + expect(await firstResponse.json()).toEqual([{ state: "live-1" }]); + expect(await secondResponse.json()).toEqual([{ state: "live-1" }]); + expect(secondResponse.headers.get(GITHUB_RESPONSE_CACHE_REPLAY_HEADER)).toBe("coalesced"); + expect(await later.json()).toEqual([{ state: "live-2" }]); + expect(getFetches).toBe(2); + expect(cacheGet).not.toHaveBeenCalled(); + expect(cacheSet).not.toHaveBeenCalled(); + const metrics = await renderMetrics(); + expect(metrics).toContain('gittensory_github_response_cache_total{class="sensitive",result="bypassed"} 2'); + expect(metrics).toContain('gittensory_github_response_cache_total{class="sensitive",result="coalesced"} 1'); + }); + + it("single-flights concurrent mutable GitHub GETs even when Redis is disabled", async () => { + let releaseFetch!: () => void; + const fetchGate = new Promise((resolve) => { + releaseFetch = resolve; + }); + let markFetchStarted!: () => void; + const fetchStarted = new Promise((resolve) => { + markFetchStarted = resolve; + }); + let getFetches = 0; + vi.stubGlobal("fetch", async () => { + getFetches += 1; + if (getFetches === 1) { + markFetchStarted(); + await fetchGate; + } + return Response.json([{ state: `live-${getFetches}` }]); + }); + + const url = "https://api.github.com/repos/o/r/issues/7/events?per_page=100&page=1"; + const first = timeoutFetch(url); + await fetchStarted; + const second = timeoutFetch(url); + releaseFetch(); + const [firstResponse, secondResponse] = await Promise.all([first, second]); + const later = await timeoutFetch(url); + + expect(await firstResponse.json()).toEqual([{ state: "live-1" }]); + expect(await secondResponse.json()).toEqual([{ state: "live-1" }]); + expect(secondResponse.headers.get(GITHUB_RESPONSE_CACHE_REPLAY_HEADER)).toBe("coalesced"); + expect(await later.json()).toEqual([{ state: "live-2" }]); + expect(getFetches).toBe(2); + }); + + it("falls back to a fresh mutable GET when the volatile leader cannot be replayed", async () => { + class UncloneableResponse extends Response { + override clone(): Response { + throw new Error("cannot replay"); + } + } + let releaseFetch!: () => void; + const fetchGate = new Promise((resolve) => { + releaseFetch = resolve; + }); + let markFetchStarted!: () => void; + const fetchStarted = new Promise((resolve) => { + markFetchStarted = resolve; + }); + let getFetches = 0; + vi.stubGlobal("fetch", async () => { + getFetches += 1; + if (getFetches === 1) { + markFetchStarted(); + await fetchGate; + return new UncloneableResponse(JSON.stringify({ state: "leader-only" }), { headers: { "content-type": "application/json" } }); + } + return Response.json({ state: "fresh-follower" }); + }); + + const url = "https://api.github.com/repos/o/r/pulls/7/reviews?per_page=100&page=1"; + const first = timeoutFetch(url); + await fetchStarted; + const second = timeoutFetch(url); + releaseFetch(); + + await expect(first.then((response) => response.json())).resolves.toEqual({ state: "leader-only" }); + await expect(second.then((response) => response.json())).resolves.toEqual({ state: "fresh-follower" }); + expect(getFetches).toBe(2); + }); + + it("does not volatile-single-flight raw contents reads that callers stream-cap", async () => { + setGitHubResponseCache({ + get: async () => null, + set: async () => undefined, + }); + let releaseFetch!: () => void; + const fetchGate = new Promise((resolve) => { + releaseFetch = resolve; + }); + let getFetches = 0; + vi.stubGlobal("fetch", async () => { + const fetchId = (getFetches += 1); + await fetchGate; + return new Response(`body-${fetchId}`, { headers: { "content-type": "text/plain" } }); + }); + + const url = "https://api.github.com/repos/o/r/contents/src/big.ts?ref=main"; + const init = { headers: { accept: "application/vnd.github.raw+json" } }; + const first = timeoutFetch(url, init); + const second = timeoutFetch(url, init); + await Promise.resolve(); + expect(getFetches).toBe(2); + releaseFetch(); + + const [firstResponse, secondResponse] = await Promise.all([first, second]); + expect(firstResponse.headers.get(GITHUB_RESPONSE_CACHE_REPLAY_HEADER)).toBeNull(); + expect(secondResponse.headers.get(GITHUB_RESPONSE_CACHE_REPLAY_HEADER)).toBeNull(); + expect(await firstResponse.text()).toBe("body-1"); + expect(await secondResponse.text()).toBe("body-2"); + }); + + it("keeps volatile single-flight scoped by exact authorization identity", async () => { + let releaseFetch!: () => void; + const fetchGate = new Promise((resolve) => { + releaseFetch = resolve; + }); + let getFetches = 0; + vi.stubGlobal("fetch", async (_input: RequestInfo | URL, init?: RequestInit) => { + getFetches += 1; + await fetchGate; + const authorization = new Headers(init?.headers).get("authorization"); + return Response.json({ caller: authorization?.endsWith("token-a") ? "a" : "b" }); + }); + + const url = "https://api.github.com/repos/o/r/pulls/7/reviews?per_page=100&page=1"; + const firstA = timeoutFetch(url, { headers: { authorization: "Bearer token-a" } }); + const firstB = timeoutFetch(url, { headers: { authorization: "Bearer token-b" } }); + await Promise.resolve(); + expect(getFetches).toBe(2); + releaseFetch(); + + expect(await (await firstA).json()).toEqual({ caller: "a" }); + expect(await (await firstB).json()).toEqual({ caller: "b" }); + }); + + it("lets a coalesced mutable GitHub GET follower honor its own abort signal", async () => { + setGitHubResponseCache({ + get: async () => null, + set: async () => undefined, + }); + let markFetchStarted!: () => void; + const fetchStarted = new Promise((resolve) => { + markFetchStarted = resolve; + }); + let releaseFetch!: () => void; + const fetchGate = new Promise((resolve) => { + releaseFetch = resolve; + }); + vi.stubGlobal("fetch", async () => { + markFetchStarted(); + await fetchGate; + return Response.json({ state: "live" }); + }); + + const url = "https://api.github.com/repos/o/r/pulls/7/reviews?per_page=100&page=1"; + const first = timeoutFetch(url); + await fetchStarted; + const preAborted = new AbortController(); + preAborted.abort("caller stopped"); + await expect(timeoutFetch(new Request(url, { signal: preAborted.signal }))).rejects.toThrow("The operation was aborted."); + const controller = new AbortController(); + const second = timeoutFetch(url, { signal: controller.signal }); + controller.abort(new Error("caller aborted")); + + await expect(second).rejects.toThrow("caller aborted"); + releaseFetch(); + await expect(first.then((response) => response.json())).resolves.toEqual({ state: "live" }); + expect(await renderMetrics()).toContain('gittensory_github_response_cache_total{class="sensitive",result="coalesced"} 2'); + }); + it("bypasses conditional GitHub GETs so validator headers keep shaping the live response", async () => { const store = installMemoryResponseCache(); let getFetches = 0; @@ -501,13 +894,16 @@ describe("timeoutFetch", () => { }); const url = "https://api.github.com/repos/o/r/branches/main/protection/required_status_checks"; - const first = timeoutFetch(url).then((response) => response.status); + const first = timeoutFetch(url); const second = timeoutFetch(url); await bothCacheReads; releaseFetch(); - await expect(first).resolves.toBe(500); - await expect(second.then((response) => response.json())).resolves.toEqual({ contexts: ["after-fallback"] }); + const responses = await Promise.all([first, second]); + expect(responses.map((response) => response.status).sort()).toEqual([200, 500]); + const replayable = responses.find((response) => response.status === 200); + expect(replayable).toBeDefined(); + expect(await replayable!.json()).toEqual({ contexts: ["after-fallback"] }); expect(getFetches).toBe(2); }); diff --git a/test/unit/grounding-wiring.test.ts b/test/unit/grounding-wiring.test.ts index 3cb5ba0c09..f225d09f1c 100644 --- a/test/unit/grounding-wiring.test.ts +++ b/test/unit/grounding-wiring.test.ts @@ -9,6 +9,8 @@ import { makeGithubFileFetcher, } from "../../src/review/grounding-wire"; import { upsertCheckSummary, upsertRepositoryFromGitHub } from "../../src/db/repositories"; +import * as githubApp from "../../src/github/app"; +import { githubRateLimitAdmissionKeyForInstallation, latestGitHubRestRateLimitObservation } from "../../src/github/client"; import type { Advisory, CheckSummaryRecord, JsonValue, PullRequestFileRecord, RepositorySettings } from "../../src/types"; import { createTestEnv } from "../helpers/d1"; @@ -511,6 +513,43 @@ describe("makeGithubFileFetcher (GitHub Contents-API-backed FileFetcher)", () => expect(sawAuth).toBe("Bearer ghp_public"); fetchSpy.mockRestore(); }); + + it("uses installation-token contents reads and records admission telemetry when token mint succeeds", async () => { + const env = createTestEnv({ GITHUB_PUBLIC_TOKEN: "ghp_public" }); + const key = githubRateLimitAdmissionKeyForInstallation(12345); + const tokenSpy = vi.spyOn(githubApp, "createInstallationToken").mockResolvedValue("install-token"); + let sawAuth: string | null = null; + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const fetchSpy = vi.spyOn(globalThis, "fetch").mockImplementation(async (_url, init) => { + const headers = new Headers(init?.headers); + sawAuth = headers.get("authorization"); + return new Response("private body", { + status: 200, + headers: { + "x-ratelimit-resource": "core", + "x-ratelimit-remaining": "22", + "x-ratelimit-reset": String(Date.parse("2026-06-24T12:10:00.000Z") / 1000), + }, + }); + }); + + try { + const fetcher = await makeGithubFileFetcher(env, "acme/widgets", 12345); + expect(await fetcher.getFileContent("ok.ts", "sha7")).toBe("private body"); + expect(tokenSpy).toHaveBeenCalledWith(env, 12345); + expect(sawAuth).toBe("Bearer install-token"); + expect(latestGitHubRestRateLimitObservation(key)).toEqual({ + remaining: 22, + resetAt: "2026-06-24T12:10:00.000Z", + observedAtMs: Date.parse("2026-06-24T12:00:00.000Z"), + }); + } finally { + fetchSpy.mockRestore(); + tokenSpy.mockRestore(); + vi.useRealTimers(); + } + }); }); // ── checkSummaryText empty fallback + outer-catch fail-safe ───────────────────────────────────────── diff --git a/test/unit/mcp-output-schemas.test.ts b/test/unit/mcp-output-schemas.test.ts index 64e71717ef..51be3e9645 100644 --- a/test/unit/mcp-output-schemas.test.ts +++ b/test/unit/mcp-output-schemas.test.ts @@ -1,6 +1,6 @@ import { Client } from "@modelcontextprotocol/sdk/client/index.js"; import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js"; -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { persistSignalSnapshot, upsertBounty, upsertIssueFromGitHub, upsertPullRequestFromGitHub, upsertRepositoryFromGitHub, updatePullRequestSlopAssessment } from "../../src/db/repositories"; import type { AuthIdentity } from "../../src/auth/security"; import { GittensoryMcp } from "../../src/mcp/server"; @@ -562,10 +562,33 @@ function repoOutcomePatternsPayload(repoFullName: string, generatedAt: string) { }; } +function stubMcpSchemaValidationNetwork(): void { + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + const url = String(input); + if (/^https:\/\/api\.github\.com\/users\/[^/]+\/repos(?:\?|$)/.test(url)) { + return Response.json([{ language: "TypeScript" }]); + } + if (/^https:\/\/api\.github\.com\/users\/[^/?#]+(?:[?#]|$)/.test(url)) { + return Response.json({ + login: "oktofeesh1", + name: "Okto", + public_repos: 1, + followers: 1, + created_at: "2024-01-01T00:00:00Z", + updated_at: "2026-06-14T00:00:00Z", + }); + } + if (url === "https://api.gittensor.io/miners") return Response.json([]); + if (url.startsWith("https://raw.githubusercontent.com/")) return new Response("not found", { status: 404 }); + return Response.json({}, { status: 404 }); + }); +} + // ── #550: the previously-unschematized tools are now call-tested so a future schema/type mismatch // (which surfaces as an "Output validation error" → isError) can't slip through CI. ───────────── describe("MCP output schemas validate on real tool calls (#550)", () => { it("every newly-schematized tool returns schema-valid structured content", async () => { + stubMcpSchemaValidationNetwork(); const env = createTestEnv(); await persistRegistrySnapshot( env, @@ -612,6 +635,7 @@ describe("MCP output schemas validate on real tool calls (#550)", () => { const fetched = await client.callTool({ name: "gittensory_agent_get_run", arguments: { runId } }); expect(fetched.isError, `agent_get_run errored: ${JSON.stringify(fetched.content)}`).toBeFalsy(); expect(fetched.structuredContent).toBeDefined(); + vi.unstubAllGlobals(); }, 30_000); }); diff --git a/test/unit/preview-url.test.ts b/test/unit/preview-url.test.ts new file mode 100644 index 0000000000..dbc101ae78 --- /dev/null +++ b/test/unit/preview-url.test.ts @@ -0,0 +1,48 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { clearGitHubResponseCacheForTest, githubRateLimitAdmissionKeyForInstallation, latestGitHubRestRateLimitObservation } from "../../src/github/client"; +import { getPreviewBuildState } from "../../src/review/visual/preview-url"; + +afterEach(() => { + clearGitHubResponseCacheForTest(); + vi.unstubAllGlobals(); + vi.useRealTimers(); +}); + +describe("preview-url GitHub reads", () => { + it("records REST admission telemetry only for installation-token preview lookups", async () => { + const key = githubRateLimitAdmissionKeyForInstallation(123); + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + vi.stubGlobal("fetch", async () => + Response.json( + { check_runs: [] }, + { + headers: { + "x-ratelimit-resource": "core", + "x-ratelimit-remaining": "42", + "x-ratelimit-reset": String(Date.parse("2026-06-24T12:10:00.000Z") / 1000), + }, + }, + ), + ); + + await expect( + getPreviewBuildState({ token: "dummy-user-token", repo: { owner: "o", repo: "r" }, sha: "abc123" }), + ).resolves.toBe("absent"); + expect(latestGitHubRestRateLimitObservation(key)).toBeNull(); + + await expect( + getPreviewBuildState({ + token: "dummy-installation-token", + repo: { owner: "o", repo: "r" }, + sha: "abc123", + rateLimitAdmissionKey: key, + }), + ).resolves.toBe("absent"); + expect(latestGitHubRestRateLimitObservation(key)).toEqual({ + remaining: 42, + resetAt: "2026-06-24T12:10:00.000Z", + observedAtMs: Date.parse("2026-06-24T12:00:00.000Z"), + }); + }); +}); diff --git a/test/unit/rag-index.test.ts b/test/unit/rag-index.test.ts index 3cb59f8cef..4e9c1dac94 100644 --- a/test/unit/rag-index.test.ts +++ b/test/unit/rag-index.test.ts @@ -4,6 +4,8 @@ import { MAX_CHUNKS_PER_REPO, MAX_FILE_BYTES, RAG_DIMENSIONS, ragNamespace } fro import { processJob, splitRepoForRag } from "../../src/queue/processors"; import { upsertRepositoryFromGitHub } from "../../src/db/repositories"; import { upsertRepoFocusManifest } from "../../src/signals/focus-manifest-loader"; +import * as githubApp from "../../src/github/app"; +import { githubRateLimitAdmissionKeyForInstallation, latestGitHubRestRateLimitObservation } from "../../src/github/client"; import { createTestEnv, TestD1Database } from "../helpers/d1"; // A valid bge-m3-width (1024-d) embedding vector — embedTexts rejects any other width. @@ -204,6 +206,43 @@ describe("indexRepo: full repo index (tree → chunk → embed → upsert)", () expect(inits.every((i) => i?.signal instanceof AbortSignal)).toBe(true); }); + it("uses installation-token reads for private RAG indexing and records admission telemetry", async () => { + const { env } = indexEnv(); + const key = githubRateLimitAdmissionKeyForInstallation(123); + const tokenSpy = vi.spyOn(githubApp, "createInstallationToken").mockResolvedValue("install-token"); + const authHeaders: Array = []; + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => { + authHeaders.push(new Headers(init?.headers).get("authorization")); + const headers = { + "x-ratelimit-resource": "core", + "x-ratelimit-remaining": "44", + "x-ratelimit-reset": String(Date.parse("2026-06-24T12:10:00.000Z") / 1000), + }; + const url = String(input); + if (url.includes("/git/trees/")) return Response.json({ tree: [{ type: "blob", path: "src/private.ts", size: 20 }] }, { headers }); + if (url.includes("/contents/src/private.ts")) return new Response("export const privateFile = true;\n", { status: 200, headers }); + return new Response("missing", { status: 404, headers }); + }); + + try { + const result = await indexRepo(env, PROJECT, { ...REPO, installationId: 123 }); + + expect(result).toMatchObject({ files: 1, indexed: 1, capped: false }); + expect(tokenSpy).toHaveBeenCalledWith(env, 123); + expect(authHeaders).toEqual(["Bearer install-token", "Bearer install-token"]); + expect(latestGitHubRestRateLimitObservation(key)).toEqual({ + remaining: 44, + resetAt: "2026-06-24T12:10:00.000Z", + observedAtMs: Date.parse("2026-06-24T12:00:00.000Z"), + }); + } finally { + tokenSpy.mockRestore(); + vi.useRealTimers(); + } + }); + it("a storage error while listing stored paths is fail-safe (prunes nothing, still indexes) + surfaces it at ERROR for Sentry (#5)", async () => { const { env } = indexEnv(); const errSpy = vi.spyOn(console, "error").mockImplementation(() => {}); diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index f9ce7b785a..843b54d900 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -36,17 +36,35 @@ interface MockPool { /** Pre-load a job to be returned by the next RETURNING claim query. */ enqueueJob(id: string, payload: object, attempts?: number, jobKey?: string | null): void; setDeferUpdateRowCount(rowCount: number): void; - setRateLimitRows(rows: Array<{ remaining: number | string | null; reset_at: string | null }>): void; + setRateLimitRows(rows: Array<{ admission_key?: string | null; repo_full_name?: string | null; remaining: number | string | null; reset_at: string | null; observed_at?: string | null }>): void; } function makePool(): MockPool { const results: Partial[] = []; let deferUpdateRowCount = 1; - let rateLimitRows: Array<{ remaining: number | string | null; reset_at: string | null }> = []; - const fn = vi.fn().mockImplementation(async (sql: unknown) => { + let rateLimitRows: Array<{ admission_key?: string | null; repo_full_name?: string | null; remaining: number | string | null; reset_at: string | null; observed_at?: string | null }> = []; + const fn = vi.fn().mockImplementation(async (sql: unknown, params?: unknown[]) => { const q = String(sql); if (q.includes("FROM github_rate_limit_observations")) { - return { rows: rateLimitRows, rowCount: rateLimitRows.length }; + const admissionKey = typeof params?.[0] === "string" ? params[0] : null; + const remainingFloor = typeof params?.[1] === "number" ? params[1] : Number.POSITIVE_INFINITY; + const rows = rateLimitRows + .filter( + (row) => + (admissionKey !== null && row.admission_key === admissionKey) || + row.admission_key === undefined || + row.admission_key === null, + ) + .sort((a, b) => { + const unsafe = + (Number(b.remaining) <= remainingFloor ? 1 : 0) - (Number(a.remaining) <= remainingFloor ? 1 : 0); + if (unsafe !== 0) return unsafe; + const observed = Date.parse(b.observed_at ?? "") - Date.parse(a.observed_at ?? ""); + if (Number.isFinite(observed) && observed !== 0) return observed; + return 0; + }) + .slice(0, 16); + return { rows, rowCount: rows.length }; } if (q.includes("SET status='pending', run_after=GREATEST")) { return { rows: [], rowCount: deferUpdateRowCount }; @@ -399,7 +417,7 @@ describe("createPgQueue (durable #977)", () => { process.env.QUEUE_RATE_LIMIT_JITTER_MS = "0"; try { const m = makePool(); - m.setRateLimitRows([{ remaining: "120", reset_at: "2026-06-24T12:10:00.000Z" }]); + m.setRateLimitRows([{ admission_key: "installation:123", repo_full_name: "owner/other-repo", remaining: "120", reset_at: "2026-06-24T12:10:00.000Z", observed_at: "2026-06-24T11:59:30.000Z" }]); m.enqueueJob("background", { type: "agent-regate-pr", deliveryId: "sweep:owner/repo#7", @@ -427,6 +445,191 @@ describe("createPgQueue (durable #977)", () => { } }); + it("pre-yields GitHub-budget background jobs without repo fields from global REST observations", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const oldJitter = process.env.QUEUE_RATE_LIMIT_JITTER_MS; + process.env.QUEUE_RATE_LIMIT_JITTER_MS = "0"; + try { + const m = makePool(); + m.setRateLimitRows([{ admission_key: null, repo_full_name: "owner/repo", remaining: "120", reset_at: "2026-06-24T12:10:00.000Z", observed_at: "2026-06-24T12:00:00.000Z" }]); + m.enqueueJob("background", { + type: "agent-regate-sweep", + requestedBy: "schedule", + }); + const seen: string[] = []; + const q = createPgQueue(m.pool, async (j) => void seen.push(typeOf(j))); + + await q.drain(); + + expect(seen).toEqual([]); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("SET status='pending', run_after=GREATEST"), + [Date.parse("2026-06-24T12:10:15.000Z"), "github rate-limit background admission", "background"], + ); + } finally { + if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; + else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; + } + }); + + it("pre-yields repo-scoped background jobs from global unkeyed REST observations", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const oldJitter = process.env.QUEUE_RATE_LIMIT_JITTER_MS; + process.env.QUEUE_RATE_LIMIT_JITTER_MS = "0"; + try { + const m = makePool(); + m.setRateLimitRows([{ admission_key: null, repo_full_name: "owner/other-repo", remaining: "120", reset_at: "2026-06-24T12:10:00.000Z", observed_at: "2026-06-24T12:00:00.000Z" }]); + m.enqueueJob("background", { + type: "rag-index-repo", + requestedBy: "schedule", + repoFullName: "owner/repo", + }); + const seen: string[] = []; + const q = createPgQueue(m.pool, async (j) => void seen.push(typeOf(j))); + + await q.drain(); + + expect(seen).toEqual([]); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("SET status='pending', run_after=GREATEST"), + [Date.parse("2026-06-24T12:10:15.000Z"), "github rate-limit background admission", "background"], + ); + } finally { + if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; + else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; + } + }); + + it("pre-yields webhook jobs when the persisted REST bucket is exhausted", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const oldJitter = process.env.QUEUE_RATE_LIMIT_JITTER_MS; + process.env.QUEUE_RATE_LIMIT_JITTER_MS = "0"; + try { + const m = makePool(); + m.setRateLimitRows([{ admission_key: "installation:123", repo_full_name: "owner/other-repo", remaining: "50", reset_at: "2026-06-24T12:10:00.000Z", observed_at: "2026-06-24T11:59:30.000Z" }]); + m.enqueueJob("webhook", { type: "github-webhook", deliveryId: "fresh", eventName: "pull_request", payload: { installation: { id: 123 }, repository: { full_name: "owner/repo" } } }); + const seen: string[] = []; + const q = createPgQueue(m.pool, async (j) => void seen.push(typeOf(j))); + + await q.drain(); + + expect(seen).toEqual([]); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("SET status='pending', run_after=GREATEST"), + [Date.parse("2026-06-24T12:10:15.000Z"), "github rate-limit webhook admission", "webhook"], + ); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("INSERT INTO _selfhost_job_stats"), + ["gittensory_jobs_rate_limit_deferred_total", 1], + ); + } finally { + if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; + else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; + } + }); + + it("pre-yields webhook jobs from global legacy observations when an installation id is present", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const oldJitter = process.env.QUEUE_RATE_LIMIT_JITTER_MS; + process.env.QUEUE_RATE_LIMIT_JITTER_MS = "0"; + try { + const m = makePool(); + m.setRateLimitRows([{ admission_key: null, repo_full_name: "owner/other-repo", remaining: "50", reset_at: "2026-06-24T12:10:00.000Z", observed_at: "2026-06-24T12:00:00.000Z" }]); + m.enqueueJob("webhook", { type: "github-webhook", deliveryId: "fresh", eventName: "pull_request", payload: { installation: { id: 123 }, repository: { full_name: "owner/repo" } } }); + const seen: string[] = []; + const q = createPgQueue(m.pool, async (j) => void seen.push(typeOf(j))); + + await q.drain(); + + expect(seen).toEqual([]); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("SET status='pending', run_after=GREATEST"), + [Date.parse("2026-06-24T12:10:15.000Z"), "github rate-limit webhook admission", "webhook"], + ); + } finally { + if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; + else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; + } + }); + + it("pre-yields from legacy repo exhaustion before older healthy exact observations", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const oldJitter = process.env.QUEUE_RATE_LIMIT_JITTER_MS; + process.env.QUEUE_RATE_LIMIT_JITTER_MS = "0"; + try { + const m = makePool(); + m.setRateLimitRows([ + { admission_key: "installation:123", repo_full_name: "owner/other-repo", remaining: "4000", reset_at: "2026-06-24T12:20:00.000Z", observed_at: "2026-06-24T11:59:00.000Z" }, + { admission_key: null, repo_full_name: "owner/repo", remaining: "0", reset_at: "2026-06-24T12:10:00.000Z", observed_at: "2026-06-24T12:00:00.000Z" }, + ]); + m.enqueueJob("webhook", { type: "github-webhook", deliveryId: "fresh", eventName: "pull_request", payload: { installation: { id: 123 }, repository: { full_name: "owner/repo" } } }); + const seen: string[] = []; + const q = createPgQueue(m.pool, async (j) => void seen.push(typeOf(j))); + + await q.drain(); + + expect(seen).toEqual([]); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("SET status='pending', run_after=GREATEST"), + [Date.parse("2026-06-24T12:10:15.000Z"), "github rate-limit webhook admission", "webhook"], + ); + } finally { + if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; + else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; + } + }); + + it("pre-yields from exact admission exhaustion before newer legacy repo observations", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const oldJitter = process.env.QUEUE_RATE_LIMIT_JITTER_MS; + process.env.QUEUE_RATE_LIMIT_JITTER_MS = "0"; + try { + const m = makePool(); + m.setRateLimitRows([ + { admission_key: "installation:123", repo_full_name: "owner/other-repo", remaining: "0", reset_at: "2026-06-24T12:10:00.000Z", observed_at: "2026-06-24T11:59:00.000Z" }, + { admission_key: null, repo_full_name: "owner/repo", remaining: "4000", reset_at: "2026-06-24T12:20:00.000Z", observed_at: "2026-06-24T12:00:00.000Z" }, + ]); + m.enqueueJob("webhook", { type: "github-webhook", deliveryId: "fresh", eventName: "pull_request", payload: { installation: { id: 123 }, repository: { full_name: "owner/repo" } } }); + const seen: string[] = []; + const q = createPgQueue(m.pool, async (j) => void seen.push(typeOf(j))); + + await q.drain(); + + expect(seen).toEqual([]); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("SET status='pending', run_after=GREATEST"), + [Date.parse("2026-06-24T12:10:15.000Z"), "github rate-limit webhook admission", "webhook"], + ); + } finally { + if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; + else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; + } + }); + + it("does not pre-yield webhook jobs for another installation's persisted REST exhaustion", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const m = makePool(); + m.setRateLimitRows([{ admission_key: "installation:456", repo_full_name: "owner/repo-a", remaining: "0", reset_at: "2026-06-24T12:10:00.000Z", observed_at: "2026-06-24T11:59:30.000Z" }]); + m.enqueueJob("webhook", { type: "github-webhook", deliveryId: "fresh", eventName: "pull_request", payload: { installation: { id: 123 }, repository: { full_name: "owner/repo-b" } } }); + const seen: string[] = []; + const q = createPgQueue(m.pool, async (j) => void seen.push(typeOf(j))); + + await q.drain(); + + expect(seen).toEqual(["github-webhook"]); + expect(m.pool.query).not.toHaveBeenCalledWith( + expect.stringContaining("SET status='pending', run_after=GREATEST"), + expect.anything(), + ); + }); + it("skips the background-admission metric when the defer update changes no rows", async () => { vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); @@ -435,10 +638,11 @@ describe("createPgQueue (durable #977)", () => { try { const m = makePool(); m.setDeferUpdateRowCount(0); - m.setRateLimitRows([{ remaining: 120, reset_at: "2026-06-24T12:10:00.000Z" }]); + m.setRateLimitRows([{ repo_full_name: "owner/repo", remaining: 120, reset_at: "2026-06-24T12:10:00.000Z" }]); m.enqueueJob("background", { type: "rag-index-repo", requestedBy: "schedule", + repoFullName: "owner/repo", }); const warned = vi.spyOn(console, "warn").mockImplementation(() => undefined); const seen: string[] = []; diff --git a/test/unit/selfhost-queue-common.test.ts b/test/unit/selfhost-queue-common.test.ts index 91d2986c61..54e5b8cfe5 100644 --- a/test/unit/selfhost-queue-common.test.ts +++ b/test/unit/selfhost-queue-common.test.ts @@ -1,9 +1,12 @@ -import { describe, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { FOREGROUND_QUEUE_PRIORITY_FLOOR, consumingRetryDelayMs, + githubRateLimitAdmissionDelayMs, + githubRateLimitAdmissionKeyForJob, githubBackgroundRateLimitDelayMs, githubRateLimitRetryDelayMs, + githubWebhookRateLimitDelayMs, isGitHubBudgetBackgroundJob, isForegroundJobPriority, jobCoalesceKey, @@ -15,11 +18,18 @@ import { queueStartupJitterMinJobs, queueStartupJitterMs, } from "../../src/selfhost/queue-common"; +import { clearGitHubResponseCacheForTest, githubRateLimitAdmissionKeyForInstallation, timeoutFetch } from "../../src/github/client"; import { RetryableJobError } from "../../src/queue/retryable"; import type { JobMessage } from "../../src/types"; const payload = (value: unknown): string => JSON.stringify(value); +afterEach(() => { + clearGitHubResponseCacheForTest(); + vi.unstubAllGlobals(); + vi.useRealTimers(); +}); + describe("self-host queue common helpers", () => { it("classifies job priority by job type and webhook sender", () => { expect(jobPriority(payload({ type: "github-webhook" }))).toBe(10); @@ -60,6 +70,12 @@ describe("self-host queue common helpers", () => { expect(isGitHubBudgetBackgroundJob({ type: "refresh-installation-health", requestedBy: "schedule" })).toBe(false); }); + it("derives admission keys from both installation-backed jobs and webhook payloads", () => { + expect(githubRateLimitAdmissionKeyForJob({ type: "agent-regate-pr", deliveryId: "sweep:owner/repo#1", repoFullName: "owner/repo", prNumber: 1, installationId: 123 })).toBe("installation:123"); + expect(githubRateLimitAdmissionKeyForJob({ type: "github-webhook", deliveryId: "d1", eventName: "pull_request", payload: { installation: { id: 456 } } })).toBe("installation:456"); + expect(githubRateLimitAdmissionKeyForJob({ type: "github-webhook", deliveryId: "d2", eventName: "pull_request", payload: {} })).toBeNull(); + }); + it("computes background admission delays from persisted GitHub REST observations", () => { const now = Date.parse("2026-06-24T12:00:00.000Z"); expect(githubBackgroundRateLimitDelayMs(null, now)).toBeNull(); @@ -72,6 +88,106 @@ describe("self-host queue common helpers", () => { expect(githubBackgroundRateLimitDelayMs({ remaining: 120, reset_at: "2026-06-24T14:00:00.000Z" }, now)).toBe(900_000); }); + it("computes webhook admission delays only when the shared REST bucket is exhausted", () => { + const now = Date.parse("2026-06-24T12:00:00.000Z"); + expect(githubWebhookRateLimitDelayMs(null, now)).toBeNull(); + expect(githubWebhookRateLimitDelayMs({ remaining: 76, reset_at: "2026-06-24T12:10:00.000Z" }, now)).toBeNull(); + expect(githubWebhookRateLimitDelayMs({ remaining: 75, reset_at: "2026-06-24T12:10:00.000Z" }, now)).toBe(615_000); + expect(githubWebhookRateLimitDelayMs({ remaining: "50", reset_at: "2026-06-24T12:10:00.000Z" }, now)).toBe(615_000); + expect(githubWebhookRateLimitDelayMs({ remaining: 50, resetAt: "2026-06-24T12:00:05.000Z" }, now)).toBe(30_000); + expect(githubWebhookRateLimitDelayMs({ remaining: 50, reset_at: "2026-06-24T11:59:00.000Z" }, now)).toBeNull(); + }); + + it("computes admission delays from any unsafe persisted candidate", () => { + const now = Date.parse("2026-06-24T12:00:00.000Z"); + expect( + githubRateLimitAdmissionDelayMs( + "webhook", + null, + [ + { remaining: 4000, reset_at: "2026-06-24T12:20:00.000Z", observed_at: "2026-06-24T12:00:00.000Z" }, + { remaining: 0, reset_at: "2026-06-24T12:10:00.000Z", observed_at: "2026-06-24T11:59:00.000Z" }, + ], + now, + ), + ).toBe(615_000); + expect(githubRateLimitAdmissionDelayMs("background", null, [], now)).toBeNull(); + }); + + it("uses the newest local REST rate-limit observation for admission control", async () => { + const now = Date.parse("2026-06-24T12:00:00.000Z"); + const key = githubRateLimitAdmissionKeyForInstallation(123); + const unrelatedKey = githubRateLimitAdmissionKeyForInstallation(456); + vi.useFakeTimers(); + vi.setSystemTime(now); + vi.stubGlobal( + "fetch", + async () => + new Response("{}", { + status: 200, + headers: { + "content-type": "application/json", + "x-ratelimit-resource": "core", + "x-ratelimit-remaining": "50", + "x-ratelimit-reset": String(Math.floor(Date.parse("2026-06-24T12:10:00.000Z") / 1000)), + }, + }), + ); + + await timeoutFetch("https://api.github.com/repos/owner/repo/issues", { githubRateLimitAdmission: true, githubRateLimitAdmissionKey: key }); + + expect(githubRateLimitAdmissionDelayMs("webhook", key, null, now)).toBe(615_000); + expect(githubRateLimitAdmissionDelayMs("webhook", unrelatedKey, null, now)).toBeNull(); + expect( + githubRateLimitAdmissionDelayMs( + "webhook", + key, + { remaining: 500, reset_at: "2026-06-24T12:10:00.000Z", observed_at: "2026-06-24T11:59:00.000Z" }, + now, + ), + ).toBe(615_000); + expect( + githubRateLimitAdmissionDelayMs( + "webhook", + key, + { remaining: 500, reset_at: "2026-06-24T12:10:00.000Z", observed_at: "2026-06-24T12:01:00.000Z" }, + now, + ), + ).toBeNull(); + expect( + githubRateLimitAdmissionDelayMs( + "webhook", + key, + { remaining: 500, reset_at: "2026-06-24T12:10:00.000Z", observedAtMs: now + 60_000 } as unknown as { remaining: number; reset_at: string }, + now, + ), + ).toBeNull(); + expect( + githubRateLimitAdmissionDelayMs( + "webhook", + key, + { remaining: 500, reset_at: "2026-06-24T12:10:00.000Z", observedAt: "2026-06-24T12:01:00.000Z" }, + now, + ), + ).toBeNull(); + expect( + githubRateLimitAdmissionDelayMs( + "webhook", + key, + { remaining: 500, reset_at: "2026-06-24T12:10:00.000Z", observed_at: "not-a-date" }, + now, + ), + ).toBe(615_000); + expect( + githubRateLimitAdmissionDelayMs( + "webhook", + key, + { remaining: 500, reset_at: "2026-06-24T12:10:00.000Z" }, + now, + ), + ).toBe(615_000); + }); + it("demotes bot-authored issue-comment edit webhooks without demoting human reruns", () => { const issueCommentEdit = (sender: { login?: string; type?: string }) => payload({ diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index b97353c663..aa4e9c07d6 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -120,6 +120,7 @@ describe("createSqliteQueue (durable #980)", () => { `CREATE TABLE github_rate_limit_observations ( id TEXT PRIMARY KEY, repo_full_name TEXT NOT NULL, + admission_key TEXT, resource TEXT NOT NULL, path TEXT NOT NULL, status_code INTEGER NOT NULL, @@ -131,15 +132,20 @@ describe("createSqliteQueue (durable #980)", () => { [], ); driver.query( - `INSERT INTO github_rate_limit_observations (id, repo_full_name, resource, path, status_code, limit_value, remaining, reset_at, observed_at) - VALUES (?, ?, 'rest', ?, 200, 5000, 120, ?, ?)`, - ["rl-bg", "owner/repo", "/x", "2026-06-24T12:10:00.000Z", "2026-06-24T12:00:00.000Z"], + `INSERT INTO github_rate_limit_observations (id, repo_full_name, admission_key, resource, path, status_code, limit_value, remaining, reset_at, observed_at) + VALUES (?, ?, ?, 'rest', ?, 200, 5000, 120, ?, ?)`, + ["rl-bg-installation", "owner/other-repo", "installation:123", "/x", "2026-06-24T12:10:00.000Z", "2026-06-24T12:00:00.000Z"], + ); + driver.query( + `INSERT INTO github_rate_limit_observations (id, repo_full_name, admission_key, resource, path, status_code, limit_value, remaining, reset_at, observed_at) + VALUES (?, ?, NULL, 'rest', ?, 200, 5000, 120, ?, ?)`, + ["rl-bg-legacy", "owner/repo", "/x", "2026-06-24T12:10:00.000Z", "2026-06-24T12:00:00.000Z"], ); const seen: string[] = []; const q = createSqliteQueue(driver, async (m) => void seen.push(typeOf(m))); await q.binding.send({ type: "agent-regate-pr", deliveryId: "sweep:owner/repo#7", repoFullName: "owner/repo", prNumber: 7, installationId: 123 }); - await q.binding.send({ type: "rag-index-repo", requestedBy: "schedule" }); + await q.binding.send({ type: "rag-index-repo", requestedBy: "schedule", repoFullName: "owner/repo" }); await q.binding.send({ type: "github-webhook", deliveryId: "fresh", eventName: "pull_request", payload: {} }); await q.drain(); @@ -167,6 +173,357 @@ describe("createSqliteQueue (durable #980)", () => { } }); + it("pre-yields GitHub-budget background jobs without repo fields from global REST observations", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const oldJitter = process.env.QUEUE_RATE_LIMIT_JITTER_MS; + process.env.QUEUE_RATE_LIMIT_JITTER_MS = "0"; + try { + const driver = makeDriver(); + driver.query( + `CREATE TABLE github_rate_limit_observations ( + id TEXT PRIMARY KEY, + repo_full_name TEXT NOT NULL, + admission_key TEXT, + resource TEXT NOT NULL, + path TEXT NOT NULL, + status_code INTEGER NOT NULL, + limit_value INTEGER, + remaining INTEGER, + reset_at TEXT, + observed_at TEXT NOT NULL + )`, + [], + ); + driver.query( + `INSERT INTO github_rate_limit_observations (id, repo_full_name, admission_key, resource, path, status_code, limit_value, remaining, reset_at, observed_at) + VALUES (?, ?, NULL, 'rest', ?, 200, 5000, 120, ?, ?)`, + ["rl-bg-global", "owner/repo", "/x", "2026-06-24T12:10:00.000Z", "2026-06-24T12:00:00.000Z"], + ); + const seen: string[] = []; + const q = createSqliteQueue(driver, async (m) => void seen.push(typeOf(m))); + + await q.binding.send({ type: "agent-regate-sweep", requestedBy: "schedule" }); + await q.drain(); + + expect(seen).toEqual([]); + const row = driver.query( + "SELECT status, attempts, run_after, last_error FROM _selfhost_jobs", + [], + ).rows[0] as { status: string; attempts: number; run_after: number; last_error: string }; + expect(row).toMatchObject({ + status: "pending", + attempts: 0, + run_after: Date.parse("2026-06-24T12:10:15.000Z"), + last_error: "github rate-limit background admission", + }); + } finally { + if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; + else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; + vi.useRealTimers(); + } + }); + + it("pre-yields repo-scoped background jobs from global unkeyed REST observations", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const oldJitter = process.env.QUEUE_RATE_LIMIT_JITTER_MS; + process.env.QUEUE_RATE_LIMIT_JITTER_MS = "0"; + try { + const driver = makeDriver(); + driver.query( + `CREATE TABLE github_rate_limit_observations ( + id TEXT PRIMARY KEY, + repo_full_name TEXT NOT NULL, + admission_key TEXT, + resource TEXT NOT NULL, + path TEXT NOT NULL, + status_code INTEGER NOT NULL, + limit_value INTEGER, + remaining INTEGER, + reset_at TEXT, + observed_at TEXT NOT NULL + )`, + [], + ); + driver.query( + `INSERT INTO github_rate_limit_observations (id, repo_full_name, admission_key, resource, path, status_code, limit_value, remaining, reset_at, observed_at) + VALUES (?, ?, NULL, 'rest', ?, 200, 5000, 120, ?, ?)`, + ["rl-bg-global-repo", "owner/other-repo", "/x", "2026-06-24T12:10:00.000Z", "2026-06-24T12:00:00.000Z"], + ); + const seen: string[] = []; + const q = createSqliteQueue(driver, async (m) => void seen.push(typeOf(m))); + + await q.binding.send({ type: "rag-index-repo", requestedBy: "schedule", repoFullName: "owner/repo" }); + await q.drain(); + + expect(seen).toEqual([]); + const row = driver.query( + "SELECT status, attempts, run_after, last_error FROM _selfhost_jobs", + [], + ).rows[0] as { status: string; attempts: number; run_after: number; last_error: string }; + expect(row).toMatchObject({ + status: "pending", + attempts: 0, + run_after: Date.parse("2026-06-24T12:10:15.000Z"), + last_error: "github rate-limit background admission", + }); + } finally { + if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; + else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; + vi.useRealTimers(); + } + }); + + it("pre-yields webhook jobs when the persisted REST bucket is exhausted", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const oldJitter = process.env.QUEUE_RATE_LIMIT_JITTER_MS; + process.env.QUEUE_RATE_LIMIT_JITTER_MS = "0"; + try { + const driver = makeDriver(); + driver.query( + `CREATE TABLE github_rate_limit_observations ( + id TEXT PRIMARY KEY, + repo_full_name TEXT NOT NULL, + admission_key TEXT, + resource TEXT NOT NULL, + path TEXT NOT NULL, + status_code INTEGER NOT NULL, + limit_value INTEGER, + remaining INTEGER, + reset_at TEXT, + observed_at TEXT NOT NULL + )`, + [], + ); + driver.query( + `INSERT INTO github_rate_limit_observations (id, repo_full_name, admission_key, resource, path, status_code, limit_value, remaining, reset_at, observed_at) + VALUES (?, ?, ?, 'rest', ?, 403, 5000, 50, ?, ?)`, + ["rl-webhook", "owner/other-repo", "installation:123", "/x", "2026-06-24T12:10:00.000Z", "2026-06-24T12:00:00.000Z"], + ); + const seen: string[] = []; + const q = createSqliteQueue(driver, async (m) => void seen.push(typeOf(m))); + + await q.binding.send({ type: "github-webhook", deliveryId: "fresh", eventName: "pull_request", payload: { installation: { id: 123 }, repository: { full_name: "owner/repo" } } }); + await q.drain(); + + expect(seen).toEqual([]); + const row = driver.query( + "SELECT status, attempts, run_after, last_error FROM _selfhost_jobs", + [], + ).rows[0] as { status: string; attempts: number; run_after: number; last_error: string }; + expect(row).toMatchObject({ + status: "pending", + attempts: 0, + run_after: Date.parse("2026-06-24T12:10:15.000Z"), + last_error: "github rate-limit webhook admission", + }); + expect(q.stats()).toMatchObject({ gittensory_jobs_rate_limit_deferred_total: 1 }); + } finally { + if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; + else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; + vi.useRealTimers(); + } + }); + + it("pre-yields webhook jobs from global legacy observations when an installation id is present", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const oldJitter = process.env.QUEUE_RATE_LIMIT_JITTER_MS; + process.env.QUEUE_RATE_LIMIT_JITTER_MS = "0"; + try { + const driver = makeDriver(); + driver.query( + `CREATE TABLE github_rate_limit_observations ( + id TEXT PRIMARY KEY, + repo_full_name TEXT NOT NULL, + admission_key TEXT, + resource TEXT NOT NULL, + path TEXT NOT NULL, + status_code INTEGER NOT NULL, + limit_value INTEGER, + remaining INTEGER, + reset_at TEXT, + observed_at TEXT NOT NULL + )`, + [], + ); + driver.query( + `INSERT INTO github_rate_limit_observations (id, repo_full_name, admission_key, resource, path, status_code, limit_value, remaining, reset_at, observed_at) + VALUES (?, ?, NULL, 'rest', ?, 403, 5000, 50, ?, ?)`, + ["rl-webhook-legacy", "owner/other-repo", "/x", "2026-06-24T12:10:00.000Z", "2026-06-24T12:00:00.000Z"], + ); + const seen: string[] = []; + const q = createSqliteQueue(driver, async (m) => void seen.push(typeOf(m))); + + await q.binding.send({ type: "github-webhook", deliveryId: "fresh", eventName: "pull_request", payload: { installation: { id: 123 }, repository: { full_name: "owner/repo" } } }); + await q.drain(); + + expect(seen).toEqual([]); + const row = driver.query( + "SELECT status, attempts, run_after, last_error FROM _selfhost_jobs", + [], + ).rows[0] as { status: string; attempts: number; run_after: number; last_error: string }; + expect(row).toMatchObject({ + status: "pending", + attempts: 0, + run_after: Date.parse("2026-06-24T12:10:15.000Z"), + last_error: "github rate-limit webhook admission", + }); + } finally { + if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; + else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; + vi.useRealTimers(); + } + }); + + it("pre-yields from legacy repo exhaustion before older healthy exact observations", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const oldJitter = process.env.QUEUE_RATE_LIMIT_JITTER_MS; + process.env.QUEUE_RATE_LIMIT_JITTER_MS = "0"; + try { + const driver = makeDriver(); + driver.query( + `CREATE TABLE github_rate_limit_observations ( + id TEXT PRIMARY KEY, + repo_full_name TEXT NOT NULL, + admission_key TEXT, + resource TEXT NOT NULL, + path TEXT NOT NULL, + status_code INTEGER NOT NULL, + limit_value INTEGER, + remaining INTEGER, + reset_at TEXT, + observed_at TEXT NOT NULL + )`, + [], + ); + driver.query( + `INSERT INTO github_rate_limit_observations (id, repo_full_name, admission_key, resource, path, status_code, limit_value, remaining, reset_at, observed_at) + VALUES (?, ?, ?, 'rest', ?, 200, 5000, 4000, ?, ?)`, + ["rl-webhook-installation-old", "owner/other-repo", "installation:123", "/x", "2026-06-24T12:20:00.000Z", "2026-06-24T11:59:00.000Z"], + ); + driver.query( + `INSERT INTO github_rate_limit_observations (id, repo_full_name, admission_key, resource, path, status_code, limit_value, remaining, reset_at, observed_at) + VALUES (?, ?, NULL, 'rest', ?, 403, 5000, 0, ?, ?)`, + ["rl-webhook-legacy-new", "owner/repo", "/x", "2026-06-24T12:10:00.000Z", "2026-06-24T12:00:00.000Z"], + ); + const seen: string[] = []; + const q = createSqliteQueue(driver, async (m) => void seen.push(typeOf(m))); + + await q.binding.send({ type: "github-webhook", deliveryId: "fresh", eventName: "pull_request", payload: { installation: { id: 123 }, repository: { full_name: "owner/repo" } } }); + await q.drain(); + + expect(seen).toEqual([]); + const row = driver.query( + "SELECT status, attempts, run_after, last_error FROM _selfhost_jobs", + [], + ).rows[0] as { status: string; attempts: number; run_after: number; last_error: string }; + expect(row).toMatchObject({ + status: "pending", + attempts: 0, + run_after: Date.parse("2026-06-24T12:10:15.000Z"), + last_error: "github rate-limit webhook admission", + }); + } finally { + if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; + else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; + vi.useRealTimers(); + } + }); + + it("pre-yields from exact admission exhaustion before newer legacy repo observations", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const oldJitter = process.env.QUEUE_RATE_LIMIT_JITTER_MS; + process.env.QUEUE_RATE_LIMIT_JITTER_MS = "0"; + try { + const driver = makeDriver(); + driver.query( + `CREATE TABLE github_rate_limit_observations ( + id TEXT PRIMARY KEY, + repo_full_name TEXT NOT NULL, + admission_key TEXT, + resource TEXT NOT NULL, + path TEXT NOT NULL, + status_code INTEGER NOT NULL, + limit_value INTEGER, + remaining INTEGER, + reset_at TEXT, + observed_at TEXT NOT NULL + )`, + [], + ); + driver.query( + `INSERT INTO github_rate_limit_observations (id, repo_full_name, admission_key, resource, path, status_code, limit_value, remaining, reset_at, observed_at) + VALUES (?, ?, ?, 'rest', ?, 403, 5000, 0, ?, ?)`, + ["rl-webhook-installation-old", "owner/other-repo", "installation:123", "/x", "2026-06-24T12:10:00.000Z", "2026-06-24T11:59:00.000Z"], + ); + driver.query( + `INSERT INTO github_rate_limit_observations (id, repo_full_name, admission_key, resource, path, status_code, limit_value, remaining, reset_at, observed_at) + VALUES (?, ?, NULL, 'rest', ?, 200, 5000, 4000, ?, ?)`, + ["rl-webhook-legacy-new", "owner/repo", "/x", "2026-06-24T12:20:00.000Z", "2026-06-24T12:00:00.000Z"], + ); + const seen: string[] = []; + const q = createSqliteQueue(driver, async (m) => void seen.push(typeOf(m))); + + await q.binding.send({ type: "github-webhook", deliveryId: "fresh", eventName: "pull_request", payload: { installation: { id: 123 }, repository: { full_name: "owner/repo" } } }); + await q.drain(); + + expect(seen).toEqual([]); + const row = driver.query( + "SELECT status, attempts, run_after, last_error FROM _selfhost_jobs", + [], + ).rows[0] as { status: string; attempts: number; run_after: number; last_error: string }; + expect(row).toMatchObject({ + status: "pending", + attempts: 0, + run_after: Date.parse("2026-06-24T12:10:15.000Z"), + last_error: "github rate-limit webhook admission", + }); + } finally { + if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; + else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; + vi.useRealTimers(); + } + }); + + it("does not pre-yield webhook jobs for another installation's persisted REST exhaustion", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + const driver = makeDriver(); + driver.query( + `CREATE TABLE github_rate_limit_observations ( + id TEXT PRIMARY KEY, + repo_full_name TEXT NOT NULL, + admission_key TEXT, + resource TEXT NOT NULL, + path TEXT NOT NULL, + status_code INTEGER NOT NULL, + limit_value INTEGER, + remaining INTEGER, + reset_at TEXT, + observed_at TEXT NOT NULL + )`, + [], + ); + driver.query( + `INSERT INTO github_rate_limit_observations (id, repo_full_name, admission_key, resource, path, status_code, limit_value, remaining, reset_at, observed_at) + VALUES (?, ?, ?, 'rest', ?, 403, 5000, 0, ?, ?)`, + ["rl-webhook", "owner/repo-a", "installation:456", "/x", "2026-06-24T12:10:00.000Z", "2026-06-24T12:00:00.000Z"], + ); + const seen: string[] = []; + const q = createSqliteQueue(driver, async (m) => void seen.push(typeOf(m))); + + await q.binding.send({ type: "github-webhook", deliveryId: "fresh", eventName: "pull_request", payload: { installation: { id: 123 }, repository: { full_name: "owner/repo-b" } } }); + await q.drain(); + + expect(seen).toEqual(["github-webhook"]); + expect(q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); + }); + it("skips the background-admission metric when the defer update changes no rows", async () => { vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); @@ -189,6 +546,7 @@ describe("createSqliteQueue (durable #980)", () => { `CREATE TABLE github_rate_limit_observations ( id TEXT PRIMARY KEY, repo_full_name TEXT NOT NULL, + admission_key TEXT, resource TEXT NOT NULL, path TEXT NOT NULL, status_code INTEGER NOT NULL, @@ -200,15 +558,15 @@ describe("createSqliteQueue (durable #980)", () => { [], ); driver.query( - `INSERT INTO github_rate_limit_observations (id, repo_full_name, resource, path, status_code, limit_value, remaining, reset_at, observed_at) - VALUES (?, ?, 'rest', ?, 200, 5000, 120, ?, ?)`, + `INSERT INTO github_rate_limit_observations (id, repo_full_name, admission_key, resource, path, status_code, limit_value, remaining, reset_at, observed_at) + VALUES (?, ?, NULL, 'rest', ?, 200, 5000, 120, ?, ?)`, ["rl-bg", "owner/repo", "/x", "2026-06-24T12:10:00.000Z", "2026-06-24T12:00:00.000Z"], ); const warned = vi.spyOn(console, "warn").mockImplementation(() => undefined); const seen: string[] = []; const q = createSqliteQueue(driver, async (m) => void seen.push(typeOf(m))); - await q.binding.send({ type: "rag-index-repo", requestedBy: "schedule" }); + await q.binding.send({ type: "rag-index-repo", requestedBy: "schedule", repoFullName: "owner/repo" }); await q.drain(); expect(seen).toEqual([]); diff --git a/test/unit/visual-capture.test.ts b/test/unit/visual-capture.test.ts new file mode 100644 index 0000000000..9cdb46a797 --- /dev/null +++ b/test/unit/visual-capture.test.ts @@ -0,0 +1,79 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + clearGitHubResponseCacheForTest, + githubRateLimitAdmissionKeyForInstallation, + latestGitHubRestRateLimitObservation, +} from "../../src/github/client"; +import { buildCapture } from "../../src/review/visual/capture"; +import { createTestEnv } from "../helpers/d1"; + +afterEach(() => { + clearGitHubResponseCacheForTest(); + vi.unstubAllGlobals(); + vi.useRealTimers(); +}); + +describe("visual capture preview discovery", () => { + it("threads admission telemetry through deployment, checks, comments, and build-state fallbacks", async () => { + const key = githubRateLimitAdmissionKeyForInstallation(123); + const seenUrls: string[] = []; + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-24T12:00:00.000Z")); + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + const url = String(input); + seenUrls.push(url); + const init = { + status: 200, + headers: { + "content-type": "application/json", + "x-ratelimit-resource": "core", + "x-ratelimit-remaining": "33", + "x-ratelimit-reset": String(Date.parse("2026-06-24T12:10:00.000Z") / 1000), + }, + }; + if (url.includes("/deployments?")) return Response.json([], init); + if (url.includes("/status")) return Response.json({ statuses: [] }, init); + if (url.includes("/issues/7/comments")) return Response.json([], init); + if (url.includes("/check-runs")) { + return Response.json( + { check_runs: [{ name: "Cloudflare Workers Builds", status: "completed", conclusion: "failure" }] }, + init, + ); + } + return Response.json({}, init); + }); + + const result = await buildCapture( + createTestEnv({ PUBLIC_API_ORIGIN: "https://worker.example", PUBLIC_SITE_ORIGIN: "" }), + "installation-token", + { + repoFullName: "owner/repo", + prNumber: 7, + headSha: "abc123", + previewFromChecks: true, + }, + ["apps/gittensory-ui/src/routes/app.index.tsx"], + key, + ); + + expect(seenUrls.some((url) => url.includes("/deployments?sha=abc123"))).toBe(true); + expect(seenUrls.some((url) => url.includes("/commits/abc123/status"))).toBe(true); + expect(seenUrls.some((url) => url.includes("/commits/abc123/check-runs"))).toBe(true); + expect(seenUrls.some((url) => url.includes("/issues/7/comments"))).toBe(true); + expect(result.previewPending).toBe(false); + expect(result.routes).toEqual([ + { + path: "/app", + beforeUrl: undefined, + beforeUrlMobile: undefined, + afterUrl: "https://worker.example/gittensory/shot?placeholder=failed", + afterUrlMobile: "https://worker.example/gittensory/shot?placeholder=failed", + }, + ]); + expect(latestGitHubRestRateLimitObservation(key)).toEqual({ + remaining: 33, + resetAt: "2026-06-24T12:10:00.000Z", + observedAtMs: Date.parse("2026-06-24T12:00:00.000Z"), + }); + }); +});