From 42c03d45187d07c929ff76dcaf427ba528f3847c Mon Sep 17 00:00:00 2001 From: philluiz2323 Date: Fri, 24 Jul 2026 18:27:38 +0400 Subject: [PATCH] fix(queue): isolate per-repo failures in backfill-registered-repos fan-out (#8355) The cron-scheduled full-fleet sweep fanned out one per-repo job via a bare Promise.all -- a single transient env.JOBS.send rejection aborted the whole fan-out, and the queue's retry then re-ran the entire repositories map from scratch, duplicate-dispatching every repo whose send had already succeeded before the failure. Switches to Promise.allSettled so every repo's send is attempted exactly once regardless of an earlier one's outcome, logs each individual failure (repoFullName + reason, matching this file's existing structured-logging convention), and still throws after the settle so the cron invocation is marked failed for observability -- but only once every send has been attempted. --- src/queue/job-dispatch.ts | 19 +++++++++- test/unit/job-dispatch.test.ts | 65 ++++++++++++++++++++++++++++++++++ 2 files changed, 83 insertions(+), 1 deletion(-) diff --git a/src/queue/job-dispatch.ts b/src/queue/job-dispatch.ts index eaffbfaf37..96435c281a 100644 --- a/src/queue/job-dispatch.ts +++ b/src/queue/job-dispatch.ts @@ -107,7 +107,7 @@ export async function processJob(env: Env, message: JobMessage): Promise { if (repositories.length > 0) { const delayStepSeconds = message.mode === "full" || message.mode === "resume" ? 45 : 15; - await Promise.all( + const settled = await Promise.allSettled( repositories.map((repo, index) => { const repoMessage: JobMessage = { type: "backfill-registered-repos", @@ -124,6 +124,23 @@ export async function processJob(env: Env, message: JobMessage): Promise { : env.JOBS.send(repoMessage); }), ); + // #8355: Promise.allSettled (not Promise.all) so one repo's send failure never blocks or duplicates + // a sibling repo's send -- a bare Promise.all rejecting mid-fan-out previously caused the queue's own + // retry to re-run this ENTIRE repositories.map from scratch, duplicate-dispatching every repo whose + // send had already succeeded before the failure. Every repo's send is attempted exactly once + // regardless of an earlier one's outcome; a genuine failure still surfaces below (after every send + // has been attempted) so the cron invocation itself is marked failed for observability. + const failedRepoFullNames: string[] = []; + settled.forEach((result, index) => { + if (result.status === "rejected") { + const repoFullName = repositories[index]!.fullName; + failedRepoFullNames.push(repoFullName); + console.error(JSON.stringify({ level: "error", event: "backfill_registered_repos_fanout_send_failed", repoFullName, reason: String(result.reason) })); + } + }); + if (failedRepoFullNames.length > 0) { + throw new Error(`backfill-registered-repos fan-out: ${failedRepoFullNames.length}/${repositories.length} repo send(s) failed: ${failedRepoFullNames.join(", ")}`); + } return; } } diff --git a/test/unit/job-dispatch.test.ts b/test/unit/job-dispatch.test.ts index fa3fafc0b2..27b6962c10 100644 --- a/test/unit/job-dispatch.test.ts +++ b/test/unit/job-dispatch.test.ts @@ -2,6 +2,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { processJob } from "../../src/queue/job-dispatch"; import { createTestEnv } from "../helpers/d1"; +import { upsertInstallation, upsertRepositoryFromGitHub } from "../../src/db/repositories"; import type { JobMessage } from "../../src/types"; describe("processJob unknown job type (#5836)", () => { @@ -39,3 +40,67 @@ describe("processJob unknown job type (#5836)", () => { expect(warnLogs.some((line) => line.includes("unknown_job_type_ignored"))).toBe(false); }); }); + +describe("processJob backfill-registered-repos fan-out isolation (#8355)", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it("attempts every OTHER repo's send even when one repo's send rejects, and throws after the settle", async () => { + const sentTo: string[] = []; + const env = createTestEnv({ + JOBS: { + async send(message: unknown) { + const repoFullName = (message as { repoFullName?: string }).repoFullName; + if (repoFullName) sentTo.push(repoFullName); + if (repoFullName === "owner/fails") throw new Error("simulated transient queue-send failure"); + return undefined; + }, + } as unknown as Queue, + }); + await upsertInstallation(env, { action: "created", installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); + await upsertRepositoryFromGitHub(env, { name: "ok-1", full_name: "owner/ok-1", private: false, owner: { login: "owner" } }, 9001); + await upsertRepositoryFromGitHub(env, { name: "fails", full_name: "owner/fails", private: false, owner: { login: "owner" } }, 9001); + await upsertRepositoryFromGitHub(env, { name: "ok-2", full_name: "owner/ok-2", private: false, owner: { login: "owner" } }, 9001); + + const errorLogs: string[] = []; + vi.spyOn(console, "error").mockImplementation((...args: unknown[]) => { + errorLogs.push(String(args[0])); + }); + + await expect(processJob(env, { type: "backfill-registered-repos", requestedBy: "schedule" } as JobMessage)).rejects.toThrow( + /backfill-registered-repos fan-out: 1\/3 repo send\(s\) failed: owner\/fails/, + ); + + // Every repo was attempted exactly once, regardless of the middle one's rejection. + expect(sentTo.sort()).toEqual(["owner/fails", "owner/ok-1", "owner/ok-2"]); + + const failureLog = errorLogs.map((line) => JSON.parse(line) as Record).find((log) => log.event === "backfill_registered_repos_fanout_send_failed"); + expect(failureLog).toMatchObject({ level: "error", event: "backfill_registered_repos_fanout_send_failed", repoFullName: "owner/fails" }); + }); + + it("does not throw or log a failure when every repo's send succeeds", async () => { + const sentTo: string[] = []; + const env = createTestEnv({ + JOBS: { + async send(message: unknown) { + const repoFullName = (message as { repoFullName?: string }).repoFullName; + if (repoFullName) sentTo.push(repoFullName); + return undefined; + }, + } as unknown as Queue, + }); + await upsertInstallation(env, { action: "created", installation: { id: 9002, account: { login: "owner2", id: 2, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); + await upsertRepositoryFromGitHub(env, { name: "ok-1", full_name: "owner2/ok-1", private: false, owner: { login: "owner2" } }, 9002); + await upsertRepositoryFromGitHub(env, { name: "ok-2", full_name: "owner2/ok-2", private: false, owner: { login: "owner2" } }, 9002); + + const errorLogs: string[] = []; + vi.spyOn(console, "error").mockImplementation((...args: unknown[]) => { + errorLogs.push(String(args[0])); + }); + + await expect(processJob(env, { type: "backfill-registered-repos", requestedBy: "schedule" } as JobMessage)).resolves.toBeUndefined(); + expect(sentTo.sort()).toEqual(["owner2/ok-1", "owner2/ok-2"]); + expect(errorLogs.some((line) => line.includes("backfill_registered_repos_fanout_send_failed"))).toBe(false); + }); +});