diff --git a/apps/hub/src/github/manager-runs.postgres.test.ts b/apps/hub/src/github/manager-runs.postgres.test.ts index 21957dca2..8bd54eb3d 100644 --- a/apps/hub/src/github/manager-runs.postgres.test.ts +++ b/apps/hub/src/github/manager-runs.postgres.test.ts @@ -180,3 +180,22 @@ test("PostgreSQL: a manager up between passes awaits the reviewers on its head t expect(after?.awaiting).toBeUndefined(); }); }, 60_000); + +test("PostgreSQL: a head whose manager ended while it waited for CI shows as waiting with no run going, and is not mailed again", async () => { + await withDatabase(async function ended(db) { + const now = new Date(); + const managerRuns = createManagerRuns(db, createVerdictRecordStore(db), () => now); + await managerUp(db, "run_waited"); + await mailed(db, pr8, "h1"); + await take(db, TENANT, pr8, "run_waited"); + expect(await done(db, TENANT, pr8, "run_waited", [], { report: { head: "h1", waiting: true } })).toEqual({ outcome: "ended" }); + + const observed = await managerRuns(TENANT, [pr8]); + expect(observed.byRepo.get(REPO)?.get(runKey(8, "h1"))).toEqual([expect.objectContaining({ runId: "run_waited", status: "completed", waitingSince: expect.any(String) })]); + const firstSeenAt = new Date(now.getTime() - MINUTE).toISOString(); + const row: PrTriageRow = { number: 8, headSha: "h1", status: "new", attempts: 0, firstSeenAt, updatedAt: firstSeenAt }; + const plan = planTenant({ repos: [{ name: REPO, prs: [open(8, now)], rows: [row] }], runs: observed.byRepo, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!; + expect(plan.rows).toEqual([expect.objectContaining({ number: 8, status: "triaged", runId: "run_waited" })]); + expect(plan.enqueue).toEqual([]); + }); +}, 60_000); diff --git a/apps/hub/src/github/manager-runs.ts b/apps/hub/src/github/manager-runs.ts index ace0a75ea..63427b7ad 100644 --- a/apps/hub/src/github/manager-runs.ts +++ b/apps/hub/src/github/manager-runs.ts @@ -33,10 +33,12 @@ export function createManagerRuns(db: DB["db"], verdicts: Pick 0 && { awaiting: live.awaiting }), ...(live.waitingSince !== undefined && { waitingSince: live.waitingSince }) }); + // No manager is up while the head waits only on CI; the reconciler wakes one once CI settles. + if (waiting) observed(pr.repo, pr.number, waiting.head, { runId: waiting.runId, source, status: "completed", startedAt: waiting.since, waitingSince: waiting.since }); if (gone) observed(pr.repo, pr.number, gone.pass.head, { runId: gone.runId, source, status: "failed", startedAt: gone.pass.takenAt }); for (const verdict of recorded.filter((row) => row.repo === pr.repo && row.number === pr.number)) { observed(pr.repo, pr.number, verdict.headSha, { runId: verdict.runId, source, status: "completed", startedAt: verdict.startedAt.toISOString(), ...settlesBy(verdict), ...(verdict.trigger !== null && { trigger: verdict.trigger }) }); diff --git a/apps/hub/src/github/reconcile-plan.ts b/apps/hub/src/github/reconcile-plan.ts index b384641fd..1d74f63dc 100644 --- a/apps/hub/src/github/reconcile-plan.ts +++ b/apps/hub/src/github/reconcile-plan.ts @@ -21,7 +21,7 @@ export type ObservedRun = { passAt?: string; /** What a running manager's reviewers on its head have yet to report. */ awaiting?: AwaitedStage[]; - /** Since when a running manager has waited for its head's CI. */ + /** Since when the head has waited for its CI: a running manager's, or the last pass of a manager that ended to wait. */ waitingSince?: string; }; @@ -244,6 +244,18 @@ function runningHeads(runs: TenantPlanInput["runs"], now: Date, policy: Reconcil return busy; } +/** The tenant's in-flight slots `plans` leave free: `maxInFlight` less the heads queued or running, a manager waiting for CI aside. */ +export function freeSlots(plans: ReadonlyMap, runs: TenantPlanInput["runs"], now: Date, policy: ReconcilePolicy): number { + const busy = runningHeads(runs, now, policy); + for (const [name, plan] of plans) { + for (const row of plan.rows) { + const key = runKey(row.number, row.headSha); + if (inFlight(row) && !parked(runs.get(name)?.get(key) ?? [])) busy.add(`${name} ${key}`); + } + } + return Math.max(policy.maxInFlight - busy.size, 0); +} + type Due = { repo: string; index: number; row: PrTriageRow }; /** A head with no run and no verdict, even if its mail failed to deliver. */ diff --git a/apps/hub/src/github/triage-reconciler.test.ts b/apps/hub/src/github/triage-reconciler.test.ts index fc5b5b622..532b14306 100644 --- a/apps/hub/src/github/triage-reconciler.test.ts +++ b/apps/hub/src/github/triage-reconciler.test.ts @@ -228,6 +228,43 @@ describe("triage reconciler", () => { expect(mailedNumbers(nowCancelled.delivered)).toEqual([1]); }); + test("a head whose manager ended to wait for CI is queued once CI settles, no checks start within the grace, or the hour is up, reading its checks once a pass until then", async () => { + function waitedFor(minutes: number): Runs { + const since = at(-minutes).toISOString(); + return runsOf([1, { runId: "run_1", source: "github", status: "completed", startedAt: since, waitingSince: since }]); + } + const running = [{ name: "build", status: "in_progress", conclusion: null }]; + const pending = harness(waitedFor(5), [head(1)], { headChecks: running }); + await pending.reconcile(); + expect(pending.delivered).toHaveLength(0); + expect(pending.logged).toContainEqual(expect.objectContaining({ msg: "waiting_ci_read", repo: REPO, heads: 1 })); + const finished = harness(waitedFor(5), [head(1)], { headChecks: [{ name: "build", status: "completed", conclusion: "success" }] }); + await finished.reconcile(); + expect(mailedNumbers(finished.delivered)).toEqual([1]); + expect(finished.logged).toContainEqual(expect.objectContaining({ msg: "triage_requeued", pr: 1, reason: "ci_settled" })); + const starting = harness(waitedFor(1), [head(1)], { headChecks: [] }); + await starting.reconcile(); + expect(starting.delivered).toHaveLength(0); + const none = harness(waitedFor(3), [head(1)], { headChecks: [] }); + await none.reconcile(); + expect(mailedNumbers(none.delivered)).toEqual([1]); + const over = harness(waitedFor(61), [head(1)], { headChecks: running }); + await over.reconcile(); + expect(mailedNumbers(over.delivered)).toEqual([1]); + expect(over.logged).not.toContainEqual(expect.objectContaining({ msg: "waiting_ci_read" })); + }); + + test("woken heads take the tenant's free in-flight slots, longest waiting first, and the rest wait for a later pass", async () => { + function waited(number: number, minutes: number): [number, ObservedRun] { + const since = at(-minutes).toISOString(); + return [number, { runId: `run_${number}`, source: "github", status: "completed", startedAt: since, waitingSince: since }]; + } + const finished = [{ name: "build", status: "completed", conclusion: "success" }]; + const { reconcile, delivered } = harness(runsOf(waited(1, 5), waited(2, 20)), [head(1), head(2)], { maxInFlight: "1", headChecks: finished }); + await reconcile(); + expect(mailedNumbers(delivered)).toEqual([2]); + }); + test("a verdict that read CI as cancelled is not queued again while it stays cancelled, however many reads, and is once a check is re-run", async () => { const cancelled = { ...current(1), repo: REPO, number: 1, state: "awaiting-review", actor: "system", humanGated: false, checks: [{ check: "ci-green", kind: "machine", result: "unconfirmed", output: { kind: "values", values: { status: "cancelled" } } }] }; let now = NOW; diff --git a/apps/hub/src/github/triage-reconciler.ts b/apps/hub/src/github/triage-reconciler.ts index 1349ad6b3..f0b7f2843 100644 --- a/apps/hub/src/github/triage-reconciler.ts +++ b/apps/hub/src/github/triage-reconciler.ts @@ -5,13 +5,13 @@ // live listener is left alone. A settled head whose verdict waited on CI is // queued again here once its CI finished, on hosts and pull requests whose CI // results send no webhook. -import { hasEnabledRules, repoPolicy, type PrTriageRow, type RepoPolicy, type RepoSource, type VerdictDetail } from "@corbits/triage-contracts"; +import { CI_START_MS, CI_WAIT_MS, hasEnabledRules, repoPolicy, type PrTriageRow, type RepoPolicy, type RepoSource, type VerdictDetail } from "@corbits/triage-contracts"; import { mailPayload, unreadablePayload, type PolicyUnreadable } from "./bridge.js"; import type { TriageJob } from "./triage-queue.js"; import { GithubRateLimited } from "./github-usage.js"; import { failDelivery } from "./triage-queue.js"; import type { PolicyReader, ResolvedPack } from "./repo-policy.js"; -import { inFlight, planTenant, requeue, runKey, type OpenPr, type QueuedPr, type ReconcilePolicy, type RepoPlan } from "./reconcile-plan.js"; +import { freeSlots, inFlight, planTenant, requeue, runKey, type ObservedRun, type OpenPr, type QueuedPr, type ReconcilePolicy, type RepoPlan } from "./reconcile-plan.js"; import { summarizeChecks } from "@corbits/github-reads"; import { repoRecords, sourceOf, triageNs, triageStateKey, type RepoRecord } from "./tenant-config.js"; import type { HeadChecksReader, OpenHeadsReader } from "./tenant-open-heads.js"; @@ -120,6 +120,8 @@ function enabledRepos(config: unknown, source: RepoSource): EnabledRepo[] { } type Claimed = { rows: PrTriageRow[]; version: TriageStateVersion; enqueue: QueuedPr[] }; +/** A head whose manager ended to wait for its CI, ready to wake, with since when it has waited. */ +type Settled = { plan: RepoPlan; row: PrTriageRow; since: string }; function headKey(row: Pick): string { return runKey(row.number, row.headSha); @@ -145,6 +147,7 @@ function mergeRows(read: ReadonlyMap, theirs: readonly PrTr } const BUILD_STATUS_CHANGED = "build_status_changed"; +const CI_SETTLED = "ci_settled"; type BuildPoll = { polls: number; at: number }; @@ -245,6 +248,36 @@ export function createTriageReconciler(deps: TriageReconcilerDeps) { } } + /** + * The heads whose manager ended to wait for their CI that a woken manager has something to decide on: CI no longer + * pending, no checks past the start grace, or the CI wait over. One read of the head's checks per head per pass. + */ + async function settledWaiting(pass: TenantPass, repo: LoadedRepo, plan: RepoPlan, runs: ReadonlyMap | undefined): Promise { + const name = repo.record.name; + const now = deps.now(); + const settled: Settled[] = []; + let read = 0; + for (const row of plan.rows.filter((candidate) => candidate.status === "triaged")) { + const since = runs?.get(runKey(row.number, row.headSha))?.find((run) => run.status === "completed" && run.waitingSince !== undefined)?.waitingSince; + if (since === undefined) continue; + const waited = now.getTime() - Date.parse(since); + if (waited < CI_WAIT_MS) { + read += 1; + let ci: ReturnType; + try { + ci = summarizeChecks(await deps.headChecks(pass.tenantId, repo.record, row.headSha)); + } catch (err) { + deps.log({ level: "warn", msg: "build_status_unread", tenantId: pass.tenantId, repo: name, pr: row.number, headSha: row.headSha, error: String(err) }); + continue; + } + if (ci === "pending" || (ci === "none" && waited < CI_START_MS)) continue; + } + settled.push({ plan, row, since }); + } + if (read > 0) deps.log({ level: "info", msg: "waiting_ci_read", tenantId: pass.tenantId, repo: name, heads: read }); + return settled; + } + async function loadRepo(pass: TenantPass, repo: EnabledRepo): Promise { const read = await deps.readCheckPack(pass.tenantId, repo.record.name, sourceOf(repo.record)); const pauseLabel = read.status === "ok" ? read.pauseLabel : undefined; @@ -378,9 +411,24 @@ export function createTriageReconciler(deps: TriageReconcilerDeps) { now: deps.now(), policy: deps.policy, }); + const polled: LoadedRepo[] = []; + const settled: Settled[] = []; for (const repo of loaded) { try { await pollBuilds(pass, repo, plans.get(repo.record.name)!); + settled.push(...(await settledWaiting(pass, repo, plans.get(repo.record.name)!, managed.byRepo.get(repo.record.name)))); + polled.push(repo); + } catch (err) { + if (reconcileFailed({ tenantId: tenant.id, repo: repo.record.name }, err)) failed = true; + } + } + // Woken heads take in-flight slots like new ones, longest waiting first; the rest wait for a later pass. + const now = deps.now(); + for (const { plan, row } of settled.sort((a, b) => Date.parse(a.since) - Date.parse(b.since)).slice(0, freeSlots(plans, managed.byRepo, now, deps.policy))) { + requeue(plan, row, CI_SETTLED, now, deps.policy); + } + for (const repo of polled) { + try { await deliverRepo(pass, repo, plans.get(repo.record.name)!); } catch (err) { if (reconcileFailed({ tenantId: tenant.id, repo: repo.record.name }, err)) failed = true; diff --git a/apps/hub/src/github/triage-verdicts.postgres.test.ts b/apps/hub/src/github/triage-verdicts.postgres.test.ts index 3fdd23f27..13b7df5fe 100644 --- a/apps/hub/src/github/triage-verdicts.postgres.test.ts +++ b/apps/hub/src/github/triage-verdicts.postgres.test.ts @@ -59,7 +59,7 @@ function observed(byHead: Array<[string, ObservedRun[]]>): ObservedRuns { const NO_RUNS: ObservedRuns = { byRepo: new Map() }; /** - * The managers' runs: #8's newest run failed after its verdict, #9 and #10 are still running, #7's waits for its CI, and #6's run failed early. + * The managers' runs: #8's newest run failed after its verdict, #9 and #10 are still running, #7's ended waiting for its CI, and #6's run failed early. * #4 of the repository GitHub could not list has a failed run. */ const MANAGED: ObservedRuns = { @@ -67,7 +67,7 @@ const MANAGED: ObservedRuns = { ...observed([ ["8@sha_run_3", [live("run_3", "failed", minutes(9))]], ["9@sha9", [live("run_9", "running", minutes(-5), "pull_request:synchronize")]], - ["7@sha_waiting", [{ ...live("run_7", "running", minutes(-3)), waitingSince: minutes(-3).toISOString() }]], + ["7@sha_waiting", [{ ...live("run_7", "completed", minutes(-3)), waitingSince: minutes(-3).toISOString() }]], ["10@sha10", [live("run_10", "running", minutes(-120))]], ["6@sha6", [live("run_6", "failed", minutes(-3))]], ]).byRepo, diff --git a/apps/hub/src/github/triage-verdicts.ts b/apps/hub/src/github/triage-verdicts.ts index 3ee52b1cd..32f3ce542 100644 --- a/apps/hub/src/github/triage-verdicts.ts +++ b/apps/hub/src/github/triage-verdicts.ts @@ -110,11 +110,12 @@ function livePulls(runs: ObservedRuns): Map { return byPull; } -/** A manager waiting for its head's CI shows that; one up between passes runs nothing, so it is awaiting its reviewers or has none out; a running one counts from the pass it took. */ +/** A head waiting for its CI shows that, whether its manager is up or ended to wait; one up between passes runs nothing, so it is awaiting its reviewers or has none out; a running one counts from the pass it took. */ function runOf(live: LivePull["runs"][number] | undefined, now: Date): Run | null { - if (!live || live.status === "completed") return null; - const { runId } = live; + if (!live) return null; if (live.waitingSince !== undefined) return { status: "waiting-ci", since: live.waitingSince, headSha: live.headSha }; + if (live.status === "completed") return null; + const { runId } = live; if (live.status === "failed" || live.status === "cancelled") return { runId, startedAt: live.startedAt, status: live.status }; if (live.passAt === undefined) return live.awaiting === undefined ? null : { runId, status: "awaiting", on: live.awaiting }; const startedAt = live.passAt; diff --git a/apps/hub/src/manager-intake.postgres.test.ts b/apps/hub/src/manager-intake.postgres.test.ts index 5c163e2b4..4e53f2ffe 100644 --- a/apps/hub/src/manager-intake.postgres.test.ts +++ b/apps/hub/src/manager-intake.postgres.test.ts @@ -54,6 +54,7 @@ function stack(db: Db, listener: string | undefined, pull = PULL, gate: Gate = p const logged: Record[] = []; const cancelled: { runId: string; reason: string }[] = []; const refusing = new Set(); + const unroutable = { on: false }; const { supersede, supersedeAll } = createSupersede({ db, async cancel(_tenantId, runId, reason) { @@ -71,13 +72,14 @@ function stack(db: Db, listener: string | undefined, pull = PULL, gate: Gate = p withAppLogin: async (_tenantId, payload) => ({ ...payload, appLogin: "triage[bot]" }), async deliver(_tenantId, address, payload) { if (!routable) throw new RunUnavailableError("the listener is not running"); + if (unroutable.on) throw new Error("the listener is unroutable"); mailed.push({ address, payload }); }, supersede, gate, log: (entry) => logged.push(entry), }); - return { toListener, mailed, logged, cancelled, refusing, supersedeAll }; + return { toListener, mailed, logged, cancelled, refusing, unroutable, supersedeAll }; } async function events(db: Db): Promise<{ key: string; state: { head: string; payload: Record } }[]> { @@ -114,7 +116,7 @@ test("PostgreSQL: a held Do's approval while the hub holds no route to the liste }); }, 60_000); -test("PostgreSQL: each pull request a mail names is recorded as its own event, a webhook once however often it is delivered and any other mail every time, and the listener is mailed the heads", async () => { +test("PostgreSQL: each pull request a mail names is recorded as its own event, a webhook once however often it is delivered and any other mail every time, and the listener is asked once for the manager they wait for", async () => { await withDatabase(async function listener(db) { const { toListener, mailed } = stack(db, LISTENER); const webhook = mailPayload(RECORD, POLICY, STORED, { prNumber: 7, headSha: "h1", deliveryId: "d1" }); @@ -139,7 +141,7 @@ test("PostgreSQL: each pull request a mail names is recorded as its own event, a const item = recorded.find((row) => row.state.payload["kind"] === "pr" && row.state.payload["deliveryId"] === undefined)!; expect(item.state.payload).toMatchObject({ prNumber: 7, headSha: "h1", appLogin: "triage[bot]" }); expect(item.state.payload).not.toHaveProperty("items"); - expect(mailed.map((mail) => mail.address)).toEqual(Array(6).fill(LISTENER)); + expect(mailed.map((mail) => mail.address)).toEqual([LISTENER]); expect(mailed[0]!.payload).toEqual({ kind: "pr", repo: RECORD.name, tenantId: TENANT, items: [{ prNumber: 7, headSha: "h1" }] }); }); }, 60_000); @@ -165,6 +167,38 @@ test("PostgreSQL: a pull request whose manager is up is mailed that manager, and }); }, 60_000); +test("PostgreSQL: a mail for a pull request with no manager up asks the listener once until a manager takes it, then again, and a failed ask is asked again", async () => { + await withDatabase(async function onItsWay(db) { + await claimManager(db, "run_mgr", 7, "completed"); + await db.execute(sql`UPDATE corbits_triage.manager_state SET state = (state - 'runId') || '{"status":"ended"}' WHERE kind = 'claim'`); + const { toListener, mailed } = stack(db, LISTENER); + const asked = { address: LISTENER, payload: { kind: "pr", repo: RECORD.name, tenantId: TENANT, items: [{ prNumber: 7, headSha: "h1" }] } }; + expect(await toListener(TENANT, mailPayload(RECORD, POLICY, STORED, { prNumber: 7, headSha: "h1" }))).toBe(true); + expect(await toListener(TENANT, requeued(7, "h1"))).toBe(true); + expect(mailed).toEqual([asked]); + await db.execute(sql`UPDATE corbits_triage.manager_state SET state = state || jsonb_build_object('claimedAt', ${new Date(Date.now() + 1000).toISOString()}::text) WHERE kind = 'claim'`); + expect(await toListener(TENANT, requeued(7, "h1"))).toBe(true); + expect(mailed).toEqual([asked, asked]); + + // A pull request no manager has taken yet asks once too. + const fresh = { address: LISTENER, payload: { kind: "pr", repo: RECORD.name, tenantId: TENANT, items: [{ prNumber: 8, headSha: "h8" }] } }; + expect(await toListener(TENANT, requeued(8, "h8"))).toBe(true); + expect(await toListener(TENANT, requeued(8, "h8"))).toBe(true); + expect(mailed).toEqual([asked, asked, fresh]); + }); +}, 60_000); + +test("PostgreSQL: an ask whose mail to the listener failed is withdrawn, so the retry within the start grace mails the listener", async () => { + await withDatabase(async function retried(db) { + const { toListener, mailed, unroutable } = stack(db, LISTENER); + unroutable.on = true; + await expect(toListener(TENANT, requeued(7, "h1"))).rejects.toThrow("the listener is unroutable"); + unroutable.on = false; + expect(await toListener(TENANT, requeued(7, "h1"))).toBe(true); + expect(mailed).toEqual([{ address: LISTENER, payload: { kind: "pr", repo: RECORD.name, tenantId: TENANT, items: [{ prNumber: 7, headSha: "h1" }] } }]); + }); +}, 60_000); + test("PostgreSQL: a reviewer's result wakes the pull request's live manager, or has the listener deploy one once its claim ended, and none without a listener", async () => { await withDatabase(async function wake(db) { await claimManager(db, "run_mgr", 7, "running"); @@ -249,7 +283,7 @@ test("PostgreSQL: the reconcile pass wakes a manager for a pull request whose re // A manager the wake started took the claim and went away before done; past the start grace the next pass wakes again. const earlier = new Date(Date.now() - 3 * 60_000); - await db.execute(sql`UPDATE corbits_triage.manager_state SET state = state || ${JSON.stringify({ overdueWokenAt: earlier.toISOString(), claimedAt: new Date(earlier.getTime() + 1_000).toISOString() })}::jsonb WHERE key = ${`mgr:github:${RECORD.name}#7`}`); + await db.execute(sql`UPDATE corbits_triage.manager_state SET state = state || ${JSON.stringify({ overdueWokenAt: earlier.toISOString(), deployAskedAt: earlier.toISOString(), claimedAt: new Date(earlier.getTime() + 1_000).toISOString() })}::jsonb WHERE key = ${`mgr:github:${RECORD.name}#7`}`); await wakeOverdue(TENANT); expect(mailed).toEqual([woken, woken]); }); @@ -337,7 +371,7 @@ test("PostgreSQL: every pull request a mail names passes the gate, whatever sent const recorded = await events(db); expect(recorded.map((row) => row.state.head)).toEqual(["h1", "h1"]); expect(recorded.every((row) => JSON.stringify(row.state.payload["automatic"]) === JSON.stringify(["comment"]))).toBe(true); - expect(mailed.flatMap((mail) => (mail.payload as { items: { prNumber: number }[] }).items.map((item) => item.prNumber))).toEqual([7, 7]); + expect(mailed.flatMap((mail) => (mail.payload as { items: { prNumber: number }[] }).items.map((item) => item.prNumber))).toEqual([7]); }); }, 120_000); diff --git a/apps/hub/src/manager-intake.ts b/apps/hub/src/manager-intake.ts index a75093e2a..1682adb9f 100644 --- a/apps/hub/src/manager-intake.ts +++ b/apps/hub/src/manager-intake.ts @@ -11,7 +11,7 @@ import { APPROVE_DO_KIND, repoSourceOf, type RepoSource } from "@corbits/triage- import { reviewRequestOf, type PullRequestToReview } from "./pull-request-reader.js"; import { DeploymentNotReadyError, ListenerUnroutedError, RunUnavailableError } from "./github/deployment.js"; import { repoRecords, sourceOf, triageNs, type RepoRecord } from "./github/tenant-config.js"; -import { dropReviewer, liveManager, overduePulls, recordEvent, recordOverdueWake, reviewedPulls, superseded, type PullRequestRef } from "./manager-state.js"; +import { askDeploy, dropReviewer, liveManager, overduePulls, recordEvent, recordOverdueWake, reviewedPulls, superseded, withdrawDeploy, type PullRequestRef } from "./manager-state.js"; export type ManagerIntakeDeps = { db: DB["db"]; @@ -89,6 +89,18 @@ async function deployFor(deps: Delivery, tenantId: string, address: string, sour } } +type Asked = { pr: PullRequestRef; number: number; head: string; at: string }; + +/** Mails the listener the heads it was asked to deploy a manager for, withdrawing each ask when the mail fails so the next mail asks again. */ +async function deployAsked(deps: Delivery, tenantId: string, address: string, source: RepoSource, repo: string, asked: readonly Asked[]): Promise { + try { + await deployFor(deps, tenantId, address, source, repo, asked); + } catch (err) { + for (const each of asked) await withdrawDeploy(deps.db, tenantId, each.pr, each.at); + throw err; + } +} + /** * Wakes the manager of a pull request whose reviewer result was recorded, as a host event does: its live manager is * mailed, or the listener deploys one. False when the workspace has no listener for the host. @@ -97,7 +109,9 @@ export function createManagerWake(deps: Delivery & Pick { const address = await deps.listener(tenantId, pr.source); if (!address) return false; - if (!(await mailedLive(deps, tenantId, domainOf(address), pr))) await deployFor(deps, tenantId, address, pr.source, pr.repo, [{ number: pr.number, head }]); + if (await mailedLive(deps, tenantId, domainOf(address), pr)) return true; + const at = await askDeploy(deps.db, tenantId, pr); + if (at !== undefined) await deployAsked(deps, tenantId, address, pr.source, pr.repo, [{ pr, number: pr.number, head, at }]); return true; }; } @@ -233,7 +247,7 @@ export function createManagerIntake(deps: ManagerIntakeDeps) { throw new ListenerUnroutedError(source, address); } const domain = domainOf(address); - const start: Named[] = []; + const start: Asked[] = []; try { for (const asked of pulls) { const request = await deps.gate(tenantId, record, asked.number, asked.request); @@ -243,10 +257,12 @@ export function createManagerIntake(deps: ManagerIntakeDeps) { const pr = { source, repo, number: pull.number }; await recordEvent(deps.db, tenantId, pr, pull.id, { type: "webhook", head: pull.head, payload: recorded as Record }); await deps.supersede(tenantId, pr); - if (!(await mailedLive(deps, tenantId, domain, pr))) start.push(pull); + if (await mailedLive(deps, tenantId, domain, pr)) continue; + const at = await askDeploy(deps.db, tenantId, pr); + if (at !== undefined) start.push({ pr, number: pull.number, head: pull.head, at }); } } finally { - if (start.length > 0) await deployFor(deps, tenantId, address, source, repo, start); + if (start.length > 0) await deployAsked(deps, tenantId, address, source, repo, start); } return true; }; diff --git a/apps/hub/src/manager-state.ts b/apps/hub/src/manager-state.ts index 3b81020ad..de0a2a469 100644 --- a/apps/hub/src/manager-state.ts +++ b/apps/hub/src/manager-state.ts @@ -7,7 +7,7 @@ import { jsonb, primaryKey, text } from "drizzle-orm/pg-core"; import { type } from "arktype"; import { canExecuteWorkflowRun, schema, type DB } from "@intx/db"; import { asRecord } from "@corbits/common"; -import { APPROVE_DO_KIND, AWAITED_STAGES, REPO_SOURCES, TRIAGE_JUDGES, TRIAGE_LISTENERS, TRIAGE_MANAGERS, type AwaitedStage, type RepoSource } from "@corbits/triage-contracts"; +import { APPROVE_DO_KIND, AWAITED_STAGES, MANAGER_READY_MS, REPO_SOURCES, TRIAGE_JUDGES, TRIAGE_LISTENERS, TRIAGE_MANAGERS, type AwaitedStage, type RepoSource } from "@corbits/triage-contracts"; import type { Tx } from "./github/admin-github-grant.js"; import { resolveLiveDeployments, workflowName, type LiveDeployment } from "./github/deployment.js"; import { triageSchema } from "./triage-db.js"; @@ -32,6 +32,7 @@ const ClaimState = type({ history: PassState.and({ runId: "string > 0", doneAt: "string.date.iso", "waiting?": "true", "request?": "Record" }).array(), starts: "number.integer >= 0", "overdueWokenAt?": "string.date.iso", + "deployAskedAt?": "string.date.iso", }); const EventPayload = type("Record").narrow(function plainObject(value) { return !Array.isArray(value); @@ -573,9 +574,69 @@ export async function passRequest(db: DB["db"], tenantId: string, pr: PullReques /** * A pull request a manager claimed: its closed passes; while its manager is up, that run, the head of the newest pull * request mail recorded, the pass it has open, since when it has waited for that head's CI and what its reviewers on - * that head have yet to report before their deadlines; and the pass a manager that went away left open. + * that head have yet to report before their deadlines; the pass a manager that went away left open; and, once no + * manager is up, the newest head its last pass left waiting for CI, with that pass's run and since when it waited. */ -export type ManagedPull = { pr: PullRequestRef; history: ClosedPass[]; live?: { runId: string; head?: string; pass?: Pass; waitingSince?: string; awaiting: AwaitedStage[] }; gone?: { runId: string; pass: Pass } }; +export type ManagedPull = { + pr: PullRequestRef; + history: ClosedPass[]; + live?: { runId: string; head?: string; pass?: Pass; waitingSince?: string; awaiting: AwaitedStage[] }; + gone?: { runId: string; pass: Pass }; + waiting?: { runId: string; head: string; since: string }; +}; + +/** The newest head, when the last pass on it of an ended claim waited for its CI. */ +async function endedWaiting(tx: Tx, tenantId: string, pr: PullRequestRef, history: Claim["history"]): Promise { + const head = await newestHead(tx, tenantId, pr); + if (head === undefined) return undefined; + const since = waitingSince(history, head); + const last = history.findLast((pass) => pass.head === head); + return since === undefined || last === undefined ? undefined : { runId: last.runId, head, since }; +} + +/** + * How long an ask holds off the next: the listener's wake of the manager it deployed and of the one it deploys in its + * place when that wake fails, each waiting `MANAGER_READY_MS`, and stock's 30-second cancel grace between them. + */ +const DEPLOY_ASK_HOLD_MS = 2 * MANAGER_READY_MS + 30_000; + +/** Whether a manager the hub asked the listener for has not taken the claim yet and may still be starting. */ +function deploying(claim: Claim, now: Date): boolean { + if (claim.deployAskedAt === undefined) return false; + const asked = Date.parse(claim.deployAskedAt); + const taken = claim.starts > 0 && Date.parse(claim.claimedAt) >= asked; + return !taken && now.getTime() - asked < DEPLOY_ASK_HOLD_MS; +} + +/** + * When the hub asked the listener to deploy a manager for the pull request, recorded on its claim, which a pull request + * no manager has taken yet gets with no start. Undefined while one asked for earlier is still on its way: the mail is + * recorded, and that manager takes it. + */ +export async function askDeploy(db: DB["db"], tenantId: string, pr: PullRequestRef): Promise { + return db.transaction(async function asked(tx) { + const key = claimKey(pr); + const now = new Date(); + const unstarted = { status: "ended" as const, claimId: randomUUID(), claimedAt: now.toISOString(), reviewers: [], history: [], starts: 0 }; + await tx.insert(managerState).values({ tenantId, key, kind: "claim", state: claimOf(key, unstarted) }).onConflictDoNothing(); + const claim = await lockClaim(tx, tenantId, pr); + if (!claim) throw new Error(`manager_state ${key} vanished under its lock`); + if (deploying(claim, now)) return undefined; + const at = now.toISOString(); + await writeClaim(tx, tenantId, pr, { ...claim, deployAskedAt: at }); + return at; + }); +} + +/** Withdraws the ask made `at` once its mail to the listener failed, so the next mail asks again. */ +export async function withdrawDeploy(db: DB["db"], tenantId: string, pr: PullRequestRef, at: string): Promise { + await db.transaction(async function withdrawn(tx) { + const claim = await lockClaim(tx, tenantId, pr); + if (!claim) return; + const { deployAskedAt, ...kept } = claim; + if (deployAskedAt === at) await writeClaim(tx, tenantId, pr, kept); + }); +} function awaitedOn(reviewers: readonly Reviewer[], head: string | undefined, now: Date): AwaitedStage[] { return AWAITED_STAGES.filter((kind) => reviewers.some((reviewer) => reviewer.kind === kind && reviewer.head === head && reviewer.result === undefined && Date.parse(reviewer.deadline) > now.getTime())); @@ -734,7 +795,8 @@ export async function managedPulls(db: DB["db"], tenantId: string, pulls: readon if (!pr) throw new Error(`manager_state ${key} is not a claim asked for`); const runId = claim.status === "live" ? claim.runId : undefined; if (runId === undefined) { - managed.push({ pr, history: claim.history }); + const waiting = await endedWaiting(tx, tenantId, pr, claim.history); + managed.push({ pr, history: claim.history, ...(waiting && { waiting }) }); } else if (await runGone(tx, runId, now)) { managed.push({ pr, history: claim.history, ...(claim.pass && { gone: { runId, pass: claim.pass } }) }); } else { diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index a9354bec7..11a81d8aa 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -244,7 +244,13 @@ any other tenant or an unknown sender. new manager to publish its key. When a wake mail fails for any reason, the listener cancels that manager through the stock stop-run route under the `manage` grant stock gives a deployment's creator, and its `unwoken` step's - output names each run it cancelled and why the mail failed. A run stock + output names each run it cancelled and why the mail failed. Its `retry` + step then deploys one more manager for each of those pull requests and + wakes it the same way, and `unwokenAgain` cancels that one too if its mail + fails, with no third. The claim's hold on a new ask (`deployAskedAt`) + covers both wakes, each waiting two minutes, plus 30 seconds, so it lapses + only after a double failure and the pull request's next mail asks the + listener again. A run stock no longer lets execute, as one whose cancellation is requested or whose deadline has passed, is mailed nothing and its own take is refused; one that holds the claim keeps it until it ends, so no second manager writes @@ -417,38 +423,53 @@ any other tenant or an unknown sender. that waits for CI, or the verdict it posted as its evaluate, mirror, execute and rules steps output it, with the facts the verdict keeps rather than the diff. Each branch that reports one of these has its own done step. A pass - asks done to `wait` while its head awaits CI, or a judge or code review it - mailed for that head has not reported. Done then answers `more` when an + asks done to `wait` only while a judge or code review it mailed for its + head is out and within its deadline. Done then answers `more` when an event is pending, and the next pass runs at once; `more` with `wait` after a pass that asked, and the next pass runs once 30 seconds have passed, - taking whatever arrived meanwhile; otherwise `ended`. So a manager ends - only once its head has its final verdict and the policy's actions ran or - were held: a check finishing, a reviewer's result or a new head reaches the - hub, which mails the live manager, and a new head's pass starts over. A - pull request with nothing outstanding has no manager. The hub keeps, on - the claim, whether each closed pass waited for CI, and answers take with - `waitingSince`, when the head's unbroken run of waiting passes began, and - `headSince`, when the head's first closed pass was taken. Until 2 minutes - after that, a pass sends `ciStarting`, and where CI green is on facts read a - head with no checks as checks pending ("checks have not started yet"), so - it waits for its CI to start; the pass at 2 minutes reads it again. Every - 5 minutes of that wait a pass triages the head again with no event, which - settles a Bitbucket head, whose builds send no webhook, and a GitHub head - whose `check_run` never arrived. A pass whose head has waited 60 minutes is - triaged with `noWait`, so it is - decided with CI still running. A manager lives three hours from its claim - (`claimedAt`, which take answers), time for CI, the judge and code review - to each take their hour; a new head does not extend it. The pass after that - stops waiting (`stopped` on take): it takes every reviewer still out as past - its deadline and decides with `noWait`, and its verdict says what Triage - stopped waiting for and holds every Do for a person. The loop's pass count - is only a backstop. For up to three hours a waiting manager holds one of - the `TRIAGE_MAX_LOCAL_SIDECARS` local sidecars, and its judge and code - review take more. Past the cap, deployments go to the sandbox provisioner; - with none configured they wait on stock's placement retry and then fail, - and the pull request is tried again on its next event or reconcile pass. That is - the bound (CL-11587); nothing else limits waiting managers. A manager - waiting for CI is parked: the reconciler does not count it toward the + taking whatever arrived meanwhile; otherwise `ended`. A head that waits + only on the host (its CI, or mergeability) posts that it waits and its + manager ends, so no manager sits on a sidecar for CI; a reviewer's result + that lands while the head still waits on the host posts the waiting state + the same way, and the judgement stays on the reviewer for the pass after + CI. A check finishing, a reviewer's result or a new head reaches the hub, + which mails the live manager or has the listener deploy one, and a new + head's pass starts over. The listener is asked once per manager start: the + claim keeps when it was asked (`deployAskedAt`), and until a manager takes + the claim, or for four and a half minutes at most (both of the listener's + wakes, below), a further mail is only + recorded for that manager to take; a mail that failed to reach the listener + withdraws its ask. A pull request with nothing outstanding has no + manager. The hub keeps, on the claim, whether each closed pass waited for + CI, and answers take with `waitingSince`, when the head's unbroken run of + waiting passes began, and `headSince`, when the head's first closed pass + was taken. Until 2 minutes after that (`CI_START_MS`), a pass sends + `ciStarting`, and where CI green is on facts read a head with no checks as + checks pending ("checks have not started yet"), so it waits for its CI to + start. For a head whose manager ended to wait, the triage reconcile pass + reads the head's checks once per pass (`waiting_ci_read` logs how many) + and queues it again (`triage_requeued`, `reason: ci_settled`) once CI is no + longer pending, the head still has no checks past the start grace, or it + has waited 60 minutes (`CI_WAIT_MS`), without reading. That settles a + Bitbucket head, whose builds send no webhook, and a GitHub head whose + `check_run` never arrived. A pass whose head has waited 60 minutes is + triaged with `noWait`, so it is decided with CI still running. The portal's + Waiting on CI comes from the claim's last pass on the newest head, whether + or not its manager is up. A manager lives three hours from its claim + (`claimedAt`, which take answers); a new head does not extend it. Since a + manager ends while it waits only on CI, the clock bounds one its reviewers + keep alive, such as a judge or code review deployed again on each new + push. The pass after that stops waiting (`stopped` on take): it takes every + reviewer still out as past its deadline and decides with `noWait`, and its + verdict says what Triage stopped waiting for and holds every Do for a + person. The loop's pass count is only a backstop. A manager holds one of + the `TRIAGE_MAX_LOCAL_SIDECARS` local sidecars only while it runs a pass or + waits on a reviewer, and its judge and code review take more. Past the cap, + deployments go to the sandbox provisioner; with none configured they wait + on stock's placement retry and then fail, and the pull request is tried + again on its next event or reconcile pass. That is the bound (CL-11587). A + live manager waiting for CI + with a reviewer out is parked: the reconciler does not count it toward the tenant's in-flight slots, which only managers judging or reviewing hold. Once the passes end, fail or run out, or take finds another run holds the claim, the run's last step stops it through the stop-run route under a diff --git a/docs/BITBUCKET_DATACENTER.md b/docs/BITBUCKET_DATACENTER.md index 305b1db6b..30c13995c 100644 --- a/docs/BITBUCKET_DATACENTER.md +++ b/docs/BITBUCKET_DATACENTER.md @@ -168,9 +168,9 @@ and the hex HMAC of the raw body, in `X-Hub-Signature`, with the event in `pr:comment:added`, `pr:merged`, `pr:declined`, `pr:deleted` - Repository: `repo:refs_changed` -Data Center sends no webhook when a build finishes, so the manager of a head -waiting for CI reads its build statuses again every 5 minutes and decides once -they are no longer in progress. While it waits, its `Corbits Triage` build +Data Center sends no webhook when a build finishes, so for a head waiting for +CI the hub's reconcile pass reads its build statuses every 5 minutes and wakes +a manager once they are no longer in progress. While it waits, its `Corbits Triage` build status reads `INPROGRESS` "Triage: Waiting for CI" where Comment is on. A verdict left unsure or failed by `ci-green` is re-read on the unconfirmed retry interval (5 minutes, doubling) and queued again once its builds changed (`triage_requeued` with `reason: build_status_changed`). diff --git a/docs/Corbits Triage User Manual/Corbits Triage User Manual.md b/docs/Corbits Triage User Manual/Corbits Triage User Manual.md index 24ac96b34..a29ca05e1 100644 --- a/docs/Corbits Triage User Manual/Corbits Triage User Manual.md +++ b/docs/Corbits Triage User Manual/Corbits Triage User Manual.md @@ -354,7 +354,7 @@ When Triage may not post its verdict, the pull request's line says so, such as * Which actions Triage takes on its own is set once for the whole workspace, under [Settings → Triage](#actions-triage-takes-on-its-own). In a new workspace Triage takes every action the policy file names, a few minutes after triage is turned on, and an action taken out of the file goes back to waiting for you. A workspace set to ask first posts only the verdict on its own, Comment and Add labels (Comment only on Bitbucket, which has no labels), so you can watch what Triage suggests before you let it act; **Always, in this repository** on a held action lets Triage take it on its own there from then on. Turning triage off, or removing the repository, takes back everything Triage did there on its own; turning it on again starts from Comment and Add labels. Two actions are always held for a person: Close on Unsure, and Merge on anything but Pass. -Triage checks a pull request as soon as it changes. When a rule is waiting on the host (CI still running, or the host has not yet worked out whether the pull request can merge), whatever else it could not confirm, or CI is still running and a step reads it (a CI condition, approve or merge), it gives no verdict yet and waits for the host, so you get one result instead of an early Unsure: no action runs, and Triage shows **In progress: Waiting on CI**, and the pull request waits under **Waiting** rather than **Needs you**. The pull request's page notes **New commits since Triage checked** above the earlier result, which stays until the new one lands. On GitHub, where Triage adds labels on its own, a pull request labelled **Triage: Not Ready** or **Triage: Needs Human Review** keeps it, so new commits change no label unless the verdict changes; any other, **Triage: Ready** included, shows **Triage: Waiting for CI**, or no Triage label while the commit has no checks. On Bitbucket, where Triage comments on its own, the new commit's **Corbits Triage** build status shows **Triage: Waiting for CI**. The verdict replaces it once CI finishes, as soon as any check fails, or after an hour at most. When the **CI must pass** rule is on, a commit with no checks two minutes after Triage first saw it is taken as having no CI. **Triage again**, or the host still not having worked out mergeability after Triage's last retry, gives the verdict at once. Where Triage comments on its own, its comment on the pull request reads **Triage: In progress** while it waits, and names what it is waiting for: CI, the decision model's answer or code review. Triage keeps that one comment current, and the verdict replaces it in place once it decides. If Triage is still waiting three hours after it started on the pull request, it gives the verdict anyway, says what it stopped waiting for, and runs no action on its own, so you decide. If Triage gives the verdict while CI is still running, on **Triage again** or its last retry, it holds every action that depends on CI, and its comment on the pull request lists them under **Waiting for checks to finish**. +Triage checks a pull request as soon as it changes. When a rule is waiting on the host (CI still running, or the host has not yet worked out whether the pull request can merge), whatever else it could not confirm, or CI is still running and a step reads it (a CI condition, approve or merge), it gives no verdict yet and waits for the host, so you get one result instead of an early Unsure: no action runs, and Triage shows **In progress: Waiting on CI**, and the pull request waits under **Waiting** rather than **Needs you**. The pull request's page notes **New commits since Triage checked** above the earlier result, which stays until the new one lands. On GitHub, where Triage adds labels on its own, a pull request labelled **Triage: Not Ready** or **Triage: Needs Human Review** keeps it, so new commits change no label unless the verdict changes; any other, **Triage: Ready** included, shows **Triage: Waiting for CI**, or no Triage label while the commit has no checks. On Bitbucket, where Triage comments on its own, the new commit's **Corbits Triage** build status shows **Triage: Waiting for CI**. The verdict replaces it once CI finishes, as soon as any check fails, or after an hour at most; on a host that does not tell Triage when a build finishes, such as Bitbucket, within about five minutes of it finishing. When the **CI must pass** rule is on, a commit still with no checks a few minutes after Triage first saw it is taken as having no CI. **Triage again**, or the host still not having worked out mergeability after Triage's last retry, gives the verdict at once. Where Triage comments on its own, its comment on the pull request reads **Triage: In progress** while it waits, and names what it is waiting for: CI, the decision model's answer or code review. Triage keeps that one comment current, and the verdict replaces it in place once it decides. If the decision model or code review keep Triage waiting three hours after it started on the pull request, it gives the verdict anyway, says what it stopped waiting for, and runs no action on its own, so you decide. If Triage gives the verdict while CI is still running, on **Triage again** or its last retry, it holds every action that depends on CI, and its comment on the pull request lists them under **Waiting for checks to finish**. An **Unsure** comment lists each rule Triage could not confirm, with checks still running or cancelled under **Waiting on**. diff --git a/docs/PRODUCT.md b/docs/PRODUCT.md index 9b5fcdf9a..66e2ae4a4 100644 --- a/docs/PRODUCT.md +++ b/docs/PRODUCT.md @@ -245,9 +245,9 @@ the verdict waits for CI, the decision model or code review, that comment reads **Triage: In progress** and names what it waits for, such as "Triage is waiting for the decision model's answer and code review before it decides". It is rewritten when what it waits for changes, and the verdict replaces it in -place. If Triage is still waiting three hours after it started on the pull -request, it gives the verdict anyway, says what it stopped waiting for, and -runs no action on its own. +place. If the decision model or code review keep Triage waiting three hours +after it started on the pull request, it gives the verdict anyway, says what it +stopped waiting for, and runs no action on its own. ### Waiting for CI diff --git a/packages/triage-contracts/src/triage-state.ts b/packages/triage-contracts/src/triage-state.ts index 2080b2cb2..c272456c2 100644 --- a/packages/triage-contracts/src/triage-state.ts +++ b/packages/triage-contracts/src/triage-state.ts @@ -6,6 +6,15 @@ export const TRIAGE_STATE_SCHEMA_VERSION = 1; /** A pr-triage run still going after this long lost its sidecar: three agent steps on either gate branch of up to fifteen minutes each, plus their retries. */ export const PR_TRIAGE_STUCK_RUN_MS = 90 * 60_000; +/** How long a head waits for its CI, from its first pass that waited for it, before it is decided without it. */ +export const CI_WAIT_MS = 60 * 60_000; + +/** How long a head with no checks yet waits for its CI to start before it is taken as having none. */ +export const CI_START_MS = 2 * 60_000; + +/** How long a listener's wake mail waits for a just-deployed manager to reach a sidecar before it fails. */ +export const MANAGER_READY_MS = 2 * 60_000; + /** `stale`: the base branch moved after the head's verdict, so it is due again. */ export const PR_TRIAGE_STATUSES = ["new", "queued", "running", "triaged", "failed", "stale"] as const; export type PrTriageStatus = (typeof PR_TRIAGE_STATUSES)[number]; diff --git a/packages/triage-manager/README.md b/packages/triage-manager/README.md index 5b240d04e..c19bb3244 100644 --- a/packages/triage-manager/README.md +++ b/packages/triage-manager/README.md @@ -4,19 +4,22 @@ A workspace's listener and a pull request's manager. - **Listener** (`triageListener`, `triage-listener-`): the hub mails it the pull requests it recorded. For each, it deploys a manager through Interchange's deploy route with its run token and wakes it with a mail that names only the - pull request. A manager whose wake mail fails is cancelled, and the pull request's next event, or the reconciler - re-queuing its head, deploys another. + pull request. A manager whose wake mail fails is cancelled and one more is deployed and woken in its place; if that + mail fails too, it is cancelled, and the pull request's next event, or the reconciler re-queuing its head, deploys + another. - **Manager** (`triageManager`, `triage-manager-`): one live run per pull request, which lives until its head has its final verdict, and the only workflow that writes to the pull request. A mail starts a loop of passes: `take` claims the pull request and its pending events; each held Do approved since runs as pr-triage runs an approval; then pr-triage's own steps run on the newest mail the hub recorded for the head. A policy of only deterministic rules posts on that pass. One whose rules need the judge, or that asks for code review, posts once, after each of those has reported or given - up. The next pass runs at once while an event is pending. While the head waits for CI, or for a judge or code review - it mailed, the next runs 30 seconds later, taking what arrived meanwhile: a check finishing, a reviewer's result, or a - new head, whose pass starts over. Meanwhile the one triage comment reads In progress and names what it waits for; the - verdict replaces it in place. A head that waited an hour for CI is decided anyway, and three hours after its claim a - manager stops waiting: it decides, says what it stopped waiting for and runs no action on its own. The manager ends once its head - has its final verdict and the policy's actions ran or were held, so an idle pull request has no manager. With a + up. The next pass runs at once while an event is pending. While a judge or code review it mailed is out and within its + deadline, the next runs 30 seconds later, taking what arrived meanwhile: a check finishing, a reviewer's result, or a + new head, whose pass starts over. A head that waits only on CI posts that it waits and its manager ends; the next + check event, or the hub's reconcile pass once CI settles, the start grace passes or the hour is up, wakes a fresh + manager for it. Meanwhile the one triage comment reads In progress and names what it waits for; the verdict replaces + it in place. A head that waited an hour for CI is decided anyway, and a manager its reviewers keep alive three hours + after its claim stops waiting: it decides, says what it stopped waiting for and runs no action on its own. The manager + ends once its head has its final verdict and the policy's actions ran or were held, so an idle pull request has no manager. With a published judge, a pass deploys [`@corbits/triage-judge`](../triage-judge) once the host's checks have finished and mails it the rules output; while they run the head waits. The pass that collects its report posts the judge's verdict while the rules still need the judge, its judgement as the judge made it and its Dos made again from the diff --git a/packages/triage-manager/src/actions/hub.ts b/packages/triage-manager/src/actions/hub.ts index 09dd2f2c2..e4d7f5eb7 100644 --- a/packages/triage-manager/src/actions/hub.ts +++ b/packages/triage-manager/src/actions/hub.ts @@ -32,9 +32,14 @@ export interface ManagerMail { export function managerMailOf(payload: unknown): ManagerMail { const { request, runId } = mailOf(payload); + return { ...managedBy(request), runId }; +} + +/** The tenant and pull request a manager's mail, as `manageMail` writes it, names. */ +export function managedBy(request: Record): Omit { const { kind, tenantId, repo, prNumber, repoSource } = request; if (kind !== MANAGE_KIND || typeof tenantId !== "string" || typeof repo !== "string" || typeof prNumber !== "number") throw new Error("the mail names no tenant, repository or pull request"); - return { tenantId, runId, pullRequest: { source: repoSourceOf(repoSource), repo, number: prNumber } }; + return { tenantId, pullRequest: { source: repoSourceOf(repoSource), repo, number: prNumber } }; } /** The mail that wakes the manager of `pullRequest`. */ diff --git a/packages/triage-manager/src/actions/index.ts b/packages/triage-manager/src/actions/index.ts index cdf4c9643..5a3bd70a5 100644 --- a/packages/triage-manager/src/actions/index.ts +++ b/packages/triage-manager/src/actions/index.ts @@ -1,6 +1,6 @@ export { execute, facts, mirror, rules, unready } from "@corbits/triage-workflows/actions/index"; export { executeApprovals } from "./approvals.js"; -export { forward, unwoken } from "./listener.js"; +export { forward, retry, unwoken, unwokenAgain } from "./listener.js"; export { deployJudge, judgeMailed, judgeScope, verdict } from "./judge.js"; export { done, stop, take } from "./manager.js"; export { cleared } from "./review-clean.js"; diff --git a/packages/triage-manager/src/actions/listener.ts b/packages/triage-manager/src/actions/listener.ts index f6f0c13fd..719b858bd 100644 --- a/packages/triage-manager/src/actions/listener.ts +++ b/packages/triage-manager/src/actions/listener.ts @@ -1,12 +1,9 @@ import type { EffectContext } from "@intx/workflow"; import { type } from "arktype"; -import { isFromTriageSender, repoSourceOf, TRIAGE_LISTENERS } from "@corbits/triage-contracts"; +import { isFromTriageSender, MANAGER_READY_MS, repoSourceOf, TRIAGE_LISTENERS } from "@corbits/triage-contracts"; import { Mailed, MailInput } from "../mail.js"; import { publishedFor } from "../release.js"; -import { deploy, endRun, mailOf, manageMail } from "./hub.js"; - -/** How long a just-deployed manager has to reach a sidecar before its first mail fails. */ -const MANAGER_READY_MS = 2 * 60_000; +import { deploy, endRun, mailOf, managedBy, manageMail } from "./hub.js"; /** The hub's mail to a listener: the repository and, as a reconciler's mail does, the heads it names as `items`. */ const ListenerMail = type({ @@ -40,18 +37,53 @@ export async function forward(input: unknown, ctx: EffectContext, _signal: Abort return { mails }; } -const UnwokenInput = type({ forward: { output: { mails: MailInput.array() } }, mail: { output: type({ reply: "string" }).array() } }); +const Replies = type({ reply: "string" }).array(); +const Cancelled = type({ runId: "string > 0", reason: "string" }).array(); +const UnwokenInput = type({ forward: { output: { mails: MailInput.array() } }, mail: { output: Replies } }); +const RetryInput = type({ forward: { output: { mails: MailInput.array() } }, unwoken: { output: { cancelled: Cancelled } } }); +const RetriedInput = type({ retry: { output: { mails: MailInput.array() } }, remail: { output: Replies } }); /** * Cancels each manager this body deployed whose wake mail did not go, through the stock stop-run route, so a manager - * that was never woken does not stay deployed; the output names each with why its mail failed. The pull request's next - * event, or the reconciler re-queuing its head, deploys a fresh one. + * that was never woken does not stay deployed; the output names each with why its mail failed. */ export async function unwoken(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ cancelled: { runId: string; reason: string }[] }> { const parsed = UnwokenInput(input); if (parsed instanceof type.errors) throw new Error(`unwoken: ${parsed.summary}`); - const { mails } = parsed.forward.output; - const replies = parsed.mail.output; + return cancelUnwoken(ctx, parsed.forward.output.mails, parsed.mail.output); +} + +/** + * Deploys one fresh manager for each pull request whose manager `unwoken` cancelled, as a sidecar slow to start makes + * the first wake fail, and wakes it with the same mail. + */ +export async function retry(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ mails: MailInput[] }> { + const parsed = RetryInput(input); + if (parsed instanceof type.errors) throw new Error(`retry: ${parsed.summary}`); + const cancelled = new Set(parsed.unwoken.output.cancelled.map((run) => run.runId)); + const mails: MailInput[] = []; + for (const mail of parsed.forward.output.mails.filter((sent) => cancelled.has(sent.to))) { + const { tenantId, pullRequest } = managedBy(JSON.parse(mail.content) as Record); + const source = repoSourceOf(pullRequest.source); + const release = publishedFor(source, "manager"); + if (!release) throw new Error(`${TRIAGE_LISTENERS[source]} needs the release's manager in release.json, which the hub publishes beside it`); + const runId = await deploy(ctx, `redeploy:${pullRequest.number}`, tenantId, release); + mails.push({ ...mail, to: runId, readyBy: new Date(Date.now() + MANAGER_READY_MS).toISOString() }); + } + return { mails }; +} + +/** + * Cancels each manager `retry` deployed whose wake mail did not go either, and deploys no other: the hub's start grace + * on the claim lapses, and the pull request's next event or the reconciler re-queuing its head asks again. + */ +export async function unwokenAgain(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ cancelled: { runId: string; reason: string }[] }> { + const parsed = RetriedInput(input); + if (parsed instanceof type.errors) throw new Error(`unwokenAgain: ${parsed.summary}`); + return cancelUnwoken(ctx, parsed.retry.output.mails, parsed.remail.output); +} + +async function cancelUnwoken(ctx: EffectContext, mails: readonly MailInput[], replies: readonly { reply: string }[]): Promise<{ cancelled: { runId: string; reason: string }[] }> { if (replies.length !== mails.length) throw new Error(`unwoken: ${replies.length} mail replies for ${mails.length} managers`); const cancelled: { runId: string; reason: string }[] = []; for (const [index, mail] of mails.entries()) { diff --git a/packages/triage-manager/src/actions/manager.ts b/packages/triage-manager/src/actions/manager.ts index 52e77e856..d26ca873f 100644 --- a/packages/triage-manager/src/actions/manager.ts +++ b/packages/triage-manager/src/actions/manager.ts @@ -1,6 +1,6 @@ import type { EffectContext } from "@intx/workflow"; import { type, type Type } from "arktype"; -import { APPROVE_DO_KIND, CodeReviewOutput, JudgeVerdict, StageResult, VerdictFacts } from "@corbits/triage-contracts"; +import { APPROVE_DO_KIND, CI_START_MS, CI_WAIT_MS, CodeReviewOutput, JudgeVerdict, StageResult, VerdictFacts } from "@corbits/triage-contracts"; import type { Item } from "@corbits/triage-workflows/logic/item"; import { hostAwaited, type Awaits } from "@corbits/triage-workflows/logic/render"; import { decided } from "@corbits/triage-workflows/logic/settled"; @@ -15,27 +15,13 @@ const Latest = { "request?": "Record", reviewers: Reviewer.arra const Taken = type({ outcome: "'idle'", ...Latest }).or({ outcome: "'pass'", pass: { head: "string" }, events: TakenEvent.array(), ...Latest }); const Done = type({ outcome: "'ended'" }).or({ outcome: "'more'", wait: "boolean" }); -/** How long a manager lives from claiming its pull request, time for CI, the judge and code review to each take their hour; a new head does not extend it. */ +/** + * How long a manager lives from claiming its pull request, time for the judge and code review to each take their hour + * with CI in between; a new head does not extend it. A manager ends while it waits only on CI, so this bounds one its + * reviewers keep alive. + */ export const LIFETIME_MS = 3 * 60 * 60_000; -/** How long a manager waits for its head's CI, from its first pass that waited for it, before it decides without it. */ -const CI_WAIT_MS = 60 * 60_000; - -/** How often a manager reads its waiting head's CI again itself, for a host whose builds send no event when they finish. */ -const RECHECK_MS = 5 * 60_000; - -/** How long a head with no checks yet waits for its CI to start before it is taken as having none. */ -const CI_START_MS = 2 * 60_000; - -/** Polls come about 30 seconds apart, so a window of a minute holds each recheck boundary at least once. */ -const RECHECK_WINDOW_MS = 60_000; - -/** Whether the head, waiting for CI `elapsed` milliseconds, reached its CI start grace or a recheck since about the last poll. */ -function rechecks(elapsed: number): boolean { - const before = Math.max(0, elapsed - RECHECK_WINDOW_MS); - return Math.floor(elapsed / RECHECK_MS) > Math.floor(before / RECHECK_MS) || (before < CI_START_MS && elapsed >= CI_START_MS); -} - /** * What a pass works on. `run` when there is anything to do: a pass, a reviewer the manager waits for is past its * deadline, or its head waited out its CI. `approvals` are the held Dos approved since, each run as pr-triage runs one. @@ -158,14 +144,13 @@ export async function take(input: unknown, ctx: EffectContext, _signal: AbortSig const overdue = late(reviewers, head, "code-review", report.review !== undefined || report.reviewFailed !== undefined) || late(reviewers, head, "judge", report.judge !== undefined || report.judgeFailed !== undefined); const waited = waitingSince === undefined ? undefined : Date.now() - Date.parse(waitingSince); const waitedOut = stopped || (waited !== undefined && waited >= CI_WAIT_MS); - const recheck = waited !== undefined && !waitedOut && rechecks(waited); // A pass of approvals alone, before any pull request mail was recorded, has nothing to triage. - const triage = request !== undefined && (forHead !== undefined || report.review !== undefined || report.reviewFailed !== undefined || report.judge !== undefined || report.judgeFailed !== undefined || overdue || waitedOut || recheck); + const triage = request !== undefined && (forHead !== undefined || report.review !== undefined || report.reviewFailed !== undefined || report.judge !== undefined || report.judgeFailed !== undefined || overdue || waitedOut); const starting = headSince === undefined || Date.now() - Date.parse(headSince) < CI_START_MS; const taken = request && { ...request, ...(waitedOut && { noWait: true }), ...(!waitedOut && starting && { ciStarting: true }) }; const verdict = report.judge ?? storedVerdict(reviewers, head); const waiting = waitingSince !== undefined && !waitedOut; - return { run: answer.outcome === "pass" || overdue || waitedOut || recheck, approvals, triage, ...(taken && { taken }), head, ...report, ...(verdict && { verdict }), overdue, reviewers, ...(waiting && { waitingSince }), ...(stopped && { stopped: outstanding }) }; + return { run: answer.outcome === "pass" || overdue || waitedOut, approvals, triage, ...(taken && { taken }), head, ...report, ...(verdict && { verdict }), overdue, reviewers, ...(waiting && { waitingSince }), ...(stopped && { stopped: outstanding }) }; } const Output = { output: "Record" } as const; @@ -262,21 +247,19 @@ function abandoned({ head, reviewers }: TakeOutput, runId: string): Reviewer[] { } /** - * Whether the pass's head still waits for something that comes to this manager: a reviewer of it still out, or CI, as - * the verdict it posted waits for the host's checks, or, on a pass that triaged nothing, as the hub says it does. + * Whether the pass's head still waits for a reviewer of its own, out and within its deadline, whose result comes to this + * manager. A head that waits only on the host's checks does not keep it: the next host event, or the reconciler once + * CI settles, wakes a manager for it. */ -function awaiting({ take, cleared }: DoneInput, report: Record | undefined, reviewers: readonly Reviewer[]): boolean { - const { head, waitingSince, stopped } = take.output; - if (stopped) return false; - if (reviewers.some((reviewer) => reviewer.head === head && reviewer.result === undefined)) return true; - if (report === undefined) return waitingSince !== undefined; - return report["waiting"] === true || cleared?.output["waiting"] === true; +function awaiting(take: TakeOutput, reviewers: readonly Reviewer[]): boolean { + if (take.stopped) return false; + return reviewers.some((reviewer) => reviewer.head === take.head && reviewer.result === undefined && Date.parse(reviewer.deadline) > Date.now()); } /** - * Closes the pass with what it ran and stops the reviewers it no longer waits for. A pass whose head still awaits CI or - * a reviewer asks the hub to `wait`. `more` runs the next pass at once, or with `wait` after polling for what it awaits; - * `ended` ends the run once the head has its final verdict, and a later event starts the next one. + * Closes the pass with what it ran and stops the reviewers it no longer waits for. A pass whose head still awaits a + * reviewer asks the hub to `wait`. `more` runs the next pass at once, or with `wait` after polling for the reviewer; + * `ended` ends the run once the head has its final verdict or waits only on CI, and a later event starts the next one. */ export async function done(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ outcome: "ended" | "refused" } | { outcome: "more"; wait: boolean }> { const parsed = DoneInput(input); @@ -286,7 +269,7 @@ export async function done(input: unknown, ctx: EffectContext, _signal: AbortSig const mailed = [parsed.judgeMailed, parsed.earlyReviewMailed, parsed.reviewMailed].flatMap((step) => (step?.output.reviewer ? [step.output.reviewer] : [])); const report = doneReport(parsed); const reviewers = reviewersAfter(parsed.take.output, mailed); - const wait = awaiting(parsed, report, reviewers); + const wait = awaiting(parsed.take.output, reviewers); const body = { ...mail.pullRequest, reviewers, wait, ...(report && { report }), ...(parsed.approvals && { approvals: parsed.approvals.output }) }; const answer = await asManager(ctx, mail, "done", body, Done); if (answer === undefined) return { outcome: "refused" }; diff --git a/packages/triage-manager/src/triage-listener.ts b/packages/triage-manager/src/triage-listener.ts index 4dec8ef8c..6c500644d 100644 --- a/packages/triage-manager/src/triage-listener.ts +++ b/packages/triage-manager/src/triage-listener.ts @@ -10,7 +10,8 @@ import { deployGrants, type Release } from "./release.js"; /** * A workspace's listener for one host under `id`, its mail address, which the hub mails the pull requests it records: * it deploys each one's manager from the host's published release and wakes it, and cancels a manager whose wake mail - * did not go, so it is not left deployed. With `manager`, the release published as the definition loads, it declares the + * did not go, so it is not left deployed, then deploys and wakes one more for that pull request, cancelled in turn if + * its mail fails too. With `manager`, the release published as the definition loads, it declares the * grants to deploy that asset; without, none, and its body fails. It writes nothing to the pull request and holds no * grant on its state; a manager whose take finds another live one stops itself. */ @@ -33,6 +34,21 @@ export function triageListener({ id, manager }: { id: string; manager: Release | credentials: [HUB_CREDENTIAL_HANDLE], effect: { requires: [HUB_WRITE] }, }), + retry: action({ + handler: "retry", + input: { project: { from: "steps" }, fields: ["forward", "unwoken"] }, + after: ["unwoken"], + credentials: [HUB_CREDENTIAL_HANDLE], + effect: { requires: [HUB_WRITE] }, + }), + remail: map({ over: { from: "steps.retry.output.mails" }, step: step({ agent: mailAgentOf(mailDirector, judgeAgent.inference, { tolerate: true }), input: { from: "trigger.payload" } }), after: ["retry"] }), + unwokenAgain: action({ + handler: "unwokenAgain", + input: { project: { from: "steps" }, fields: ["retry", "remail"] }, + after: ["remail"], + credentials: [HUB_CREDENTIAL_HANDLE], + effect: { requires: [HUB_WRITE] }, + }), }, }); return defineWorkflow({ diff --git a/packages/triage-manager/src/triage-manager.test.ts b/packages/triage-manager/src/triage-manager.test.ts index a974cb8b2..321c2c97f 100644 --- a/packages/triage-manager/src/triage-manager.test.ts +++ b/packages/triage-manager/src/triage-manager.test.ts @@ -88,7 +88,8 @@ type HubCall = { method: string; path: string; body?: unknown }; type HubAnswer = Record | typeof REFUSED; /** The hub's manager routes, answering each take and done in turn, its deploy route and its stop-run route. */ -function fakeHub({ takes, dones = [] }: { takes: HubAnswer[]; dones?: HubAnswer[] }) { +/** `deployed`, the run ids the deploy route answers in turn, each after the last answering again. */ +function fakeHub({ takes, dones = [], deployed = [DEPLOYED] }: { takes: HubAnswer[]; dones?: HubAnswer[]; deployed?: string[] }) { const calls: HubCall[] = []; function answer(next: HubAnswer | undefined): Response { if (next === undefined) throw new Error("the hub was asked more than the test expects"); @@ -100,7 +101,7 @@ function fakeHub({ takes, dones = [] }: { takes: HubAnswer[]; dones?: HubAnswer[ if (url.pathname === `${TENANT_PATH}/manager/take`) return answer(takes.shift()); if (url.pathname === `${TENANT_PATH}/manager/done`) return answer(dones.shift()); if (method === "DELETE" && url.pathname.startsWith(`${TENANT_PATH}/workflows/runs/`)) return new Response(null, { status: 202 }); - if (method === DEPLOY.method && url.pathname === DEPLOY.path) return Response.json({ id: DEPLOYED }, { status: 201 }); + if (method === DEPLOY.method && url.pathname === DEPLOY.path) return Response.json({ id: deployed.length > 1 ? deployed.shift() : deployed[0] }, { status: 201 }); return Response.json({ error: { code: "not_found" } }, { status: 404 }); } return { hub, calls }; @@ -108,7 +109,7 @@ function fakeHub({ takes, dones = [] }: { takes: HubAnswer[]; dones?: HubAnswer[ /** The host's credentials as triage-workflows wires them, plus the hub credential deploy binds to the hub actions. */ function actionCredentials(bindings: Bindings, host: Fetch, hub: (url: URL, init?: RequestInit) => Promise) { - const consumers = ["take", "done", "deployReview", "deployJudge", "forward", "unwoken", "stop"].map(actionConsumer); + const consumers = ["take", "done", "deployReview", "deployJudge", "forward", "unwoken", "retry", "unwokenAgain", "stop"].map(actionConsumer); const cell = delivered(bindings, [{ credentialId: "hub", providerKey: "http", origin: HUB_ORIGIN, secret: "run-token" }]); cell.current = { ...cell.current, bindings: [...cell.current.bindings, ...consumers.map((consumer) => ({ handle: HUB_CREDENTIAL_HANDLE, credentialId: "hub", consumer }))] }; const hubGrants: GrantRule[] = consumers.map((tool, index) => ({ id: `hub-${index}`, resource: "credential:hub", action: "use", effect: "allow", origin: "system", conditions: { tool }, expiresAt: null, roleId: null, principalId: null })); @@ -134,7 +135,8 @@ function hurried() { * Runs one mail's body; the mail agent's director runs its one tool, so the mail it would send is recorded and answered * `mailed` in the `{ reply, turn }` a live agent step outputs. */ -async function runMail(definition: WorkflowDefinition, mail: unknown, host: Fetch, hub: ReturnType["hub"], mailed: unknown[] = [], delivers = true, clock?: () => Date) { +/** `delivers`, whether every wake mail goes, or whether each in turn does. */ +async function runMail(definition: WorkflowDefinition, mail: unknown, host: Fetch, hub: ReturnType["hub"], mailed: unknown[] = [], delivers: boolean | readonly boolean[] = true, clock?: () => Date) { const body = (definition.steps.events as OnTriggerPrimitive).body; if (!("inline" in body)) throw new Error("the body is not inline"); const { terminalStatus, outputs } = await runLocal(body.inline, { @@ -143,7 +145,8 @@ async function runMail(definition: WorkflowDefinition, mail: unknown, host: Fetc invokeStep: async function mailAgent({ agent, input }) { if (!agent.id.startsWith("triage-mail")) throw new Error(`no agent but the mail agent runs, not ${agent.id}`); mailed.push(input); - const reply = JSON.stringify(delivers ? { mailed: true } : { mailed: false, error: "public key was not published by its readyBy" }); + const goes = typeof delivers === "boolean" ? delivers : delivers[mailed.length - 1]; + const reply = JSON.stringify(goes ? { mailed: true } : { mailed: false, error: "public key was not published by its readyBy" }); return { output: { reply, turn: { role: "assistant", content: [{ type: "text", text: reply }] } } }; }, actionResolver: (ref) => (actions as Record)[ref]!, @@ -263,14 +266,12 @@ describe("triage-manager", () => { expect(doneBody(hub.calls)["wait"]).toBe(false); }); - test("a head whose CI still runs gets the Waiting label, and the manager polls on while the hub says it waits", async () => { + test("a head whose CI still runs, with no reviewer out, gets the Waiting label and the in-progress comment, and its manager ends", async () => { const { writes, host } = githubHost(HEAD, { checkRuns: [{ name: "build", status: "in_progress", conclusion: null }] }); - const waiting = { ...IDLE, waitingSince: new Date().toISOString() }; - const hub = fakeHub({ takes: [PASS, waiting], dones: [{ outcome: "more", wait: true }, ENDED] }); - const { passes, ran } = await runMail(github, wake(), host, hub.hub, [], true, hurried()); - expect(passes).toBe(2); - expect(ran).toEqual(["approvals", "done", "paused", "poll", "ran", "take"]); - expect(hub.calls.filter((call) => call.path === DONE.path).map((call) => (call.body as { wait: boolean }).wait)).toEqual([true, true]); + const hub = fakeHub({ takes: [PASS], dones: [ENDED] }); + const { passes } = await runMail(github, wake(), host, hub.hub); + expect(passes).toBe(1); + expect(doneBody(hub.calls)).toMatchObject({ wait: false, report: { head: HEAD, verdict: expect.objectContaining({ waiting: true }) } }); expect(writes).toContainEqual(expect.objectContaining({ method: "POST", path: `/repos/${PR.repo}/issues/${PR.number}/labels`, body: { labels: [WAITING_LABEL] } })); expect(writes.filter((write) => write.path.endsWith("/comments"))).toEqual([inProgress("CI to finish")]); }); @@ -310,23 +311,30 @@ describe("triage-manager", () => { expect(((outputs["cleared"] as { actions: { held?: true }[] }).actions).every((action) => action.held === true && !("ci" in action))).toBe(true); }); - test("a new head with no checks yet waits two minutes for its CI to start, then is read again and decided as having none", async () => { + test("a new head with no checks yet waits for its CI to start and its manager ends; one woken past the start grace decides it as having none", async () => { const { writes, host } = githubHost(); const fresh = { ...PASS, headSince: undefined }; const first = fakeHub({ takes: [fresh], dones: [ENDED] }); await runMail(github, wake(), host, first.hub); - expect(doneBody(first.calls)).toMatchObject({ wait: true, report: { head: HEAD, verdict: expect.objectContaining({ waiting: true, reason: expect.stringContaining("checks have not started yet") }) } }); + expect(doneBody(first.calls)).toMatchObject({ wait: false, report: { head: HEAD, verdict: expect.objectContaining({ waiting: true, reason: expect.stringContaining("checks have not started yet") }) } }); expect(writes).toContainEqual(expect.objectContaining({ method: "POST", path: `/repos/${PR.repo}/issues/${PR.number}/labels`, body: { labels: [WAITING_LABEL] } })); const since = new Date(Date.now() - 2 * 60_000 - 10_000).toISOString(); - const later = fakeHub({ takes: [{ ...IDLE, waitingSince: since, headSince: since }], dones: [ENDED] }); + const later = fakeHub({ takes: [{ ...PASS, waitingSince: since, headSince: since }], dones: [ENDED] }); await runMail(github, wake(), host, later.hub); expect(doneBody(later.calls)).toMatchObject({ wait: false, report: { head: HEAD, verdict: expect.objectContaining({ verdict: "pass" }) } }); }); - test("a head waiting for CI is read again every five minutes without an event, so CI no host event reports still settles it", async () => { + test("an idle manager whose head waits on CI, with no reviewer out, triages nothing and ends", async () => { + const { host } = githubHost(); + const hub = fakeHub({ takes: [{ ...IDLE, waitingSince: new Date(Date.now() - 5 * 60_000).toISOString() }], dones: [ENDED] }); + expect((await runMail(github, wake(), host, hub.hub)).ran).not.toContain("rules"); + expect(doneBody(hub.calls)["wait"]).toBe(false); + }); + + test("a manager woken for a head that waited on CI, once CI finished, decides it", async () => { const { writes, host } = githubHost(); const waitingSince = new Date(Date.now() - 5 * 60_000 - 10_000).toISOString(); - const hub = fakeHub({ takes: [{ ...IDLE, waitingSince }], dones: [ENDED] }); + const hub = fakeHub({ takes: [{ ...PASS, waitingSince }], dones: [ENDED] }); const { ran } = await runMail(github, wake(), host, hub.hub); expect(ran).toEqual(STAGE_1); expect(doneBody(hub.calls)).toMatchObject({ wait: false, report: { head: HEAD, verdict: expect.objectContaining({ verdict: "pass" }) } }); @@ -806,6 +814,16 @@ describe("triage-manager with a judge", () => { expect(mailed).toHaveLength(1); }); + test("a manager whose judge is still out within its deadline stays up and polls, even while CI runs", async () => { + const { host } = githubHost(HEAD, { checkRuns: [{ name: "build", status: "in_progress", conclusion: null }] }); + const judging = { ...IDLE, request: hubRequest(JUDGED), reviewers: [JUDGE] }; + const hub = fakeHub({ takes: [judging, judging], dones: [{ outcome: "more", wait: true }, ENDED] }); + const { passes, ran } = await runMail(github, wake(), host, hub.hub, [], true, hurried()); + expect(passes).toBe(2); + expect(ran).toContain("poll"); + expect(hub.calls.filter((call) => call.path === DONE.path).map((call) => (call.body as { wait: boolean }).wait)).toEqual([true, true]); + }); + test("a pass that mailed its judge and answers more runs the next pass, which ends", async () => { const { host } = githubHost(); const hub = fakeHub({ takes: [{ ...PASS, request: hubRequest(JUDGED) }, IDLE], dones: [MORE, ENDED] }); @@ -864,7 +882,7 @@ describe("triage-manager with a judge", () => { expect(green.writes).toContainEqual(POSTED_REVIEW); }); - test("a stored verdict on a head whose CI runs again on the same commit waits for CI instead of being posted, and the manager does not end", async () => { + test("a stored verdict on a head whose CI runs again on the same commit waits for CI instead of being posted, and the manager ends", async () => { const pack: CheckPack = { ...JUDGED, actions: { ...JUDGED.actions, pass: [{ do: "approve" }] } }; const request = hubRequest(pack); const verdict = await judgeVerdict(JSON.parse((await firstJudgeMail(request)).content).rules); @@ -875,7 +893,20 @@ describe("triage-manager with a judge", () => { expect(outputs["evaluateRules"]).toMatchObject({ waiting: true }); expect(writes).not.toContainEqual(JUDGE_POSTED); expect(writes).not.toContainEqual(POSTED_REVIEW); - expect(doneBody(hub.calls)["wait"]).toBe(true); + expect(doneBody(hub.calls)).toMatchObject({ wait: false, reviewers: [{ ...JUDGE, result: verdict }] }); + }); + + test("a judge's report on a head whose CI started running again posts the waiting state and ends, keeping the judgement for the pass after CI", async () => { + const pack: CheckPack = { ...JUDGED, actions: { ...JUDGED.actions, pass: [{ do: "approve" }] } }; + const request = hubRequest(pack); + const verdict = await judgeVerdict(JSON.parse((await firstJudgeMail(request)).content).rules); + const { writes, host } = githubHost(HEAD, { checkRuns: [{ name: "ci", status: "in_progress", conclusion: null }] }); + const hub = fakeHub({ takes: [judgeReported(verdict, request)], dones: [ENDED] }); + const { outputs } = await runMail(github, wake(), host, hub.hub); + expect(outputs["evaluateRules"]).toMatchObject({ waiting: true }); + expect(writes).toContainEqual(inProgress("CI to finish")); + expect(writes).not.toContainEqual(JUDGE_POSTED); + expect(doneBody(hub.calls)).toMatchObject({ wait: false, reviewers: [{ ...JUDGE, result: verdict }] }); }); test("a manager whose clock runs out while CI runs on a judge policy deploys no judge, and posts its verdict with every Do held", async () => { @@ -1034,13 +1065,28 @@ describe("triage-listener", () => { ]); }); - test("a manager whose wake mail did not go is cancelled once, and the listener deploys no other for it", async () => { - const hub = fakeHub({ takes: [] }); + test("a manager whose wake mail did not go is cancelled, and the one the listener deploys in its place is woken", async () => { + const hub = fakeHub({ takes: [], deployed: [DEPLOYED, "run_again"] }); + const mailed: unknown[] = []; const mail = triggerMail(PR, RULES, { tenantId: TENANT, items: [{ prNumber: 8, headSha: HEAD }] }); - const { terminalStatus, outputs } = await runMail(listener, mail, githubHost().host, hub.hub, [], false); + const { terminalStatus, outputs } = await runMail(listener, mail, githubHost().host, hub.hub, mailed, [false, true]); expect(terminalStatus).toBe("completed"); - expect(hub.calls.map(({ method, path }) => ({ method, path }))).toEqual([DEPLOY, { method: "DELETE", path: `${TENANT_PATH}/workflows/runs/${DEPLOYED}` }]); + expect(hub.calls.map(({ method, path }) => ({ method, path }))).toEqual([DEPLOY, { method: "DELETE", path: `${TENANT_PATH}/workflows/runs/${DEPLOYED}` }, DEPLOY]); + expect((mailed as { to: string; content: string }[]).map((sent) => [sent.to, JSON.parse(sent.content)])).toEqual([ + [DEPLOYED, JSON.parse(manageMail(TENANT, PULL))], + ["run_again", JSON.parse(manageMail(TENANT, PULL))], + ]); expect(outputs["unwoken"]).toEqual({ cancelled: [{ runId: DEPLOYED, reason: expect.stringContaining("public key") }] }); + expect(outputs["unwokenAgain"]).toEqual({ cancelled: [] }); + }); + + test("a manager deployed in place of an unwoken one whose mail fails too is cancelled, and the listener deploys no third", async () => { + const hub = fakeHub({ takes: [], deployed: [DEPLOYED, "run_again"] }); + const mail = triggerMail(PR, RULES, { tenantId: TENANT, items: [{ prNumber: 8, headSha: HEAD }] }); + const { terminalStatus, outputs } = await runMail(listener, mail, githubHost().host, hub.hub, [], false); + expect(terminalStatus).toBe("completed"); + expect(hub.calls.map(({ method, path }) => ({ method, path }))).toEqual([DEPLOY, { method: "DELETE", path: `${TENANT_PATH}/workflows/runs/${DEPLOYED}` }, DEPLOY, { method: "DELETE", path: `${TENANT_PATH}/workflows/runs/run_again` }]); + expect(outputs["unwokenAgain"]).toEqual({ cancelled: [{ runId: "run_again", reason: expect.stringContaining("public key") }] }); }); test("a mail that names no pull request, or is not from the hub, fails instead of being dropped", async () => { diff --git a/tests/chain/chain.ts b/tests/chain/chain.ts index 9d438a2f4..167ad8a2f 100644 --- a/tests/chain/chain.ts +++ b/tests/chain/chain.ts @@ -161,7 +161,7 @@ export function chainStack(env: Record = {}) { if (value !== undefined) return value; await Bun.sleep(500); } - throw new Error(`Timed out waiting for ${what}.\nRuns:\n${await runEvents()}\nGitHub misses: ${github.misses.join(", ")}`); + throw new Error(`Timed out waiting for ${what}.\nRuns:\n${await runEvents()}\nGitHub misses: ${github.misses.join(", ")}\nHub:\n${hub.logs.slice(-80).join("\n")}`); } async function rows(query: string): Promise[]> { diff --git a/tests/chain/reread.test.ts b/tests/chain/reread.test.ts index 8cabbef3d..9c1c1e04c 100644 --- a/tests/chain/reread.test.ts +++ b/tests/chain/reread.test.ts @@ -1,12 +1,10 @@ -// A head whose CI finishes with no event, read again by its waiting manager. +// A head whose CI finishes with no event: its manager ended while it waited, and the hub's reconcile pass wakes the next. import { expect, test } from "bun:test"; import { VERDICT_LABELS } from "@corbits/triage-contracts"; -import { chainStack, REREAD, SETTLE_MS, sha, pullOf, once } from "./chain.ts"; +import { RECONCILE_INTERVAL_MS } from "./stack.ts"; +import { chainStack, REREAD, SETTLE_MS, isLiveDeployment, sha, pullOf, once } from "./chain.ts"; -const { github, pushed, until, claimOf, managers, settled, verdictsOf } = chainStack(); - -// The manager reads a waiting head again 5 minutes into its wait; give that pass its polls and a settle on top. -const REREAD_MS = 5 * 60_000 + SETTLE_MS; +const { github, pushed, until, managers, settled, verdictsOf } = chainStack(); const reread = once(async function reread() { const head = sha("reread-1"); @@ -14,21 +12,23 @@ const reread = once(async function reread() { github.setChecks(REREAD, head, [{ name: "build", status: "in_progress", conclusion: null }]); await pushed(REREAD, pullOf(1, head)); await until("the in-progress comment", () => (github.pull(REREAD, 1).comments.length > 0 || undefined)); - const whileWaiting = { comments: github.pull(REREAD, 1).comments.map((comment) => comment.body), claim: (await claimOf(REREAD, 1))! }; + const ended = await settled(REREAD, 1); + const whileWaiting = { comments: github.pull(REREAD, 1).comments.map((comment) => comment.body), claim: ended, managers: (await managers()).slice(before) }; // CI finishes with no check_run, as on a host whose builds send no webhook. github.setChecks(REREAD, head, [{ name: "build", status: "completed", conclusion: "success" }]); - const claim = await settled(REREAD, 1, whileWaiting.claim.history.length + 1, REREAD_MS); + const claim = await settled(REREAD, 1, ended.history.length + 1, 2 * RECONCILE_INTERVAL_MS + SETTLE_MS); return { head, whileWaiting, claim, deployed: (await managers()).slice(before), pull: github.pull(REREAD, 1) }; }); -test("a head whose CI finishes with no event is read again by its waiting manager within five minutes, and the verdict replaces the in-progress comment in place", async () => { +test("a head whose CI finishes with no event, after its manager ended to wait, is woken by the reconcile pass, and the verdict replaces the in-progress comment in place", async () => { const { head, whileWaiting, claim, deployed, pull } = await reread(); expect(whileWaiting.comments).toEqual([expect.stringContaining("**Triage: In progress**\n\nTriage is waiting for CI to finish")]); - expect(whileWaiting.claim.status).toBe("live"); - expect(deployed).toHaveLength(1); - expect(new Set(claim.history.map((pass) => pass.runId))).toEqual(new Set([deployed[0]!.id])); + expect(whileWaiting.claim.status).toBe("ended"); + expect(whileWaiting.managers.map((run) => isLiveDeployment(run.status))).toEqual([false]); + expect(deployed).toHaveLength(2); expect(claim).toMatchObject({ status: "ended" }); + expect(claim.history.at(-1)).toMatchObject({ head, runId: deployed[1]!.id }); expect(pull.labels).toEqual([VERDICT_LABELS.pass]); expect(pull.comments.map((comment) => comment.body)).toEqual([expect.stringContaining("**Triage: Pass**")]); - expect(await verdictsOf(REREAD, 1)).toEqual([expect.objectContaining({ run_id: deployed[0]!.id, head_sha: head, run_status: "completed" })]); + expect(await verdictsOf(REREAD, 1)).toEqual([expect.objectContaining({ run_id: deployed[1]!.id, head_sha: head, run_status: "completed" })]); }); diff --git a/tests/chain/sidecar-cap.test.ts b/tests/chain/sidecar-cap.test.ts index f8f0a1c3d..e2389b373 100644 --- a/tests/chain/sidecar-cap.test.ts +++ b/tests/chain/sidecar-cap.test.ts @@ -80,7 +80,7 @@ async function until(what: string, read: () => Promise, ms = 1 throw new Error(`Timed out waiting for ${what}.\n${hub.logs.slice(-80).join("\n")}`); } -type Claim = { status: "live" | "ended"; runId?: string; history: { head: string; runId: string }[] }; +type Claim = { status: "live" | "ended"; runId?: string; starts: number; deployAskedAt?: string; history: { head: string; runId: string }[] }; async function claimOf(repo: string): Promise { const [row] = await rows(`select state from corbits_triage.manager_state where kind = 'claim' and key = 'mgr:github:${repo}#1'`); @@ -101,7 +101,7 @@ test("with every local slot taken and no sandbox, a manager waits for one to fre }); expect(waiting).toEqual({ failure_code: "capacity_unavailable" }); expect(hub.logs.some((line) => line.includes('"msg":"local_sidecar_refused"') && line.includes("2 local sidecars already run"))).toBe(true); - expect(await claimOf(SECOND)).toBeUndefined(); + expect(await claimOf(SECOND)).toMatchObject({ status: "ended", starts: 0, deployAskedAt: expect.any(String), history: [] }); releaseFirst(); const second = await until("the second manager to post its verdict", async () => { diff --git a/tests/chain/unwoken.test.ts b/tests/chain/unwoken.test.ts index 5b99756a3..0b9ad2073 100644 --- a/tests/chain/unwoken.test.ts +++ b/tests/chain/unwoken.test.ts @@ -4,16 +4,17 @@ import { chainStack, UNWOKEN, isLiveDeployment, sha, pullOf, comments, once } fr const { github, pushed, eachRunEvent, until, rows, managers, settled } = chainStack(); -// The listener waits two minutes for a manager's key before its wake mail fails; stock then stops it within its 30-second cancel grace. -const UNWOKEN_MS = 5 * 60_000; +// The listener waits two minutes for each manager's key before its wake mail fails, for the first and the one it deploys in +// its place; stock then stops each within its 30-second cancel grace. +const UNWOKEN_MS = 8 * 60_000; const INLINE = "inline:"; -/** The managers the listener's unwoken steps cancelled, as their outputs, which the hub reads back inline, name them. */ +/** The managers the listener's unwoken and unwokenAgain steps cancelled, as their outputs, which the hub reads back inline, name them. */ async function unwokenCancelled(): Promise<{ runId: string; reason: string }[]> { const cancelled: { runId: string; reason: string }[] = []; await eachRunEvent(function collect(_runId, { type, body }) { - if (type !== "StepCompleted" || body["stepId"] !== "unwoken") return; + if (type !== "StepCompleted" || (body["stepId"] !== "unwoken" && body["stepId"] !== "unwokenAgain")) return; const { ref } = body["output"] as { ref: string }; if (!ref.startsWith(INLINE)) throw new Error(`unwoken's output is not inline: ${ref}`); cancelled.push(...(JSON.parse(ref.slice(INLINE.length)) as { cancelled: { runId: string; reason: string }[] }).cancelled); @@ -36,12 +37,12 @@ const unwoken = once(async function unwoken() { const [first, second] = [sha("unwoken-1"), sha("unwoken-2")]; const [before, missed] = [(await managers()).length, github.misses.length]; await refuseKeys(); - let cancelled: { id: string; status: string }; + let cancelled: { id: string; status: string }[]; try { await pushed(UNWOKEN, pullOf(1, first)); - cancelled = await until("the unwoken manager to end", async function ended() { - const [deployed] = (await managers()).slice(before); - return deployed && !isLiveDeployment(deployed.status) ? deployed : undefined; + cancelled = await until("the unwoken manager and the one deployed in its place to end", async function ended() { + const deployed = (await managers()).slice(before); + return deployed.length === 2 && deployed.every((run) => !isLiveDeployment(run.status)) ? deployed : undefined; }, UNWOKEN_MS); } finally { await acceptKeys(); @@ -52,13 +53,13 @@ const unwoken = once(async function unwoken() { return { second, cancelled, whileUnwoken, claim, unwoken: await unwokenCancelled(), deployed: (await managers()).slice(before), writes: github.writesTo(UNWOKEN, 1), misses: github.misses.slice(missed) }; }); -test("a manager whose wake mail failed is cancelled, and the next event deploys a fresh one that posts the verdict", async () => { +test("a manager whose wake mail failed is cancelled, as is the one deployed in its place, and the next event deploys a fresh one that posts the verdict", async () => { const { second, cancelled, whileUnwoken, claim, unwoken: reported, deployed, writes, misses } = await unwoken(); expect(misses).toEqual([]); - expect(cancelled.status).toBe("cancelled"); + expect(cancelled.map((run) => run.status)).toEqual(["cancelled", "cancelled"]); expect(whileUnwoken).toEqual([]); - expect(reported).toEqual([{ runId: cancelled.id, reason: expect.stringContaining("public key") }]); - expect(deployed.map((run) => run.id)).toEqual([cancelled.id, claim.history.at(-1)!.runId]); + expect(reported).toEqual(cancelled.map((run) => ({ runId: run.id, reason: expect.stringContaining("public key") }))); + expect(deployed.map((run) => run.id)).toEqual([...cancelled.map((run) => run.id), claim.history.at(-1)!.runId]); expect(claim).toMatchObject({ status: "ended", history: [{ head: second }] }); expect(comments(writes)).toEqual([expect.stringContaining("**Triage: Pass**")]); }); diff --git a/tests/chain/waiting.test.ts b/tests/chain/waiting.test.ts index eae4753f2..8447bda56 100644 --- a/tests/chain/waiting.test.ts +++ b/tests/chain/waiting.test.ts @@ -1,11 +1,11 @@ -// A head whose CI runs after the first post, kept by one manager until its final verdict. +// A head whose CI runs after the first post: its manager ends while it waits, and the check event wakes the next. import { expect, test } from "bun:test"; import { VERDICT_LABELS, WAITING_LABEL } from "@corbits/triage-contracts"; import { deliver } from "./stack.ts"; import { chainStack, WAITING, isLiveDeployment, sha, pullOf, comments, labelled, once } from "./chain.ts"; const chain = chainStack(); -const { github, model, pushed, until, claimOf, managers, settled, verdictsOf } = chain; +const { github, model, pushed, until, managers, settled, verdictsOf } = chain; const waiting = once(async function waiting() { const head = sha("waiting-1"); @@ -13,31 +13,33 @@ const waiting = once(async function waiting() { github.setChecks(WAITING, head, [{ name: "build", status: "in_progress", conclusion: null }]); await pushed(WAITING, pullOf(1, head)); await until("the Waiting label", () => labelled(github.writesTo(WAITING, 1)).includes(WAITING_LABEL) || undefined); - const whileWaiting = { writes: [...github.writesTo(WAITING, 1)], claim: (await claimOf(WAITING, 1))!, managers: (await managers()).slice(before), asked: model.decisions.length - asked }; + // The manager ends once it posted that the head waits; the check event is sent only after. + const ended = await settled(WAITING, 1); + const whileWaiting = { writes: [...github.writesTo(WAITING, 1)], claim: ended, managers: (await managers()).slice(before), asked: model.decisions.length - asked }; github.setChecks(WAITING, head, [{ name: "build", status: "completed", conclusion: "success" }]); await deliver(chain.hub, chain.workspace, "check_run", { action: "completed", installation: { id: 1 }, repository: { full_name: WAITING }, check_run: { head_sha: head, status: "completed", conclusion: "success", pull_requests: [{ number: 1, head: { sha: head } }] }, }); - const claim = await settled(WAITING, 1, whileWaiting.claim.history.length + 2); + const claim = await settled(WAITING, 1, ended.history.length + 1); return { head, whileWaiting, claim, deployed: (await managers()).slice(before), asked: model.decisions.slice(asked).map((row) => row.id), pull: github.pull(WAITING, 1), misses: github.misses.slice(missed) }; }); -test("a head whose CI runs after the first post keeps one manager, which judges it once CI finished, posts the final verdict and only then ends", async () => { +test("a head whose CI runs after the first post ends its manager while it waits, and the check event wakes a fresh one that judges it and posts the final verdict", async () => { const { head, whileWaiting, claim, deployed, asked, pull, misses } = await waiting(); expect(misses).toEqual([]); expect(labelled(whileWaiting.writes)).toEqual([WAITING_LABEL]); expect(comments(whileWaiting.writes)).toEqual([expect.stringContaining("**Triage: In progress**\n\nTriage is waiting for CI to finish")]); expect(whileWaiting.asked).toBe(0); - expect(whileWaiting.claim).toMatchObject({ status: "live", runId: whileWaiting.managers[0]!.id }); - expect(whileWaiting.managers.map((run) => run.status)).toEqual(["running"]); - expect(deployed.map((run) => run.id)).toEqual([whileWaiting.managers[0]!.id]); - expect(isLiveDeployment(deployed[0]!.status)).toBe(false); - expect(new Set(claim.history.map((pass) => pass.runId))).toEqual(new Set([deployed[0]!.id])); + expect(whileWaiting.claim.status).toBe("ended"); + expect(whileWaiting.managers).toHaveLength(1); + expect(isLiveDeployment(whileWaiting.managers[0]!.status)).toBe(false); + expect(deployed).toHaveLength(2); + expect(deployed.every((run) => !isLiveDeployment(run.status))).toBe(true); expect(asked).toContain("focused"); expect(pull.labels).toEqual([VERDICT_LABELS.pass]); expect(pull.comments.map((comment) => comment.body)).toEqual([expect.stringContaining("**Triage: Pass**")]); expect(claim).toMatchObject({ status: "ended" }); - expect(claim.history.at(-1)!.head).toBe(head); - expect(await verdictsOf(WAITING, 1)).toEqual([expect.objectContaining({ run_id: deployed[0]!.id, head_sha: head, run_status: "completed" })]); + expect(claim.history.at(-1)).toMatchObject({ head, runId: deployed[1]!.id }); + expect(await verdictsOf(WAITING, 1)).toEqual([expect.objectContaining({ run_id: deployed[1]!.id, head_sha: head, run_status: "completed" })]); });