fix(hub): keep a row written after a reconcile pass read the state

apps/hub/src/github/interleave.postgres.test.ts

@@ -11,6 +11,7 @@ import { createBacklogStart } from "./backlog-triage.js";
 import { ListenerUnroutedError } from "./deployment.js";
 import { createGithubPrTriage } from "./pr-triage.js";
 import { DEFAULT_RECONCILE_POLICY } from "./reconcile-plan.js";
+import { failDelivery, queueHead } from "./triage-queue.js";
 import { createTriageReconciler, createUnroutedTries } from "./triage-reconciler.js";
 import { createTriageStateStore, type TriageStateStore } from "./triage-state-store.js";
 import type { ObservedRuns } from "./manager-runs.js";
@@ -62,12 +63,12 @@ function route(store: TriageStateStore, deliver: Deliver) {
   });
 }
 
-function reconciler(store: TriageStateStore, heads: number[], deliver: Deliver, pack: CheckPack = PACK, checks: CheckRun[] = []) {
+function reconciler(store: TriageStateStore, heads: number[], deliver: Deliver, pack: CheckPack = PACK, checks: CheckRun[] = [], updatedAt = LONG_AGO) {
   return createTriageReconciler({
     source: "github",
     tenants: async () => [{ id: TENANT, domain: "tenant.example", config: CONFIG }],
     listener: async () => DEPLOYMENT,
-    openHeadsFor: async () => async () => heads.map((number) => ({ number, headSha: number === 8 ? "abc123" : `sha${number}`, updatedAt: LONG_AGO, draft: false })),
+    openHeadsFor: async () => async () => heads.map((number) => ({ number, headSha: number === 8 ? "abc123" : `sha${number}`, updatedAt, draft: false })),
     activity: { stale: () => true },
     managerRuns: noRuns,
     headChecks: async () => checks,
@@ -193,3 +194,28 @@ test("PostgreSQL: a maintainer's triage request while the listener is starting a
     expect(mailsFor8(mails)).toBe(1);
   });
 }, 120_000);
+
+test("PostgreSQL: a webhook's head queued while a reconcile pass plans over it, whose mail then reaches no listener, goes out on the next pass", async () => {
+  await withStore(async (store) => {
+    const opened = { tenantId: TENANT, repo: REPO, number: 9, headSha: "sha9" };
+    let interleaved = false;
+    const racing: TriageStateStore = {
+      load: (tenantId, repo) => store.load(tenantId, repo),
+      async save(tenantId, repo, rows, expected) {
+        if (!interleaved) {
+          interleaved = true;
+          await queueHead(store, opened, new Date(NOW.getTime() - 5), () => false);
+        }
+        return store.save(tenantId, repo, rows, expected);
+      },
+    };
+    await reconciler(racing, [8, 9], async () => {}, PACK, [], NOW.toISOString())();
+    await failDelivery(store, opened, new ListenerUnroutedError("github", DEPLOYMENT.address), NOW);
+
+    const mails: Array<{ items?: Array<{ prNumber: number }> }> = [];
+    await reconciler(store, [8, 9], async (payload) => {
+      mails.push(payload as never);
+    }, PACK, [], NOW.toISOString())();
+    expect(mails.flatMap((mail) => mail.items?.map((item) => item.prNumber) ?? [])).toContain(9);
+  });
+}, 120_000);

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

@@ -125,13 +125,21 @@ function headKey(row: Pick<PrTriageRow, "number" | "headSha">): string {
   return runKey(row.number, row.headSha);
 }
 
-/** Theirs in their order, each head replaced by ours only when ours is newer; heads only we have follow. */
-function mergeRows(theirs: readonly PrTriageRow[], ours: readonly PrTriageRow[]): PrTriageRow[] {
+/** Whether someone wrote the head's row after the pass read `read`. */
+function writtenSince(read: ReadonlyMap<string, PrTriageRow>, row: PrTriageRow): boolean {
+  return read.get(headKey(row))?.updatedAt !== row.updatedAt;
+}
+
+/**
+ * Theirs in their order, each head replaced by ours unless they wrote it after the pass read it, whatever either clock
+ * says; heads only we have follow.
+ */
+function mergeRows(read: ReadonlyMap<string, PrTriageRow>, theirs: readonly PrTriageRow[], ours: readonly PrTriageRow[]): PrTriageRow[] {
   const byHead = new Map(ours.map((row) => [headKey(row), row]));
-  const merged = theirs.map(function newer(row) {
+  const merged = theirs.map(function unlessWritten(row) {
     const mine = byHead.get(headKey(row));
     byHead.delete(headKey(row));
-    return mine !== undefined && new Date(mine.updatedAt).getTime() > new Date(row.updatedAt).getTime() ? mine : row;
+    return mine === undefined || writtenSince(read, row) ? row : mine;
   });
   return [...merged, ...byHead.values()];
 }
@@ -276,10 +284,15 @@ export function createTriageReconciler(deps: TriageReconcilerDeps) {
       if (!(err instanceof TriageStateConflictError)) throw err;
     }
     const fresh = await deps.store.load(tenantId, repo);
+    const read = new Map(stored.rows.map((row) => [headKey(row), row]));
     const theirs = new Map(fresh.rows.map((row) => [headKey(row), row]));
     const headOf = (queued: QueuedPr) => headKey(plan.rows.find((row) => row.number === queued.number)!);
-    const left = new Set(plan.enqueue.map(headOf).filter((key) => theirs.has(key) && inFlight(theirs.get(key)!)));
-    const rows = mergeRows(fresh.rows, plan.rows.filter((row) => !left.has(headKey(row))));
+    function takenByThem(key: string): boolean {
+      const row = theirs.get(key);
+      return row !== undefined && (inFlight(row) || writtenSince(read, row));
+    }
+    const left = new Set(plan.enqueue.map(headOf).filter(takenByThem));
+    const rows = mergeRows(read, fresh.rows, plan.rows.filter((row) => !left.has(headKey(row))));
     try {
       const version = await deps.store.save(tenantId, repo, rows, fresh.version);
       deps.log({ level: "info", msg: "triage_state_merged", tenantId, repo, left: left.size });

tests/chain/chain.ts

@@ -8,6 +8,7 @@ import { join } from "node:path";
 import { sql } from "drizzle-orm";
 import { createDB } from "@intx/db";
 import { listWorkflowDeployments, listWorkflowRuns, readWorkflowRunEvents } from "@intx/hub-client";
+import { parseTriageState, triageStateName, type PrTriageRow } from "@corbits/triage-contracts";
 import { saveAutomationMode } from "../../apps/web/src/lib/hub-api.ts";
 import { fakeGithub, OWNER, type Pull, type Write } from "./github.ts";
 import { stubModel, type Answer, type Question } from "./model.ts";
@@ -183,16 +184,26 @@ export function chainStack(env: Record<string, string> = {}) {
     return found.map((row) => ({ id: String(row["id"]), createdAt: new Date(String(row["created_at"])), status: String(row["status"]) }));
   }
 
-  /** The claim once its manager's run is gone, whether done ended the claim or the run stopped without ending it. */
+  /** The pull request's rows in the hub's triage state. */
+  async function triageRows(repo: string, number: number): Promise<PrTriageRow[]> {
+    const [stored] = await rows(`select content from artifacts.artifact where tenant_id = '${workspace.tenantId}' and title = '${triageStateName(repo)}' and archived_at is null`);
+    return parseTriageState(stored?.["content"], repo)?.prs.filter((row) => row.number === number) ?? [];
+  }
+
+  /** The claim once its manager's run is gone, whether done ended the claim or the run stopped without ending it; on a timeout, the error names the head's triage rows. */
   async function settled(repo: string, number: number, passes = 1, ms = SETTLE_MS): Promise<Claim> {
-    return until(`${repo}#${number}'s manager to stop`, async function stopped() {
-      const claim = await claimOf(repo, number);
-      if (!claim || claim.history.length < passes) return undefined;
-      const runs = await managers();
-      const runId = claim.runId ?? claim.history.at(-1)!.runId;
-      const live = runs.some((run) => run.id === runId && isLiveDeployment(run.status));
-      return live ? undefined : claim;
-    }, ms);
+    try {
+      return await until(`${repo}#${number}'s manager to stop`, async function stopped() {
+        const claim = await claimOf(repo, number);
+        if (!claim || claim.history.length < passes) return undefined;
+        const runs = await managers();
+        const runId = claim.runId ?? claim.history.at(-1)!.runId;
+        const live = runs.some((run) => run.id === runId && isLiveDeployment(run.status));
+        return live ? undefined : claim;
+      }, ms);
+    } catch (err) {
+      throw new Error(`${String(err)}\nTriage state: ${JSON.stringify(await triageRows(repo, number))}`, { cause: err });
+    }
   }
 
   async function verdictsOf(repo: string, number: number): Promise<Record<string, unknown>[]> {