feat(manager): end the manager after it posts and wake a fresh one by mail

apps/hub/src/code-review-runs.postgres.test.ts

@@ -59,9 +59,9 @@ test("PostgreSQL: a pull request's latest code review is what its manager record
       await db.insert(schema.workflowRun).values({ id: runId, definitionId: "wfd_review", anchorRunId: runId, tenantId: TENANT, status: "running", createdAt });
       return createdAt;
     }
-    async function claim(reviewers: unknown[], taken: string[]) {
+    async function claim(reviewers: unknown[], taken: string[], status: "live" | "ended" = "live") {
       const since = new Date(0).toISOString();
-      const state = { status: "live", claimId: "claim-1", claimedAt: since, runId: "run_manager", reviewers, history: [{ id: "pass-1", head: "sha1", events: taken, takenAt: since, takes: 1, runId: "run_manager", doneAt: since }], starts: 1 };
+      const state = { status, claimId: "claim-1", claimedAt: since, ...(status === "live" && { runId: "run_manager" }), reviewers, history: [{ id: "pass-1", head: "sha1", events: taken, takenAt: since, takes: 1, runId: "run_manager", doneAt: since }], starts: 1 };
       await db.execute(sql`INSERT INTO corbits_triage.manager_state (tenant_id, key, kind, state) VALUES (${TENANT}, 'mgr:github:acme/widgets#7', 'claim', ${JSON.stringify(state)}::jsonb) ON CONFLICT (tenant_id, key) DO UPDATE SET state = excluded.state`);
     }
     let head: PullRequestToReview = { title: "Fix", description: "", headSha: "sha5", draft: false, open: true, diff: DIFF };
@@ -97,6 +97,10 @@ test("PostgreSQL: a pull request's latest code review is what its manager record
     const live = await deployed("run_live");
     await claim([reviewer("run_live", "sha4")], [eventKey("run_old"), eventKey("run_new")]);
     expect(await reviewsOf(GITHUB)).toEqual({ review: { status: "running", headSha: "sha4", startedAt: live.toISOString() }, completed: finished, carriedTo: null });
+    // A manager ends once it mailed its review, which still runs until it reports or its deadline passes, and no head takes over the finished one meanwhile.
+    await claim([reviewer("run_live", "sha4")], [eventKey("run_old"), eventKey("run_new")], "ended");
+    expect(await reviewsOf(GITHUB, "sha5")).toEqual({ review: { status: "running", headSha: "sha4", startedAt: live.toISOString() }, completed: finished, carriedTo: null });
+    expect(hostReads).toBe(0);
 
     // A finished review names its diff's hash, so a later head with the same diff takes it over.
     await recordEvent(db, TENANT, GITHUB, "run_live", { type: "reviewer-result", head: "sha4", payload: { kind: "code-review", result: OUTPUT, diffHash: diffHash(DIFF) } });

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

@@ -50,13 +50,14 @@ const pr8: PullRequestRef = { source: "github", repo: REPO, number: 8 };
 
 /** The manager posts a settled verdict on h1 and reports done, recording it as the done route does, in done's transaction. */
 async function reportedDone(db: Db, pr: PullRequestRef, runId: string, now: Date): Promise<void> {
-  const result = await done(db, TENANT, pr, runId, [], { head: "h1", request: { repo: REPO, number: pr.number, labels: [], owned: [], comment: "" } }, async function record(tx, closing) {
+  const report = { head: "h1", request: { repo: REPO, number: pr.number, labels: [], owned: [], comment: "" } };
+  const result = await done(db, TENANT, pr, runId, [], { report, record: async function record(tx, closing) {
     if (!closing) throw new Error("a reporting pass has a closing");
     await replaceVerdict(tx, TENANT, {
       runId, repo: REPO, number: pr.number, headSha: closing.head, startedAt: new Date(closing.startedAt), verdictAt: now, settled: true, state: "ready-monitoring",
       actor: "author", humanGated: false, degraded: null, checks: [], dos: [], commented: true, detail: { headSha: closing.head, checks: [] }, runStatus: "completed",
     });
-  });
+  } });
   if (result.outcome === "refused") throw new Error("done refused the manager the claim names");
 }
 
@@ -90,7 +91,7 @@ test("PostgreSQL: the reconciler does not mail a head again while a live manager
     await reportedDone(db, triaged, "run_triaged", now);
     await mailed(db, held, "h1");
     await take(db, TENANT, held, "run_held");
-    await done(db, TENANT, held, "run_held", [], { head: "h1", waiting: true });
+    await done(db, TENANT, held, "run_held", [], { report: { head: "h1", waiting: true } });
     const firstSeenAt = new Date(now.getTime() - MINUTE).toISOString();
     const fresh: PrTriageRow = { number: 8, headSha: "h1", status: "new", attempts: 0, firstSeenAt, updatedAt: firstSeenAt };
 

apps/hub/src/manager-intake.postgres.test.ts

@@ -7,7 +7,8 @@ import { APPROVE_DO_KIND, emptyPack, repoPolicy } from "@corbits/triage-contract
 import type { PullRequestToReview } from "./pull-request-reader.js";
 import { mailPayload, waitingPayload } from "./github/bridge.js";
 import { createPauseGate } from "./github/pr-workflow.js";
-import { createManagerIntake } from "./manager-intake.js";
+import { createManagerIntake, createManagerWake, createOverdueWake, createSupersede } from "./manager-intake.js";
+import { recordEvent } from "./manager-state.js";
 import { migrateTriageTables } from "./triage-db.js";
 
 const TEST_DB = { host: "localhost", port: 5432, user: "postgres", password: "postgres", database: "interchange" };
@@ -50,6 +51,16 @@ const OPEN = [{ number: 7, headSha: "h1", draft: false }, { number: 9, headSha:
 function stack(db: Db, listener: string | undefined, pull = PULL, gate: Gate = passed, open = OPEN) {
   const mailed: { address: string; payload: unknown }[] = [];
   const logged: Record<string, unknown>[] = [];
+  const cancelled: { runId: string; reason: string }[] = [];
+  const refusing = new Set<string>();
+  const { supersede, supersedeAll } = createSupersede({
+    db,
+    async cancel(_tenantId, runId, reason) {
+      if (refusing.has(runId)) throw new Error("cancel failed");
+      cancelled.push({ runId, reason });
+    },
+    log: (entry) => logged.push(entry),
+  });
   const toListener = createManagerIntake({
     db,
     listener: async () => listener,
@@ -59,10 +70,11 @@ function stack(db: Db, listener: string | undefined, pull = PULL, gate: Gate = p
     async deliver(_tenantId, address, payload) {
       mailed.push({ address, payload });
     },
+    supersede,
     gate,
     log: (entry) => logged.push(entry),
   });
-  return { toListener, mailed, logged };
+  return { toListener, mailed, logged, cancelled, refusing, supersedeAll };
 }
 
 async function events(db: Db): Promise<{ key: string; state: { head: string; payload: Record<string, unknown> } }[]> {
@@ -114,10 +126,10 @@ test("PostgreSQL: each pull request a mail names is recorded as its own event, a
   });
 }, 60_000);
 
-async function claimManager(db: Db, runId: string, number: number, status: "running" | "deployed", createdAt?: Date): Promise<void> {
+async function claimManager(db: Db, runId: string, number: number, status: "running" | "deployed" | "completed", createdAt?: Date, reviewers: object[] = []): Promise<void> {
   await db.insert(schema.workflowDefinition).values({ id: "wfd_intake", tenantId: TENANT, name: "triage-manager-github" }).onConflictDoNothing();
   await db.insert(schema.workflowRun).values({ id: runId, anchorRunId: runId, definitionId: "wfd_intake", tenantId: TENANT, status, ...(createdAt && { createdAt }) });
-  const claim = { status: "live", claimId: `claim-${runId}`, claimedAt: new Date().toISOString(), runId, reviewers: [], history: [], starts: 1 };
+  const claim = { status: "live", claimId: `claim-${runId}`, claimedAt: new Date().toISOString(), runId, reviewers, history: [], starts: 1 };
   await db.execute(sql`INSERT INTO corbits_triage.manager_state (tenant_id, key, kind, state) VALUES (${TENANT}, ${`mgr:github:${RECORD.name}#${number}`}, 'claim', ${JSON.stringify(claim)}::jsonb)`);
 }
 
@@ -135,6 +147,137 @@ test("PostgreSQL: a pull request whose manager is up is mailed that manager, and
   });
 }, 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");
+    const mailed: { address: string; payload: unknown }[] = [];
+    function waker(listener: string | undefined) {
+      return createManagerWake({
+        db,
+        listener: async () => listener,
+        async deliver(_tenantId, address, payload) {
+          mailed.push({ address, payload });
+        },
+      });
+    }
+    const pr = { source: "github" as const, repo: RECORD.name, number: 7 };
+    expect(await waker(LISTENER)(TENANT, pr, "h1")).toBe(true);
+    await db.execute(sql`UPDATE corbits_triage.manager_state SET state = (state - 'runId') || '{"status":"ended"}' WHERE kind = 'claim'`);
+    expect(await waker(LISTENER)(TENANT, pr, "h1")).toBe(true);
+    expect(await waker(undefined)(TENANT, pr, "h1")).toBe(false);
+    expect(mailed).toEqual([
+      { address: "run_mgr@intake.test", payload: { kind: "manage", tenantId: TENANT, repo: RECORD.name, repoSource: "github", prNumber: 7 } },
+      { address: LISTENER, payload: { kind: "pr", repo: RECORD.name, tenantId: TENANT, items: [{ prNumber: 7, headSha: "h1" }] } },
+    ]);
+  });
+}, 60_000);
+
+test("PostgreSQL: a push stops the judge and the code review still out on the older head, once each, and one whose stop failed stays on the claim until a reconcile pass stops it", async () => {
+  await withDatabase(async function superseded(db) {
+    const deadline = new Date(Date.now() + 60 * 60_000).toISOString();
+    const judge = { kind: "judge", head: "h1", runId: "run_judge", deployedBy: "run_old", deadline };
+    const review = { ...judge, kind: "code-review", runId: "run_review" };
+    const current = { ...judge, head: "h2", runId: "run_current" };
+    await claimManager(db, "run_old", 7, "completed", undefined, [judge, review, current]);
+    const { toListener, cancelled, logged, refusing, supersedeAll } = stack(db, LISTENER);
+    async function held(): Promise<unknown[]> {
+      const [claim] = await db.execute<{ state: { reviewers: unknown[] } }>(sql`SELECT state FROM corbits_triage.manager_state WHERE kind = 'claim'`);
+      return claim!.state.reviewers;
+    }
+    const where = { tenantId: TENANT, source: "github", repo: RECORD.name, number: 7 };
+    refusing.add("run_review");
+    expect(await toListener(TENANT, mailPayload(RECORD, POLICY, STORED, { prNumber: 7, headSha: "h2", deliveryId: "push-1" }))).toBe(true);
+    expect(cancelled).toEqual([{ runId: "run_judge", reason: "superseded by a newer head" }]);
+    expect(await held()).toEqual([review, current]);
+    expect(logged.filter((entry) => entry["msg"] === "reviewer_supersede_failed")).toEqual([{ level: "error", msg: "reviewer_supersede_failed", ...where, kind: "code-review", head: "h1", runId: "run_review", error: "Error: cancel failed" }]);
+
+    refusing.clear();
+    await supersedeAll(TENANT);
+    expect(await toListener(TENANT, mailPayload(RECORD, POLICY, STORED, { prNumber: 7, headSha: "h2", deliveryId: "push-2" }))).toBe(true);
+    expect(cancelled).toEqual([{ runId: "run_judge", reason: "superseded by a newer head" }, { runId: "run_review", reason: "superseded by a newer head" }]);
+    expect(logged.filter((entry) => entry["msg"] === "reviewer_superseded")).toEqual([
+      { level: "info", msg: "reviewer_superseded", ...where, kind: "judge", head: "h1", runId: "run_judge" },
+      { level: "info", msg: "reviewer_superseded", ...where, kind: "code-review", head: "h1", runId: "run_review" },
+    ]);
+    expect(await held()).toEqual([current]);
+  });
+}, 60_000);
+
+test("PostgreSQL: the reconcile pass wakes a manager for a pull request whose reviewer of its newest head is past its deadline with no result and no manager up", async () => {
+  await withDatabase(async function overdue(db) {
+    const past = new Date(Date.now() - 60_000).toISOString();
+    const late = { kind: "judge", head: "h1", runId: "run_judge", deployedBy: "run_gone", deadline: past };
+    await claimManager(db, "run_gone", 7, "completed", undefined, [late]);
+    await claimManager(db, "run_up", 8, "running", undefined, [{ ...late, runId: "run_judge_8" }]);
+    await claimManager(db, "run_reported", 9, "completed", undefined, [{ ...late, runId: "run_judge_9" }]);
+    for (const number of [7, 8, 9]) await recordEvent(db, TENANT, { source: "github", repo: RECORD.name, number }, `d${number}`, { type: "webhook", head: "h1", payload: { kind: "pr", headSha: "h1" } });
+    await recordEvent(db, TENANT, { source: "github", repo: RECORD.name, number: 9 }, "run_judge_9", { type: "reviewer-result", head: "h1", payload: { kind: "judge", failed: "the judge did not finish" } });
+    await db.execute(sql`UPDATE corbits_triage.manager_state SET state = state || '{"consumedBy":"pass_9"}' WHERE kind = 'event' AND key LIKE '%:run_judge_9'`);
+    const mailed: { address: string; payload: unknown }[] = [];
+    const logged: Record<string, unknown>[] = [];
+    const wake = createManagerWake({
+      db,
+      listener: async () => LISTENER,
+      async deliver(_tenantId, address, payload) {
+        mailed.push({ address, payload });
+      },
+    });
+    const wakeOverdue = createOverdueWake({ db, wake, log: (entry) => logged.push(entry) });
+    await wakeOverdue(TENANT);
+    await wakeOverdue(TENANT);
+    const woken = { address: LISTENER, payload: { kind: "pr", repo: RECORD.name, tenantId: TENANT, items: [{ prNumber: 7, headSha: "h1" }] } };
+    expect(mailed).toEqual([woken]);
+    expect(logged).toEqual([{ level: "info", msg: "manager_woken_overdue", tenantId: TENANT, source: "github", repo: RECORD.name, number: 7, head: "h1" }]);
+
+    // 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 wakeOverdue(TENANT);
+    expect(mailed).toEqual([woken, woken]);
+  });
+}, 60_000);
+
+async function overdueWokenAt(db: Db, number: number): Promise<unknown> {
+  const [claim] = await db.execute<{ state: Record<string, unknown> }>(sql`SELECT state FROM corbits_triage.manager_state WHERE key = ${`mgr:github:${RECORD.name}#${number}`}`);
+  return claim!.state["overdueWokenAt"];
+}
+
+test("PostgreSQL: an overdue wake is recorded only once it succeeded, and one pull request whose wake fails is logged while the next is still woken", async () => {
+  await withDatabase(async function partly(db) {
+    const late = { kind: "judge", head: "h1", runId: "run_judge_7", deployedBy: "run_gone_7", deadline: new Date(Date.now() - 60_000).toISOString() };
+    await claimManager(db, "run_gone_7", 7, "completed", undefined, [late]);
+    await claimManager(db, "run_gone_8", 8, "completed", undefined, [{ ...late, runId: "run_judge_8" }]);
+    for (const number of [7, 8]) await recordEvent(db, TENANT, { source: "github", repo: RECORD.name, number }, `d${number}`, { type: "webhook", head: "h1", payload: { kind: "pr", headSha: "h1" } });
+    const woken: number[] = [];
+    const logged: Record<string, unknown>[] = [];
+    let failing = true;
+    async function wake(_tenantId: string, pr: { number: number }): Promise<boolean> {
+      if (pr.number === 7 && failing) throw new Error("delivery failed");
+      woken.push(pr.number);
+      return true;
+    }
+    const wakeOverdue = createOverdueWake({ db, wake, log: (entry) => logged.push(entry) });
+    await wakeOverdue(TENANT);
+    expect(woken).toEqual([8]);
+    expect(logged).toEqual([
+      { level: "error", msg: "manager_overdue_wake_failed", tenantId: TENANT, source: "github", repo: RECORD.name, number: 7, head: "h1", error: "Error: delivery failed" },
+      { level: "info", msg: "manager_woken_overdue", tenantId: TENANT, source: "github", repo: RECORD.name, number: 8, head: "h1" },
+    ]);
+    expect(await overdueWokenAt(db, 7)).toBeUndefined();
+    expect(await overdueWokenAt(db, 8)).toBeString();
+
+    failing = false;
+    await wakeOverdue(TENANT);
+    expect(woken).toEqual([8, 7]);
+
+    const unrouted = createOverdueWake({ db, wake: async () => false, log: (entry) => logged.push(entry) });
+    await claimManager(db, "run_gone_9", 9, "completed", undefined, [{ ...late, runId: "run_judge_9" }]);
+    await recordEvent(db, TENANT, { source: "github", repo: RECORD.name, number: 9 }, "d9", { type: "webhook", head: "h1", payload: { kind: "pr", headSha: "h1" } });
+    await unrouted(TENANT);
+    expect(await overdueWokenAt(db, 9)).toBeUndefined();
+  });
+}, 60_000);
+
 test("PostgreSQL: a manager deployed without a sidecar past the startup window is not mailed, and the listener is", async () => {
   await withDatabase(async function stuck(db) {
     await claimManager(db, "run_stuck", 7, "deployed", new Date(Date.now() - 3 * 60_000));