diff --git a/apps/hub/src/github/interleave.postgres.test.ts b/apps/hub/src/github/interleave.postgres.test.ts index 83d4fc57..15025a7b 100644 --- a/apps/hub/src/github/interleave.postgres.test.ts +++ b/apps/hub/src/github/interleave.postgres.test.ts @@ -11,6 +11,7 @@ import { createBacklogStart } from "./backlog-triage.js"; import { ListenerUnroutedError } from "./deployment.js"; import { createGithubPrTriage } from "./pr-triage.js"; import { DEFAULT_RECONCILE_POLICY } from "./reconcile-plan.js"; +import { failDelivery, queueHead } from "./triage-queue.js"; import { createTriageReconciler, createUnroutedTries } from "./triage-reconciler.js"; import { createTriageStateStore, type TriageStateStore } from "./triage-state-store.js"; import type { ObservedRuns } from "./manager-runs.js"; @@ -62,12 +63,12 @@ function route(store: TriageStateStore, deliver: Deliver) { }); } -function reconciler(store: TriageStateStore, heads: number[], deliver: Deliver, pack: CheckPack = PACK, checks: CheckRun[] = []) { +function reconciler(store: TriageStateStore, heads: number[], deliver: Deliver, pack: CheckPack = PACK, checks: CheckRun[] = [], updatedAt = LONG_AGO) { return createTriageReconciler({ source: "github", tenants: async () => [{ id: TENANT, domain: "tenant.example", config: CONFIG }], listener: async () => DEPLOYMENT, - openHeadsFor: async () => async () => heads.map((number) => ({ number, headSha: number === 8 ? "abc123" : `sha${number}`, updatedAt: LONG_AGO, draft: false })), + openHeadsFor: async () => async () => heads.map((number) => ({ number, headSha: number === 8 ? "abc123" : `sha${number}`, updatedAt, draft: false })), activity: { stale: () => true }, managerRuns: noRuns, headChecks: async () => checks, @@ -193,3 +194,28 @@ test("PostgreSQL: a maintainer's triage request while the listener is starting a expect(mailsFor8(mails)).toBe(1); }); }, 120_000); + +test("PostgreSQL: a webhook's head queued while a reconcile pass plans over it, whose mail then reaches no listener, goes out on the next pass", async () => { + await withStore(async (store) => { + const opened = { tenantId: TENANT, repo: REPO, number: 9, headSha: "sha9" }; + let interleaved = false; + const racing: TriageStateStore = { + load: (tenantId, repo) => store.load(tenantId, repo), + async save(tenantId, repo, rows, expected) { + if (!interleaved) { + interleaved = true; + await queueHead(store, opened, new Date(NOW.getTime() - 5), () => false); + } + return store.save(tenantId, repo, rows, expected); + }, + }; + await reconciler(racing, [8, 9], async () => {}, PACK, [], NOW.toISOString())(); + await failDelivery(store, opened, new ListenerUnroutedError("github", DEPLOYMENT.address), NOW); + + const mails: Array<{ items?: Array<{ prNumber: number }> }> = []; + await reconciler(store, [8, 9], async (payload) => { + mails.push(payload as never); + }, PACK, [], NOW.toISOString())(); + expect(mails.flatMap((mail) => mail.items?.map((item) => item.prNumber) ?? [])).toContain(9); + }); +}, 120_000); diff --git a/apps/hub/src/github/triage-reconciler.ts b/apps/hub/src/github/triage-reconciler.ts index 59d68164..1349ad6b 100644 --- a/apps/hub/src/github/triage-reconciler.ts +++ b/apps/hub/src/github/triage-reconciler.ts @@ -125,13 +125,21 @@ function headKey(row: Pick): string { return runKey(row.number, row.headSha); } -/** Theirs in their order, each head replaced by ours only when ours is newer; heads only we have follow. */ -function mergeRows(theirs: readonly PrTriageRow[], ours: readonly PrTriageRow[]): PrTriageRow[] { +/** Whether someone wrote the head's row after the pass read `read`. */ +function writtenSince(read: ReadonlyMap, row: PrTriageRow): boolean { + return read.get(headKey(row))?.updatedAt !== row.updatedAt; +} + +/** + * Theirs in their order, each head replaced by ours unless they wrote it after the pass read it, whatever either clock + * says; heads only we have follow. + */ +function mergeRows(read: ReadonlyMap, theirs: readonly PrTriageRow[], ours: readonly PrTriageRow[]): PrTriageRow[] { const byHead = new Map(ours.map((row) => [headKey(row), row])); - const merged = theirs.map(function newer(row) { + const merged = theirs.map(function unlessWritten(row) { const mine = byHead.get(headKey(row)); byHead.delete(headKey(row)); - return mine !== undefined && new Date(mine.updatedAt).getTime() > new Date(row.updatedAt).getTime() ? mine : row; + return mine === undefined || writtenSince(read, row) ? row : mine; }); return [...merged, ...byHead.values()]; } @@ -276,10 +284,15 @@ export function createTriageReconciler(deps: TriageReconcilerDeps) { if (!(err instanceof TriageStateConflictError)) throw err; } const fresh = await deps.store.load(tenantId, repo); + const read = new Map(stored.rows.map((row) => [headKey(row), row])); const theirs = new Map(fresh.rows.map((row) => [headKey(row), row])); const headOf = (queued: QueuedPr) => headKey(plan.rows.find((row) => row.number === queued.number)!); - const left = new Set(plan.enqueue.map(headOf).filter((key) => theirs.has(key) && inFlight(theirs.get(key)!))); - const rows = mergeRows(fresh.rows, plan.rows.filter((row) => !left.has(headKey(row)))); + function takenByThem(key: string): boolean { + const row = theirs.get(key); + return row !== undefined && (inFlight(row) || writtenSince(read, row)); + } + const left = new Set(plan.enqueue.map(headOf).filter(takenByThem)); + const rows = mergeRows(read, fresh.rows, plan.rows.filter((row) => !left.has(headKey(row)))); try { const version = await deps.store.save(tenantId, repo, rows, fresh.version); deps.log({ level: "info", msg: "triage_state_merged", tenantId, repo, left: left.size }); diff --git a/tests/chain/chain.ts b/tests/chain/chain.ts index c494e52f..9d438a2f 100644 --- a/tests/chain/chain.ts +++ b/tests/chain/chain.ts @@ -8,6 +8,7 @@ import { join } from "node:path"; import { sql } from "drizzle-orm"; import { createDB } from "@intx/db"; import { listWorkflowDeployments, listWorkflowRuns, readWorkflowRunEvents } from "@intx/hub-client"; +import { parseTriageState, triageStateName, type PrTriageRow } from "@corbits/triage-contracts"; import { saveAutomationMode } from "../../apps/web/src/lib/hub-api.ts"; import { fakeGithub, OWNER, type Pull, type Write } from "./github.ts"; import { stubModel, type Answer, type Question } from "./model.ts"; @@ -183,16 +184,26 @@ export function chainStack(env: Record = {}) { return found.map((row) => ({ id: String(row["id"]), createdAt: new Date(String(row["created_at"])), status: String(row["status"]) })); } - /** The claim once its manager's run is gone, whether done ended the claim or the run stopped without ending it. */ + /** The pull request's rows in the hub's triage state. */ + async function triageRows(repo: string, number: number): Promise { + const [stored] = await rows(`select content from artifacts.artifact where tenant_id = '${workspace.tenantId}' and title = '${triageStateName(repo)}' and archived_at is null`); + return parseTriageState(stored?.["content"], repo)?.prs.filter((row) => row.number === number) ?? []; + } + + /** The claim once its manager's run is gone, whether done ended the claim or the run stopped without ending it; on a timeout, the error names the head's triage rows. */ async function settled(repo: string, number: number, passes = 1, ms = SETTLE_MS): Promise { - return until(`${repo}#${number}'s manager to stop`, async function stopped() { - const claim = await claimOf(repo, number); - if (!claim || claim.history.length < passes) return undefined; - const runs = await managers(); - const runId = claim.runId ?? claim.history.at(-1)!.runId; - const live = runs.some((run) => run.id === runId && isLiveDeployment(run.status)); - return live ? undefined : claim; - }, ms); + try { + return await until(`${repo}#${number}'s manager to stop`, async function stopped() { + const claim = await claimOf(repo, number); + if (!claim || claim.history.length < passes) return undefined; + const runs = await managers(); + const runId = claim.runId ?? claim.history.at(-1)!.runId; + const live = runs.some((run) => run.id === runId && isLiveDeployment(run.status)); + return live ? undefined : claim; + }, ms); + } catch (err) { + throw new Error(`${String(err)}\nTriage state: ${JSON.stringify(await triageRows(repo, number))}`, { cause: err }); + } } async function verdictsOf(repo: string, number: number): Promise[]> {