perf(queue): fan build-contributor-evidence out into per-batch jobs (config-driven)

.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

src/queue/processors.ts

@@ -781,7 +781,7 @@ export async function processJob(env: Env, message: JobMessage): Promise<void> {
       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<void> {
+  // 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<ReturnType<typeof listAllPullRequests>>;
+    allIssues: Awaited<ReturnType<typeof listAllIssues>>;
+  },
 ): Promise<void> {
+  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

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,

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

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,108 @@ 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("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 () => {

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