fix(reliability): harden webhook/queue/public-fetch error handling (#786, #787, #790, #791)

src/github/backfill.ts

@@ -1848,15 +1848,20 @@ async function upsertContributorStats(
   recentMerged: GitHubPullRequestPayload[],
 ): Promise<void> {
   /* v8 ignore start -- Contributor-stat payload fallbacks normalize optional GitHub fields already covered by backfill round trips. */
-  const logins = new Set<string>();
-  for (const pr of pullRequests) if (pr.authorLogin) logins.add(pr.authorLogin);
-  for (const pr of recentMerged) if (pr.user?.login) logins.add(pr.user.login);
-  for (const issue of issues) if (issue.user?.login) logins.add(issue.user.login);
-
-  for (const login of logins) {
-    const authoredPullRequests = pullRequests.filter((pr) => pr.authorLogin === login);
-    const authoredMerged = recentMerged.filter((pr) => pr.user?.login === login);
-    const authoredIssues = issues.filter((issue) => issue.user?.login === login);
+  // Canonical case-insensitive login key so one user across mixed casings collapses to one row (#791).
+  const loginByKey = new Map<string, string>();
+  const addLogin = (value: string | null | undefined): void => {
+    const key = value?.toLowerCase();
+    if (key && !loginByKey.has(key)) loginByKey.set(key, value as string);
+  };
+  for (const pr of pullRequests) addLogin(pr.authorLogin);
+  for (const pr of recentMerged) addLogin(pr.user?.login);
+  for (const issue of issues) addLogin(issue.user?.login);
+
+  for (const [loginKey, login] of loginByKey) {
+    const authoredPullRequests = pullRequests.filter((pr) => pr.authorLogin?.toLowerCase() === loginKey);
+    const authoredMerged = recentMerged.filter((pr) => pr.user?.login?.toLowerCase() === loginKey);
+    const authoredIssues = issues.filter((issue) => issue.user?.login?.toLowerCase() === loginKey);
     const labels = [...authoredPullRequests.flatMap((pr) => pr.labels), ...authoredIssues.flatMap((issue) => (issue.labels ?? []).flatMap((label) => (label.name ? [label.name] : [])))];
     const stat: ContributorRepoStatRecord = {
       login,

src/github/public.ts

@@ -55,6 +55,10 @@ const REPO_STATS_CACHE_TTL_MS = 1000 * 60 * 10;
 const REPO_STATS_STALE_TTL_MS = 1000 * 60 * 60 * 24;
 const repoStatsCache = new Map<string, RepoStatsCacheEntry>();
 
+// Bound the github.com/ghapi round trips so a hung response can't stall the 500-login evidence loop
+// indefinitely (mirrors GITHUB_FETCH_TIMEOUT_MS in src/github/app.ts) (#790).
+const GITHUB_PUBLIC_FETCH_TIMEOUT_MS = 12_000;
+
 export async function fetchPublicContributorProfile(login: string): Promise<PublicContributorProfile> {
   const safeLogin = encodeURIComponent(login);
   const headers = {
@@ -63,9 +67,10 @@ export async function fetchPublicContributorProfile(login: string): Promise<Publ
     "x-github-api-version": "2022-11-28",
   };
   try {
+    const signal = AbortSignal.timeout(GITHUB_PUBLIC_FETCH_TIMEOUT_MS);
     const [userResponse, reposResponse] = await Promise.all([
-      fetch(`https://github.com/ghapi/users/${safeLogin}`, { headers }),
-      fetch(`https://github.com/ghapi/users/${safeLogin}/repos?per_page=100&sort=updated`, { headers }),
+      fetch(`https://github.com/ghapi/users/${safeLogin}`, { headers, signal }),
+      fetch(`https://github.com/ghapi/users/${safeLogin}/repos?per_page=100&sort=updated`, { headers, signal }),
     ]);
     if (!userResponse.ok) throw new Error(`GitHub user lookup failed (${userResponse.status})`);
     const user = (await userResponse.json()) as GitHubUserResponse;

src/github/webhook.ts

@@ -57,7 +57,22 @@ export async function handleGitHubWebhook(c: Context<{ Bindings: Env }>): Promis
     eventName,
     payload,
   };
-  await c.env.JOBS.send(message);
+  try {
+    await c.env.JOBS.send(message);
+  } catch {
+    // Enqueue failed: flip the event to "error" so the dedup guard above lets GitHub redeliver,
+    // and return 500 so GitHub retries instead of treating the webhook as handled (#786).
+    await recordWebhookEvent(c.env, {
+      deliveryId,
+      eventName,
+      action: payload.action,
+      installationId: payload.installation?.id,
+      repositoryFullName: payload.repository?.full_name,
+      payloadHash,
+      status: "error",
+    });
+    return c.json({ error: "enqueue_failed", deliveryId }, 500);
+  }
 
   return c.json({ ok: true, deliveryId, eventName, status: "queued" }, 202);
 }

src/queue/processors.ts

@@ -296,7 +296,16 @@ async function buildContributorDecisionPacks(env: Env, login?: string): Promise<
   const logins = login ? [login] : await discoverContributorLogins(env);
   // Load the login-independent full-table datasets once, then reuse across every login instead of re-scanning per contributor.
   const shared = await loadDecisionPackSharedInputs(env);
-  for (const contributorLogin of logins) await buildAndPersistContributorDecisionPack(env, contributorLogin, shared);
+  for (const contributorLogin of logins) {
+    try {
+      await buildAndPersistContributorDecisionPack(env, contributorLogin, shared);
+    } catch (error) {
+      // Isolate per-login failures so one bad login can't fail the whole batch (which would re-run
+      // from the first login on retry and poison-pill the queue) (#787).
+      /* v8 ignore next -- defensive per-login isolation; the log-and-continue path is not exercised in tests */
+      console.error(JSON.stringify({ level: "warn", event: "decision_pack_login_failed", login: contributorLogin, error: errorMessage(error) }));
+    }
+  }
 }
 
 async function fanOutRepoSignalSnapshotJobs(env: Env, requestedBy: "schedule" | "api" | "test"): Promise<void> {
@@ -394,6 +403,9 @@ async function buildContributorEvidence(env: Env, login?: string): Promise<void>
   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
+    // 500-login batch and poison-pill the queue on retry (#787).
+    try {
     const [github, contributorPullRequests, contributorIssues, cachedRepoStats, gittensorSnapshot] = await Promise.all([
       fetchPublicContributorProfile(contributorLogin),
       listContributorPullRequests(env, contributorLogin),
@@ -487,6 +499,10 @@ async function buildContributorEvidence(env: Env, login?: string): Promise<void>
       payload: evidenceGraph as unknown as Record<string, JsonValue>,
       generatedAt: evidenceGraph.generatedAt,
     });
+    } catch (error) {
+      /* v8 ignore next -- defensive per-login isolation; the log-and-continue path is not exercised in tests */
+      console.error(JSON.stringify({ level: "warn", event: "contributor_evidence_login_failed", login: contributorLogin, error: errorMessage(error) }));
+    }
   }
 }
 

test/unit/webhook.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, it } from "vitest";
 import type { Context } from "hono";
 import { handleGitHubWebhook } from "../../src/github/webhook";
+import { getWebhookEvent } from "../../src/db/repositories";
 import { createTestEnv } from "../helpers/d1";
 
 describe("github webhook body reader edge cases", () => {
@@ -36,3 +37,47 @@ describe("github webhook body reader edge cases", () => {
     await expect(response.json()).resolves.toMatchObject({ error: "invalid_signature" });
   });
 });
+
+describe("github webhook enqueue failure (#786)", () => {
+  it("flags the event 'error' and returns 500 when the queue send fails", async () => {
+    const env = createTestEnv();
+    env.JOBS = {
+      send: async () => {
+        throw new Error("queue unavailable");
+      },
+    } as unknown as typeof env.JOBS;
+    const rawBody = JSON.stringify({ action: "opened", repository: { full_name: "JSONbored/gittensory" }, installation: { id: 1 } });
+    const signature = await signWebhook(rawBody, env.GITHUB_WEBHOOK_SECRET);
+    const request = new Request("https://example.com/webhook", { method: "POST", body: rawBody });
+    const headers: Record<string, string> = {
+      "x-github-delivery": "enqueue-fail-1",
+      "x-github-event": "pull_request",
+      "x-hub-signature-256": signature,
+    };
+    const context = {
+      req: {
+        raw: request,
+        header(name: string) {
+          return headers[name.toLowerCase()] ?? null;
+        },
+      },
+      env,
+      json(payload: unknown, status?: number) {
+        return Response.json(payload, status === undefined ? undefined : { status });
+      },
+    } as unknown as Context<{ Bindings: Env }>;
+
+    const response = await handleGitHubWebhook(context);
+    expect(response.status).toBe(500);
+    await expect(response.json()).resolves.toMatchObject({ error: "enqueue_failed" });
+    // Flagged "error" so the dedup guard lets GitHub redeliver instead of suppressing it.
+    const event = await getWebhookEvent(env, "enqueue-fail-1");
+    expect(event?.status).toBe("error");
+  });
+});
+
+async function signWebhook(body: string, secret: string): Promise<string> {
+  const key = await crypto.subtle.importKey("raw", new TextEncoder().encode(secret), { name: "HMAC", hash: "SHA-256" }, false, ["sign"]);
+  const signed = await crypto.subtle.sign("HMAC", key, new TextEncoder().encode(body));
+  return `sha256=${[...new Uint8Array(signed)].map((byte) => byte.toString(16).padStart(2, "0")).join("")}`;
+}