feat(manager): end the manager while it waits on CI

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);

apps/hub/src/github/manager-runs.ts

@@ -33,10 +33,12 @@ export function createManagerRuns(db: DB["db"], verdicts: Pick<VerdictRecordStor
       else byHead.set(key, [run]);
     }
 
-    for (const { pr, live, gone } of pulls) {
+    for (const { pr, live, gone, waiting } of pulls) {
       const { source } = pr;
       // The hub checked the run is up, so it is running now and never stuck.
       if (live?.head !== undefined) observed(pr.repo, pr.number, live.head, { runId: live.runId, source, status: "running", startedAt: now().toISOString(), ...(live.pass && { passAt: live.pass.takenAt }), ...(live.awaiting.length > 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 }) });

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<string, RepoPlan>, 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. */

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;

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<PrTriageRow, "number" | "headSha">): string {
   return runKey(row.number, row.headSha);
@@ -145,6 +147,7 @@ function mergeRows(read: ReadonlyMap<string, PrTriageRow>, 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<string, ObservedRun[]> | undefined): Promise<Settled[]> {
+    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<typeof summarizeChecks>;
+        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<LoadedRepo | undefined> {
     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;

apps/hub/src/github/triage-verdicts.postgres.test.ts

@@ -59,15 +59,15 @@ 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 = {
   byRepo: new Map([
     ...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,

apps/hub/src/github/triage-verdicts.ts

@@ -110,11 +110,12 @@ function livePulls(runs: ObservedRuns): Map<string, LivePull> {
   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;