From 55d511a1a2a08a90fc869875236ee0d4dffaccdb Mon Sep 17 00:00:00 2001 From: Andriy Polanski Date: Mon, 27 Jul 2026 14:15:06 +0000 Subject: [PATCH 1/2] fix(queue): isolate per-repo failures in generateSignalSnapshots (#9293) Use Promise.allSettled so one repo's data-gathering or persist error no longer skips siblings in the same multi-repo invocation, then surface an aggregate error listing the failed repos (same shape as #8355). Co-authored-by: Cursor --- src/queue/signal-snapshot.ts | 370 ++++++++++++++++++--------------- test/unit/queue-trends.test.ts | 37 +++- 2 files changed, 237 insertions(+), 170 deletions(-) diff --git a/src/queue/signal-snapshot.ts b/src/queue/signal-snapshot.ts index 1a9c30ec4c..f0ad112012 100644 --- a/src/queue/signal-snapshot.ts +++ b/src/queue/signal-snapshot.ts @@ -62,176 +62,208 @@ export async function generateSignalSnapshots( (repo) => repo.isInstalled && (!repoFullName || repo.fullName === repoFullName), ); - for (const repo of repositories) { - const trendSince = new Date( - Date.now() - QUEUE_TREND_HISTORY_DAYS * 24 * 60 * 60 * 1000, - ).toISOString(); - const [ - issues, - pullRequests, - recentMergedPullRequests, - labels, - queueCounts, - bounties, - totalsHistory, - queueHealthHistory, - ] = await Promise.all([ - listIssueSignalSample(env, repo.fullName), - listOpenPullRequests(env, repo.fullName), - listRecentMergedPullRequests(env, repo.fullName), - listRepoLabels(env, repo.fullName), - loadOpenQueueCounts(env, repo.fullName), - listBountiesByRepo(env, repo.fullName), - listRepoGithubTotalsSnapshotHistory(env, repo.fullName, { - sinceIso: trendSince, - limit: 120, - }), - listSignalSnapshots(env, "queue-health", repo.fullName), - ]); - const collisions = buildCollisionReport( - repo.fullName, - issues, - pullRequests, - recentMergedPullRequests, - ); - const queueHealth = buildQueueHealth( - repo, - issues, - pullRequests, - collisions, - queueCounts, - ); - const configQuality = buildConfigQuality( - repo, - issues, - pullRequests, - repo.fullName, - ); - const labelAudit = buildLabelAudit( - repo, - labels, - issues, - pullRequests, - repo.fullName, - ); - const maintainerLane = buildMaintainerLaneReport( - repo, - issues, - pullRequests, - repo.fullName, - collisions, - queueCounts, - ); - const maintainerCutReadiness = buildMaintainerCutReadiness( - repo, - issues, - pullRequests, - repo.fullName, - queueCounts, - collisions, - ); - const contributorIntakeHealth = buildContributorIntakeHealth( - repo, - issues, - pullRequests, - repo.fullName, - collisions, - queueCounts, - ); - const issueQuality = buildIssueQualityReport( - repo, - issues, - pullRequests, - repo.fullName, - bounties, - collisions, - recentMergedPullRequests, - ); - await replaceCollisionEdges( - env, - repo.fullName, - buildCollisionEdges(collisions), - ); - const generatedAt = new Date().toISOString(); - await persistSignalSnapshot(env, { - id: crypto.randomUUID(), - signalType: "queue-health", - targetKey: repo.fullName, - repoFullName: repo.fullName, - payload: queueHealth as unknown as Record, - generatedAt, - }); - await upsertRepoQueueTrendSnapshot(env, { - repoFullName: repo.fullName, - payload: buildQueueTrendReport({ - repoFullName: repo.fullName, - totalsSnapshots: totalsHistory, - queueHealthSnapshots: queueHealthHistory, - currentQueueHealth: queueHealth, - generatedAt, - }) as unknown as Record, - generatedAt, - }); - await persistSignalSnapshot(env, { - id: crypto.randomUUID(), - signalType: "config-quality", - targetKey: repo.fullName, - repoFullName: repo.fullName, - payload: configQuality as unknown as Record, - generatedAt, - }); - await persistSignalSnapshot(env, { - id: crypto.randomUUID(), - signalType: "label-audit", - targetKey: repo.fullName, - repoFullName: repo.fullName, - payload: labelAudit as unknown as Record, - generatedAt, - }); - await persistSignalSnapshot(env, { - id: crypto.randomUUID(), - signalType: "maintainer-lane", - targetKey: repo.fullName, - repoFullName: repo.fullName, - payload: maintainerLane as unknown as Record, - generatedAt, - }); - await persistSignalSnapshot(env, { - id: crypto.randomUUID(), - signalType: "maintainer-cut-readiness", - targetKey: repo.fullName, - repoFullName: repo.fullName, - payload: maintainerCutReadiness as unknown as Record, - generatedAt, - }); - await persistSignalSnapshot(env, { - id: crypto.randomUUID(), - signalType: "contributor-intake-health", - targetKey: repo.fullName, - repoFullName: repo.fullName, - payload: contributorIntakeHealth as unknown as Record, - generatedAt, - }); - await persistSignalSnapshot(env, { - id: crypto.randomUUID(), - signalType: "issue-quality", - targetKey: repo.fullName, - repoFullName: repo.fullName, - payload: issueQuality as unknown as Record, - generatedAt, - }); - const repoOutcomePatterns = await computeRepoOutcomePatterns( - env, - repo.fullName, - repo, + // #9293: Promise.allSettled (not a bare sequential for-loop) so one repo's data-gathering or + // persist failure never silently skips every subsequent repo in the same multi-repo invocation — + // same isolation shape as job-dispatch.ts's "backfill-registered-repos" fan-out (#8355). Every + // repo is attempted exactly once; successes persist regardless of a sibling's outcome; failures + // are collected and rethrown as one aggregate error so the invocation stays observably failed. + const settled = await Promise.allSettled( + repositories.map((repo) => generateSignalSnapshotForRepo(env, repo)), + ); + const failedRepoFullNames: string[] = []; + settled.forEach((result, index) => { + if (result.status === "rejected") { + const failedRepoFullName = repositories[index]!.fullName; + failedRepoFullNames.push(failedRepoFullName); + console.error( + JSON.stringify({ + level: "error", + event: "generate_signal_snapshots_repo_failed", + repoFullName: failedRepoFullName, + reason: String(result.reason), + }), + ); + } + }); + if (failedRepoFullNames.length > 0) { + throw new Error( + `generate-signal-snapshots: ${failedRepoFullNames.length}/${repositories.length} repo(s) failed: ${failedRepoFullNames.join(", ")}`, ); - await persistSignalSnapshot(env, { - id: crypto.randomUUID(), - signalType: REPO_OUTCOME_PATTERNS_SIGNAL, - targetKey: repo.fullName, + } +} + +async function generateSignalSnapshotForRepo( + env: Env, + repo: Awaited>[number], +): Promise { + const trendSince = new Date( + Date.now() - QUEUE_TREND_HISTORY_DAYS * 24 * 60 * 60 * 1000, + ).toISOString(); + const [ + issues, + pullRequests, + recentMergedPullRequests, + labels, + queueCounts, + bounties, + totalsHistory, + queueHealthHistory, + ] = await Promise.all([ + listIssueSignalSample(env, repo.fullName), + listOpenPullRequests(env, repo.fullName), + listRecentMergedPullRequests(env, repo.fullName), + listRepoLabels(env, repo.fullName), + loadOpenQueueCounts(env, repo.fullName), + listBountiesByRepo(env, repo.fullName), + listRepoGithubTotalsSnapshotHistory(env, repo.fullName, { + sinceIso: trendSince, + limit: 120, + }), + listSignalSnapshots(env, "queue-health", repo.fullName), + ]); + const collisions = buildCollisionReport( + repo.fullName, + issues, + pullRequests, + recentMergedPullRequests, + ); + const queueHealth = buildQueueHealth( + repo, + issues, + pullRequests, + collisions, + queueCounts, + ); + const configQuality = buildConfigQuality( + repo, + issues, + pullRequests, + repo.fullName, + ); + const labelAudit = buildLabelAudit( + repo, + labels, + issues, + pullRequests, + repo.fullName, + ); + const maintainerLane = buildMaintainerLaneReport( + repo, + issues, + pullRequests, + repo.fullName, + collisions, + queueCounts, + ); + const maintainerCutReadiness = buildMaintainerCutReadiness( + repo, + issues, + pullRequests, + repo.fullName, + queueCounts, + collisions, + ); + const contributorIntakeHealth = buildContributorIntakeHealth( + repo, + issues, + pullRequests, + repo.fullName, + collisions, + queueCounts, + ); + const issueQuality = buildIssueQualityReport( + repo, + issues, + pullRequests, + repo.fullName, + bounties, + collisions, + recentMergedPullRequests, + ); + await replaceCollisionEdges( + env, + repo.fullName, + buildCollisionEdges(collisions), + ); + const generatedAt = new Date().toISOString(); + await persistSignalSnapshot(env, { + id: crypto.randomUUID(), + signalType: "queue-health", + targetKey: repo.fullName, + repoFullName: repo.fullName, + payload: queueHealth as unknown as Record, + generatedAt, + }); + await upsertRepoQueueTrendSnapshot(env, { + repoFullName: repo.fullName, + payload: buildQueueTrendReport({ repoFullName: repo.fullName, - payload: repoOutcomePatterns as unknown as Record, + totalsSnapshots: totalsHistory, + queueHealthSnapshots: queueHealthHistory, + currentQueueHealth: queueHealth, generatedAt, - }); - } + }) as unknown as Record, + generatedAt, + }); + await persistSignalSnapshot(env, { + id: crypto.randomUUID(), + signalType: "config-quality", + targetKey: repo.fullName, + repoFullName: repo.fullName, + payload: configQuality as unknown as Record, + generatedAt, + }); + await persistSignalSnapshot(env, { + id: crypto.randomUUID(), + signalType: "label-audit", + targetKey: repo.fullName, + repoFullName: repo.fullName, + payload: labelAudit as unknown as Record, + generatedAt, + }); + await persistSignalSnapshot(env, { + id: crypto.randomUUID(), + signalType: "maintainer-lane", + targetKey: repo.fullName, + repoFullName: repo.fullName, + payload: maintainerLane as unknown as Record, + generatedAt, + }); + await persistSignalSnapshot(env, { + id: crypto.randomUUID(), + signalType: "maintainer-cut-readiness", + targetKey: repo.fullName, + repoFullName: repo.fullName, + payload: maintainerCutReadiness as unknown as Record, + generatedAt, + }); + await persistSignalSnapshot(env, { + id: crypto.randomUUID(), + signalType: "contributor-intake-health", + targetKey: repo.fullName, + repoFullName: repo.fullName, + payload: contributorIntakeHealth as unknown as Record, + generatedAt, + }); + await persistSignalSnapshot(env, { + id: crypto.randomUUID(), + signalType: "issue-quality", + targetKey: repo.fullName, + repoFullName: repo.fullName, + payload: issueQuality as unknown as Record, + generatedAt, + }); + const repoOutcomePatterns = await computeRepoOutcomePatterns( + env, + repo.fullName, + repo, + ); + await persistSignalSnapshot(env, { + id: crypto.randomUUID(), + signalType: REPO_OUTCOME_PATTERNS_SIGNAL, + targetKey: repo.fullName, + repoFullName: repo.fullName, + payload: repoOutcomePatterns as unknown as Record, + generatedAt, + }); } diff --git a/test/unit/queue-trends.test.ts b/test/unit/queue-trends.test.ts index 4c678054f5..ff98221f58 100644 --- a/test/unit/queue-trends.test.ts +++ b/test/unit/queue-trends.test.ts @@ -1,4 +1,5 @@ -import { describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import * as repositoriesModule from "../../src/db/repositories"; import { getRepoQueueTrendSnapshot, persistRepoGithubTotalsSnapshot, persistSignalSnapshot, upsertPullRequestFromGitHub, upsertRepositoryFromGitHub } from "../../src/db/repositories"; import { generateSignalSnapshots } from "../../src/queue/processors"; import { buildQueueTrendReport, buildUnavailableQueueTrendReport, type QueueTrendReport } from "../../src/services/queue-trends"; @@ -6,6 +7,9 @@ import type { RepoGithubTotalsSnapshotRecord } from "../../src/types"; import { createTestEnv } from "../helpers/d1"; describe("queue trend windows", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); it("builds deterministic 7/14/30-day queue pressure and review velocity windows", () => { const report = buildQueueTrendReport({ repoFullName: "owner/repo", @@ -220,6 +224,37 @@ describe("queue trend windows", () => { await expect(getRepoQueueTrendSnapshot(env, "acme/registered-only")).resolves.toBeNull(); }); + + it("#9293: one repo's data-gathering failure still persists siblings and surfaces an aggregate error", async () => { + const env = createTestEnv(); + await upsertRepositoryFromGitHub( + env, + { name: "ok", full_name: "owner/ok", private: false, owner: { login: "owner" }, default_branch: "main" }, + 701, + ); + await upsertRepositoryFromGitHub( + env, + { name: "boom", full_name: "owner/boom", private: false, owner: { login: "owner" }, default_branch: "main" }, + 702, + ); + + const originalListIssueSignalSample = repositoriesModule.listIssueSignalSample; + vi.spyOn(repositoriesModule, "listIssueSignalSample").mockImplementation(async (listEnv, fullName) => { + if (fullName === "owner/boom") throw new Error("D1 boom for owner/boom"); + return originalListIssueSignalSample(listEnv, fullName); + }); + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => undefined); + + await expect(generateSignalSnapshots(env)).rejects.toThrow( + /generate-signal-snapshots: 1\/2 repo\(s\) failed: owner\/boom/, + ); + + await expect(getRepoQueueTrendSnapshot(env, "owner/ok")).resolves.not.toBeNull(); + await expect(getRepoQueueTrendSnapshot(env, "owner/boom")).resolves.toBeNull(); + expect(errorSpy).toHaveBeenCalledWith( + expect.stringMatching(/"event":"generate_signal_snapshots_repo_failed".*"repoFullName":"owner\/boom"/), + ); + }); }); function totals(daysAgo: number, values: { openIssues: number; openPrs: number; merged: number; closed: number }): RepoGithubTotalsSnapshotRecord { From 792a3d0a7ae791a8e92e23094563767aacbbd0d4 Mon Sep 17 00:00:00 2001 From: Andriy Polanski Date: Mon, 27 Jul 2026 14:32:45 +0000 Subject: [PATCH 2/2] fix --- test/unit/selfhost-pg-retention.test.ts | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/test/unit/selfhost-pg-retention.test.ts b/test/unit/selfhost-pg-retention.test.ts index 96c9742d70..b4abf10a0f 100644 --- a/test/unit/selfhost-pg-retention.test.ts +++ b/test/unit/selfhost-pg-retention.test.ts @@ -1,9 +1,10 @@ // Unit tests for data-retention pruning (src/db/retention.ts) against the Postgres backend (#977). Mocks // pg.Pool so no real DB is needed — real-Postgres integration coverage lives in test/integration/selfhost-pg.test.ts. -// retention.ts itself is unchanged: it still emits SQLite-dialect SQL (rowid, `?1`-style numbered -// placeholders); src/selfhost/pg-dialect.ts's translateSql() is what makes it Postgres-safe, same as every -// other query path on this backend. These tests exercise that translation end-to-end through the real -// pruneExpiredRecords()/processJob() call path, not just the dialect translator in isolation. +// retention.ts emits SQLite-dialect SQL; src/selfhost/pg-dialect.ts's translateSql() makes it Postgres-safe. +// After #9083, age-based prune deletes by a real indexable PK (`id` / `delivery_id`) rather than +// rowid/ctid; signal_snapshots dedupe still uses the rowid→ctid translation. These tests exercise that +// translation end-to-end through the real pruneExpiredRecords()/processJob() call path, not just the +// dialect translator in isolation. import { describe, expect, it, vi } from "vitest"; import type { Pool } from "pg"; import { createPgAdapter } from "../../src/selfhost/pg-adapter"; @@ -32,10 +33,14 @@ function makeRetentionPgPool(remaining: Record = {}): MockPgPool return { rows: [{ n: remaining[table] ?? 0 }], rowCount: 1 }; } - const deleteMatch = /^DELETE FROM (\w+) WHERE ctid IN \(SELECT ctid FROM \1 WHERE .*? LIMIT (\d+)\)$/i.exec(q); + // #9083 age-based prune: `DELETE ... WHERE IN (SELECT ... LIMIT n)` where is a real + // column (`id` / `delivery_id`) OR the legacy ctid fallback for tables without RETENTION_PK_COLUMN. + // signal_snapshots dedupe still uses the ctid form after rowid→ctid translation. + const deleteMatch = + /^DELETE FROM (\w+) WHERE (\w+) IN \(SELECT \2 FROM \1 WHERE .*? LIMIT (\d+)\)$/i.exec(q); if (deleteMatch) { const table = deleteMatch[1] as string; - const limit = Number(deleteMatch[2]); + const limit = Number(deleteMatch[3]); const have = remaining[table] ?? 0; const changes = Math.min(have, limit); remaining[table] = have - changes;