From 2f976579e04badde8c067110e8ad44ccc4affdb6 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Wed, 1 Jul 2026 00:49:19 -0700 Subject: [PATCH 1/2] perf(queue): fan build-contributor-evidence out into per-batch jobs (config-driven) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The scheduled build-contributor-evidence job resolved up to 500 contributors' GitHub reads (/users/{login} + its uncached repos pages) in ONE job — a burst in the 6-hourly full-sync window. The cron trigger now derives the login set and fans out into per-batch jobs (CONTRIBUTOR_EVIDENCE_BATCH_SIZE, default 150), so the per-login reads spread across the queue's paced execution + rate-limit admission. Each batch loads the aggregate data once and processes its batch; the single-login path is unchanged; 0 disables the fan-out. A batch coalesces by its FIRST login (batches are disjoint slices → unique heads) so distinct batches never collapse into one :all key (which would drop work). Advances #1936. --- .env.example | 3 + src/queue/processors.ts | 75 +++++++++++++++++++++---- src/selfhost/queue-common.ts | 17 +++++- src/types.ts | 3 + test/unit/queue.test.ts | 63 ++++++++++++++++++++- test/unit/selfhost-queue-common.test.ts | 17 ++++++ 6 files changed, 163 insertions(+), 15 deletions(-) diff --git a/.env.example b/.env.example index cccd3993fb..79106403f6 100644 --- a/.env.example +++ b/.env.example @@ -199,6 +199,9 @@ REDIS_URL=redis://redis:6379 # REQUIRED for the self-host review # --- Queue worker (#977/#1201) --- # QUEUE_CONCURRENCY=4 # max concurrent job-processing loops per instance (default 4; set 1 for strict serial processing) # QUEUE_BACKGROUND_CONCURRENCY=1 # max low-priority/background jobs allowed to occupy QUEUE_CONCURRENCY slots +# CONTRIBUTOR_EVIDENCE_BATCH_SIZE=150 # logins per build-contributor-evidence job; the scheduled run fans out into +# # per-batch jobs above this so the per-login GitHub reads spread across the +# # queue instead of bursting. Set 0 to disable the fan-out (single job). # --- Caddy HTTPS terminator (#1203; requires --profile caddy) --- # DOMAIN=gittensory.example.com # fully-qualified domain; Caddy auto-obtains a Let's Encrypt cert diff --git a/src/queue/processors.ts b/src/queue/processors.ts index a4f84e0ca3..e28c8723ed 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -781,7 +781,7 @@ export async function processJob(env: Env, message: JobMessage): Promise { await fileUpstreamDriftIssues(env); return; case "build-contributor-evidence": - await buildContributorEvidence(env, message.login); + await buildContributorEvidence(env, message.login, message.logins); return; case "build-contributor-decision-packs": await buildContributorDecisionPacks(env, message.login); @@ -2540,10 +2540,70 @@ async function loadContributorPullRequestFilePaths( return files; } +const CONTRIBUTOR_EVIDENCE_LOGIN_CAP = 500; +const DEFAULT_CONTRIBUTOR_EVIDENCE_BATCH_SIZE = 150; + +// Max logins processed per build-contributor-evidence job before the scheduled trigger fans out into per-batch jobs. +// 0 disables the fan-out (single job). Read from process.env so it works on cloud + self-host without a binding. +export function contributorEvidenceBatchSize(): number { + const raw = Number(process.env.CONTRIBUTOR_EVIDENCE_BATCH_SIZE ?? String(DEFAULT_CONTRIBUTOR_EVIDENCE_BATCH_SIZE)); + return Number.isFinite(raw) && raw >= 0 ? Math.floor(raw) : DEFAULT_CONTRIBUTOR_EVIDENCE_BATCH_SIZE; +} + async function buildContributorEvidence( env: Env, login?: string, + batchLogins?: string[], +): Promise { + // A single login or a fanned-out batch → process exactly those (no derivation). + const explicitLogins = batchLogins?.length ? batchLogins : login ? [login] : null; + if (explicitLogins) { + await processContributorEvidenceLogins(env, explicitLogins); + return; + } + // Scheduled trigger: derive the full contributor set from stored PRs + issues. + const [allPullRequests, allIssues] = await Promise.all([ + listAllPullRequests(env), + listAllIssues(env), + ]); + const derivedLogins = [ + ...new Set( + [...allPullRequests, ...allIssues].flatMap((record) => + record.authorLogin ? [record.authorLogin] : [], + ), + ), + ].slice(0, CONTRIBUTOR_EVIDENCE_LOGIN_CAP); + const batchSize = contributorEvidenceBatchSize(); + // Fan out into per-batch jobs so the per-login GitHub reads (/users/{login} + its repos pages) spread across the + // queue's paced execution + rate-limit admission instead of bursting for every contributor in one job. Stays one + // job when the set fits a batch or the fan-out is disabled (CONTRIBUTOR_EVIDENCE_BATCH_SIZE=0). + if (batchSize > 0 && derivedLogins.length > batchSize) { + const batches: string[][] = []; + for (let i = 0; i < derivedLogins.length; i += batchSize) { + batches.push(derivedLogins.slice(i, i + batchSize)); + } + await Promise.all( + batches.map((batch, index) => { + const message: JobMessage = { type: "build-contributor-evidence", requestedBy: "schedule", logins: batch }; + const delaySeconds = Math.min(index * 15, 600); + return delaySeconds > 0 ? env.JOBS.send(message, { delaySeconds }) : env.JOBS.send(message); + }), + ); + return; + } + // Small enough (or fan-out disabled): process inline, reusing the PRs + issues loaded above. + await processContributorEvidenceLogins(env, derivedLogins, { allPullRequests, allIssues }); +} + +async function processContributorEvidenceLogins( + env: Env, + logins: string[], + preloaded?: { + allPullRequests: Awaited>; + allIssues: Awaited>; + }, ): Promise { + if (logins.length === 0) return; const [ allPullRequests, allIssues, @@ -2552,22 +2612,13 @@ async function buildContributorEvidence( allBounties, snapshot, ] = await Promise.all([ - listAllPullRequests(env), - listAllIssues(env), + preloaded ? Promise.resolve(preloaded.allPullRequests) : listAllPullRequests(env), + preloaded ? Promise.resolve(preloaded.allIssues) : listAllIssues(env), listRepositories(env), listRepoSyncStates(env), listBounties(env), getOrCreateScoringModelSnapshot(env), ]); - const logins = login - ? [login] - : [ - ...new Set( - [...allPullRequests, ...allIssues].flatMap((record) => - record.authorLogin ? [record.authorLogin] : [], - ), - ), - ].slice(0, 500); const issueQualityByRepo = await loadIssueQualityReportMap(env, repositories); for (const contributorLogin of logins) { // Isolate each login so one failure (transient GitHub/D1 error) doesn't abort the whole diff --git a/src/selfhost/queue-common.ts b/src/selfhost/queue-common.ts index b03e204655..840e293553 100644 --- a/src/selfhost/queue-common.ts +++ b/src/selfhost/queue-common.ts @@ -602,6 +602,7 @@ export function jobCoalesceKey(payload: string): string | null { deliveryId?: unknown; draftId?: unknown; event?: { dedupKey?: unknown } | null; + logins?: unknown; payload?: GitHubWebhookPayload | null; }; const type = typeof message.type === "string" ? message.type : ""; @@ -663,8 +664,20 @@ export function jobCoalesceKey(payload: string): string | null { case "build-burden-forecasts": return keyOf(type, normalizedRepo(message.repoFullName) ?? "all"); case "build-contributor-evidence": - case "build-contributor-decision-packs": - return keyOf(type, normalizedLogin(message.login) ?? "all"); + case "build-contributor-decision-packs": { + const login = normalizedLogin(message.login); + if (login) return keyOf(type, login); + // A fanned-out batch (a non-empty `logins` array) keys by its FIRST login: batches are disjoint slices of the + // derived set, so heads are unique and a duplicate re-enqueue of the same batch still coalesces. A batch must + // NEVER fall through to the "all" key below — that is the scheduled TRIGGER's slot, so collapsing a batch into + // it would drop the batch's work — so a batch with no usable head is left uncoalesced (null) instead. + if (Array.isArray(message.logins) && message.logins.length > 0) { + const batchHead = normalizedLogin(message.logins[0]); + return batchHead ? keyOf(type, "batch", batchHead) : null; + } + // The scheduled trigger (no login, no batch) coalesces to a single slot. + return keyOf(type, "all"); + } case "refresh-contributor-activity": return keyOf( type, diff --git a/src/types.ts b/src/types.ts index 030dc0c013..fee2d4e0b8 100644 --- a/src/types.ts +++ b/src/types.ts @@ -100,6 +100,9 @@ export type JobMessage = type: "build-contributor-evidence"; requestedBy: "schedule" | "api" | "test"; login?: string; + // A batch of logins to process in ONE job. Set by the cron fan-out (when the derived login set exceeds + // CONTRIBUTOR_EVIDENCE_BATCH_SIZE) so the per-login GitHub reads spread across the queue instead of bursting. + logins?: string[]; } | { type: "build-contributor-decision-packs"; diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index 71201bf244..0dfecb0047 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -44,7 +44,7 @@ import { upsertRepositoryFromGitHub, putCachedAiReview, } from "../../src/db/repositories"; -import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, processJob } from "../../src/queue/processors"; +import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, contributorEvidenceBatchSize, processJob } from "../../src/queue/processors"; import { upsertRepoFocusManifest } from "../../src/signals/focus-manifest-loader"; import { normalizeRegistryPayload } from "../../src/registry/normalize"; import { persistRegistrySnapshot } from "../../src/registry/sync"; @@ -100,6 +100,67 @@ describe("queue processors", () => { afterEach(() => { vi.useRealTimers(); vi.unstubAllGlobals(); + vi.unstubAllEnvs(); + }); + + it("fans build-contributor-evidence out into per-batch jobs when the login set exceeds CONTRIBUTOR_EVIDENCE_BATCH_SIZE (#1941)", async () => { + vi.stubEnv("CONTRIBUTOR_EVIDENCE_BATCH_SIZE", "1"); // force a fan-out at > 1 derived login + const env = createTestEnv(); + // Two contributors via stored PRs with distinct authors → a derived login set of 2. + await upsertPullRequestFromGitHub(env, "owner/repo", { number: 1, title: "PR one", state: "open", user: { login: "alice" }, head: { sha: "a1" }, labels: [], body: "x" }); + await upsertPullRequestFromGitHub(env, "owner/repo", { number: 2, title: "PR two", state: "open", user: { login: "bob" }, head: { sha: "b2" }, labels: [], body: "y" }); + + const fanned: import("../../src/types").JobMessage[] = []; + const send = env.JOBS.send.bind(env.JOBS); + env.JOBS.send = (async (message: import("../../src/types").JobMessage, options?: QueueSendOptions) => { + if (message.type === "build-contributor-evidence") fanned.push(message); + return send(message, options); + }) as typeof env.JOBS.send; + + await processJob(env, { type: "build-contributor-evidence", requestedBy: "schedule" }); + env.JOBS.send = send; + + // The scheduled trigger fanned out into one per-batch job per login (batch size 1), each carrying a `logins` + // array — not a single giant inline job. + expect(fanned).toHaveLength(2); + const batched = fanned.flatMap((m) => (m as { logins?: string[] }).logins ?? []).sort(); + expect(batched).toEqual(["alice", "bob"]); + expect(fanned.every((m) => Array.isArray((m as { logins?: string[] }).logins))).toBe(true); + }); + + it("reads CONTRIBUTOR_EVIDENCE_BATCH_SIZE, defaulting on unset / invalid / negative values", () => { + expect(contributorEvidenceBatchSize()).toBe(150); // unset → default + vi.stubEnv("CONTRIBUTOR_EVIDENCE_BATCH_SIZE", "40"); + expect(contributorEvidenceBatchSize()).toBe(40); + vi.stubEnv("CONTRIBUTOR_EVIDENCE_BATCH_SIZE", "0"); + expect(contributorEvidenceBatchSize()).toBe(0); // 0 = disable fan-out + vi.stubEnv("CONTRIBUTOR_EVIDENCE_BATCH_SIZE", "-5"); + expect(contributorEvidenceBatchSize()).toBe(150); // negative → default + vi.stubEnv("CONTRIBUTOR_EVIDENCE_BATCH_SIZE", "not-a-number"); + expect(contributorEvidenceBatchSize()).toBe(150); // NaN → default + }); + + it("does NOT fan out when batching is disabled (CONTRIBUTOR_EVIDENCE_BATCH_SIZE=0) — the scheduled trigger stays one job (#1941)", async () => { + vi.stubEnv("CONTRIBUTOR_EVIDENCE_BATCH_SIZE", "0"); + vi.stubGlobal("fetch", async () => Response.json({})); // inline path makes per-login + scoring reads; keep off-network + const env = createTestEnv(); + await upsertPullRequestFromGitHub(env, "owner/repo", { number: 1, title: "PR one", state: "open", user: { login: "alice" }, head: { sha: "a1" }, labels: [], body: "x" }); + await upsertPullRequestFromGitHub(env, "owner/repo", { number: 2, title: "PR two", state: "open", user: { login: "bob" }, head: { sha: "b2" }, labels: [], body: "y" }); + const batches: import("../../src/types").JobMessage[] = []; + const send = env.JOBS.send.bind(env.JOBS); + env.JOBS.send = (async (message: import("../../src/types").JobMessage, options?: QueueSendOptions) => { + if (message.type === "build-contributor-evidence" && Array.isArray((message as { logins?: string[] }).logins)) batches.push(message); + return send(message, options); + }) as typeof env.JOBS.send; + await processJob(env, { type: "build-contributor-evidence", requestedBy: "schedule" }); + env.JOBS.send = send; + expect(batches).toHaveLength(0); // never fanned out — processed inline + }); + + it("build-contributor-evidence is a no-op when there are no contributors (empty derived set) (#1941)", async () => { + const env = createTestEnv(); + // No PRs/issues → no derived logins → the worker early-returns before loading aggregate data or making any read. + await expect(processJob(env, { type: "build-contributor-evidence", requestedBy: "schedule" })).resolves.toBeUndefined(); }); it("processes registry, backfill, installation health, and signal snapshot jobs", async () => { diff --git a/test/unit/selfhost-queue-common.test.ts b/test/unit/selfhost-queue-common.test.ts index d673117198..80c3d63a35 100644 --- a/test/unit/selfhost-queue-common.test.ts +++ b/test/unit/selfhost-queue-common.test.ts @@ -676,6 +676,23 @@ describe("self-host queue common helpers", () => { ); }); + it("keys build-contributor-evidence by login/all, and fanned-out batches by their FIRST login (never one shared key) (#1941)", () => { + // A single-login (re-index) job coalesces by login; the scheduled trigger (no login/logins) → the "all" slot. + expect(jobCoalesceKey(payload({ type: "build-contributor-evidence", requestedBy: "schedule", login: "Alice" }))).toBe("build-contributor-evidence:alice"); + expect(jobCoalesceKey(payload({ type: "build-contributor-evidence", requestedBy: "schedule" }))).toBe("build-contributor-evidence:all"); + // Fanned-out batches key by their FIRST login → DISTINCT batches get DISTINCT keys (none is dropped by coalescing). + const batchA = jobCoalesceKey(payload({ type: "build-contributor-evidence", requestedBy: "schedule", logins: ["Bob", "Carol"] })); + const batchB = jobCoalesceKey(payload({ type: "build-contributor-evidence", requestedBy: "schedule", logins: ["Dave", "Erin"] })); + expect(batchA).toBe("build-contributor-evidence:batch:bob"); + expect(batchB).toBe("build-contributor-evidence:batch:dave"); + expect(batchA).not.toBe(batchB); + expect(batchA).not.toBe("build-contributor-evidence:all"); // the footgun: a batch must never collapse into "all" + // An EMPTY batch (no logins) is the scheduled-trigger shape → the "all" slot. + expect(jobCoalesceKey(payload({ type: "build-contributor-evidence", requestedBy: "schedule", logins: [] }))).toBe("build-contributor-evidence:all"); + // A non-empty batch whose first login is unusable is left UNCOALESCED (null) — never collapsed into "all". + expect(jobCoalesceKey(payload({ type: "build-contributor-evidence", requestedBy: "schedule", logins: [""] }))).toBeNull(); + }); + it("returns no coalesce key for malformed payloads", () => { expect(jobCoalesceKey("not-json")).toBeNull(); }); From f6c162f8c29af32cb4c8cf1c65223f440630fad3 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Wed, 1 Jul 2026 01:37:18 -0700 Subject: [PATCH 2/2] test(queue): cover the fanned-batch and null-author arms of build-contributor-evidence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The evidence fan-out landed at 93.75% patch coverage with two partial branches: the explicit-`logins` batch arm (a fanned job was created but never processed) and the `record.authorLogin ? … : []` filter's empty arm (no fixture record lacked an author). Add two invariant tests: a fanned batch job processes exactly its `logins` without re-deriving or re-fanning, and a null-author (ghost/deleted account) record is filtered out of the derived contributor set. Diff-range branch coverage is now complete. --- test/unit/queue.test.ts | 41 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index 0dfecb0047..0bc7e3b884 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -163,6 +163,47 @@ describe("queue processors", () => { await expect(processJob(env, { type: "build-contributor-evidence", requestedBy: "schedule" })).resolves.toBeUndefined(); }); + it("a fanned-out batch job (explicit `logins`) processes exactly those logins — never re-derives or re-fans (#1941)", async () => { + vi.stubEnv("CONTRIBUTOR_EVIDENCE_BATCH_SIZE", "1"); + vi.stubGlobal("fetch", async () => Response.json({})); // per-login reads stay off-network + const env = createTestEnv(); + // Stored PRs from OTHER authors — an explicit batch must ignore them (no derivation from stored records). + await upsertPullRequestFromGitHub(env, "owner/repo", { number: 1, title: "PR one", state: "open", user: { login: "alice" }, head: { sha: "a1" }, labels: [], body: "x" }); + await upsertPullRequestFromGitHub(env, "owner/repo", { number: 2, title: "PR two", state: "open", user: { login: "bob" }, head: { sha: "b2" }, labels: [], body: "y" }); + const refanned: import("../../src/types").JobMessage[] = []; + const send = env.JOBS.send.bind(env.JOBS); + env.JOBS.send = (async (message: import("../../src/types").JobMessage, options?: QueueSendOptions) => { + if (message.type === "build-contributor-evidence") refanned.push(message); + return send(message, options); + }) as typeof env.JOBS.send; + // A batch carrying an explicit `logins` array processes exactly that set (even a login with no stored PRs)... + await expect( + processJob(env, { type: "build-contributor-evidence", requestedBy: "schedule", logins: ["carol"] }), + ).resolves.toBeUndefined(); + env.JOBS.send = send; + // ...and short-circuits BEFORE the fan-out: it never re-derives from stored PRs nor re-enqueues evidence jobs. + expect(refanned).toHaveLength(0); + }); + + it("derives only records that have an author — a null-author (ghost/deleted account) record contributes nothing (#1941)", async () => { + vi.stubEnv("CONTRIBUTOR_EVIDENCE_BATCH_SIZE", "1"); // force a fan-out so the derived set is observable via the batch jobs + const env = createTestEnv(); + // Two real authors + a ghost issue with no `user` (deleted account) → the ghost must NOT become a derived login. + await upsertPullRequestFromGitHub(env, "owner/repo", { number: 1, title: "PR one", state: "open", user: { login: "alice" }, head: { sha: "a1" }, labels: [], body: "x" }); + await upsertPullRequestFromGitHub(env, "owner/repo", { number: 2, title: "PR two", state: "open", user: { login: "bob" }, head: { sha: "b2" }, labels: [], body: "y" }); + await upsertIssueFromGitHub(env, "owner/repo", { number: 9, title: "ghost issue", state: "open", labels: [], body: "z" }); // no user → null authorLogin + const fanned: import("../../src/types").JobMessage[] = []; + const send = env.JOBS.send.bind(env.JOBS); + env.JOBS.send = (async (message: import("../../src/types").JobMessage, options?: QueueSendOptions) => { + if (message.type === "build-contributor-evidence") fanned.push(message); + return send(message, options); + }) as typeof env.JOBS.send; + await processJob(env, { type: "build-contributor-evidence", requestedBy: "schedule" }); + env.JOBS.send = send; + const derived = fanned.flatMap((m) => (m as { logins?: string[] }).logins ?? []).sort(); + expect(derived).toEqual(["alice", "bob"]); // only the real authors; the null-author issue is filtered out + }); + it("processes registry, backfill, installation health, and signal snapshot jobs", async () => { const env = createTestEnv({ GITHUB_PUBLIC_TOKEN: "public-token" }); vi.stubGlobal("fetch", async (input: RequestInfo | URL) => {