fix(hub): keep a head refused for capacity queued until a slot frees

Author: TheGreatAxios

apps/hub/src/github/coalescer.test.ts

@@ -333,7 +333,7 @@ test("a paused-label head whose pull request cannot be read is left failed, and
   expect(mails).toEqual([]);
   expect(state.rows()).toEqual([expect.objectContaining({ status: "failed", error: "delivery failed: Error: 502 Bad Gateway" })]);
   const pr = { number: 8, headSha: "sha1", updatedAt: new Date(START).toISOString(), draft: false };
-  const plan = planTenant({ repos: [{ name: REPO, prs: [pr], rows: state.rows() }], runs: new Map(), now: new Date(START + DEFAULT_RECONCILE_POLICY.backoffMaxMs), policy: DEFAULT_RECONCILE_POLICY });
+  const plan = planTenant({ repos: [{ name: REPO, prs: [pr], rows: state.rows() }], runs: new Map(), roomless: new Set(), now: new Date(START + DEFAULT_RECONCILE_POLICY.backoffMaxMs), policy: DEFAULT_RECONCILE_POLICY });
   expect(plan.get(REPO)?.enqueue.map((entry) => entry.number)).toEqual([8]);
 });
 

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

@@ -41,7 +41,7 @@ function stubDb() {
 }
 
 async function noRuns(): Promise<ObservedRuns> {
-  return { byRepo: new Map() };
+  return { byRepo: new Map(), roomless: new Set() };
 }
 
 function route(store: TriageStateStore, deliver: Deliver) {

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

@@ -80,7 +80,7 @@ test("PostgreSQL: the reconciler does not mail a head again while a live manager
     const queuedAt = new Date(now.getTime() - 20 * MINUTE).toISOString();
     const queued: PrTriageRow = { number: 7, headSha: "h1", status: "queued", attempts: 1, firstSeenAt: queuedAt, queuedAt, updatedAt: queuedAt };
     function plan(rows: PrTriageRow[], runs: Awaited<ReturnType<typeof managerRuns>>) {
-      return planTenant({ repos: [{ name: REPO, prs: [open(7, now), open(8, now)], rows }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!;
+      return planTenant({ repos: [{ name: REPO, prs: [open(7, now), open(8, now)], rows }], runs: runs.byRepo, roomless: runs.roomless, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!;
     }
 
     const running = plan([queued], await managerRuns(TENANT, [pr7, pr8]));
@@ -114,7 +114,7 @@ test("PostgreSQL: on the listener path a head is mailed once while its manager r
     async function reconcile(afterMs: number): Promise<PrTriageRow> {
       now = new Date(start.getTime() + afterMs);
       const runs = await managerRuns(TENANT, [pr]);
-      const plan = planTenant({ repos: [{ name: REPO, prs: [open(7, start)], rows }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!;
+      const plan = planTenant({ repos: [{ name: REPO, prs: [open(7, start)], rows }], runs: runs.byRepo, roomless: runs.roomless, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!;
       rows = plan.rows;
       mails += plan.enqueue.length;
       return rows[0]!;
@@ -149,7 +149,7 @@ test("PostgreSQL: a manager that went away mid-pass fails its pass's head, so th
     expect(runs.byRepo.get(REPO)?.get(runKey(7, "h1"))).toEqual([expect.objectContaining({ runId: "run_gone", status: "failed" })]);
     const queuedAt = new Date(now.getTime() - MINUTE).toISOString();
     const queued: PrTriageRow = { number: 7, headSha: "h1", status: "queued", attempts: 1, firstSeenAt: queuedAt, queuedAt, updatedAt: queuedAt };
-    const plan = planTenant({ repos: [{ name: REPO, prs: [open(7, now)], rows: [queued] }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!;
+    const plan = planTenant({ repos: [{ name: REPO, prs: [open(7, now)], rows: [queued] }], runs: runs.byRepo, roomless: runs.roomless, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!;
     expect(plan.rows).toEqual([expect.objectContaining({ status: "failed", runId: "run_gone", error: "run failed" })]);
     expect(plan.enqueue).toEqual([]);
   });
@@ -194,7 +194,7 @@ test("PostgreSQL: a head whose manager ended while it waited for CI shows as wai
     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)!;
+    const plan = planTenant({ repos: [{ name: REPO, prs: [open(8, now)], rows: [row] }], runs: observed.byRepo, roomless: observed.roomless, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!;
     expect(plan.rows).toEqual([expect.objectContaining({ number: 8, status: "triaged", runId: "run_waited" })]);
     expect(plan.enqueue).toEqual([]);
   });

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

@@ -4,11 +4,11 @@
 import type { DB } from "@intx/db";
 import { unsettledBy } from "@corbits/triage-workflows/logic/settled";
 import { managedPulls, type PullRequestRef } from "../manager-state.js";
-import { runKey, type ObservedRun } from "./reconcile-plan.js";
+import { pullKey, runKey, type ObservedRun } from "./reconcile-plan.js";
 import type { VerdictRecord, VerdictRecordStore } from "./verdict-record-store.js";
 
-/** Runs by repository, then by `${number}@${headSha}`. */
-export type ObservedRuns = { byRepo: Map<string, Map<string, ObservedRun[]>> };
+/** Runs by repository, then by `${number}@${headSha}`; `roomless`, by `pullKey`, the pull requests whose last manager was refused for capacity. */
+export type ObservedRuns = { byRepo: Map<string, Map<string, ObservedRun[]>>; roomless: Set<string> };
 
 /** The runs of `pulls`' managers. */
 export type ManagerRuns = (tenantId: string, pulls: readonly PullRequestRef[]) => Promise<ObservedRuns>;
@@ -23,6 +23,7 @@ export function createManagerRuns(db: DB["db"], verdicts: Pick<VerdictRecordStor
     const pulls = await managedPulls(db, tenantId, asked);
     const recorded = await verdicts.ofRuns(tenantId, [...new Set(pulls.flatMap((pull) => pull.history.map((pass) => pass.runId)))], asked);
     const byRepo: ObservedRuns["byRepo"] = new Map();
+    const roomless = new Set<string>();
 
     function observed(repo: string, number: number, head: string, run: ObservedRun): void {
       const byHead = byRepo.get(repo) ?? new Map<string, ObservedRun[]>();
@@ -33,8 +34,9 @@ export function createManagerRuns(db: DB["db"], verdicts: Pick<VerdictRecordStor
       else byHead.set(key, [run]);
     }
 
-    for (const { pr, live, gone, waiting } of pulls) {
+    for (const { pr, live, gone, waiting, roomless: refused } of pulls) {
       const { source } = pr;
+      if (refused) roomless.add(pullKey(pr.repo, pr.number));
       // 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.
@@ -44,6 +46,6 @@ export function createManagerRuns(db: DB["db"], verdicts: Pick<VerdictRecordStor
         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 }) });
       }
     }
-    return { byRepo };
+    return { byRepo, roomless };
   };
 }

apps/hub/src/github/pr-triage.test.ts

@@ -31,7 +31,7 @@ function stubDb(repo: Record<string, unknown>) {
 
 function observed(runs: ObservedRun[]): GithubPrTriageDeps["managerRuns"] {
   return async function managerRuns() {
-    return { byRepo: new Map([[REPO, new Map([["8@abc123", runs]])]]), runCount: runs.length, verdicts: [], approved: [] };
+    return { byRepo: new Map([[REPO, new Map([["8@abc123", runs]])]]), roomless: new Set<string>(), runCount: runs.length, verdicts: [], approved: [] };
   };
 }
 
@@ -263,6 +263,7 @@ describe("createGithubPrTriage", () => {
     rows = planTenant({
       repos: [{ name: REPO, prs: [{ number: 8, headSha: "abc123", updatedAt: at(3_600_000), draft: false }], rows }],
       runs: new Map([[REPO, new Map([["8@abc123", [oldVerdict]]])]]),
+      roomless: new Set(),
       now: NOW, policy: DEFAULT_RECONCILE_POLICY,
     }).get(REPO)!.rows;
     const second = await handler(ENABLED, s, deps)(request(), MEMBER);

apps/hub/src/github/reconcile-plan.test.ts

@@ -1,6 +1,6 @@
 import { describe, expect, test } from "bun:test";
-import type { PrTriageRow } from "@corbits/triage-contracts";
-import { DEFAULT_RECONCILE_POLICY, planTenant, requeue, runKey, type ObservedRun, type OpenPr, type RepoPlan } from "./reconcile-plan.js";
+import { WAITING_FOR_CAPACITY, type PrTriageRow } from "@corbits/triage-contracts";
+import { DEFAULT_RECONCILE_POLICY, planTenant, pullKey, requeue, runKey, type ObservedRun, type OpenPr, type RepoPlan } from "./reconcile-plan.js";
 
 const MINUTE = 60_000;
 const T0 = new Date("2026-10-07T00:00:00.000Z");
@@ -20,9 +20,9 @@ function run(runId: string, status: ObservedRun["status"], startedMinute: number
 
 const REPO = "acme/widgets";
 
-function plan(prs: OpenPr[], rows: readonly PrTriageRow[], now: Date, runs: Array<[OpenPr, ObservedRun[]]> = [], maxInFlight = prs.length): RepoPlan {
+function plan(prs: OpenPr[], rows: readonly PrTriageRow[], now: Date, runs: Array<[OpenPr, ObservedRun[]]> = [], maxInFlight = prs.length, roomless: readonly number[] = []): RepoPlan {
   const byHead = new Map(runs.map(([open, observed]) => [runKey(open.number, open.headSha), observed]));
-  const input = { repos: [{ name: REPO, prs, rows }], runs: new Map([[REPO, byHead]]), now, policy: { ...policy, maxInFlight } };
+  const input = { repos: [{ name: REPO, prs, rows }], runs: new Map([[REPO, byHead]]), roomless: new Set(roomless.map((number) => pullKey(REPO, number))), now, policy: { ...policy, maxInFlight } };
   return planTenant(input).get(REPO)!;
 }
 
@@ -217,7 +217,7 @@ describe("reconcile plan", () => {
     ]);
     const rows = [row(1, { headSha: "sha1-old", status: "running", runId: "r_live", queuedAt: at(-6).toISOString() })];
     const prs = [pr(1, "sha1-new"), pr(2), pr(3), pr(4), pr(5), pr(6)];
-    const plans = planTenant({ repos: [{ name: REPO, prs, rows }], runs, now: at(0), policy });
+    const plans = planTenant({ repos: [{ name: REPO, prs, rows }], runs, roomless: new Set(), now: at(0), policy });
     expect(enqueued(plans.get(REPO)!)).toEqual([1, 2]);
   });
 
@@ -272,6 +272,34 @@ describe("reconcile plan", () => {
   });
 });
 
+describe("a head whose manager was refused for capacity", () => {
+  test("stays queued past the never-started window and asks again every pass without spending an attempt, while another head's failures back off", () => {
+    const heads = [pr(1), pr(2)];
+    let rows = plan(heads, [], at(0)).rows;
+    const runs: Array<[OpenPr, ObservedRun[]]> = [[pr(2), [run("run_a", "failed", 1)]]];
+    const passes: Array<[number, number[], number]> = [[5, [1, 2], 2], [10, [1], 2], [15, [1, 2], 3], [20, [1], 3], [30, [1], 3]];
+    for (const [minute, asked, failedAttempts] of passes) {
+      if (minute === 10) runs[0]![1].push(run("run_b", "failed", 6));
+      const pass = plan(heads, rows, at(minute), runs, 2, [1]);
+      expect(enqueued(pass)).toEqual(asked);
+      expect(pass.enqueue.find((entry) => entry.number === 1)?.reason).toBe(WAITING_FOR_CAPACITY);
+      expect(status(pass, 1)).toMatchObject({ status: "queued", attempts: 1, error: WAITING_FOR_CAPACITY });
+      expect(status(pass, 2)).toMatchObject({ attempts: failedAttempts });
+      rows = pass.rows;
+    }
+  });
+
+  test("ages from the next pass as any queued head once a manager deployed after the refusal is not refused", () => {
+    const queued = plan([pr(1)], [], at(0)).rows;
+    const waiting = plan([pr(1)], queued, at(30), [], 1, [1]).rows;
+    const placed = plan([pr(1)], waiting, at(31));
+    expect(enqueued(placed)).toEqual([]);
+    expect(status(placed, 1)).toEqual(expect.not.objectContaining({ error: expect.anything() }));
+    expect(status(plan([pr(1)], placed.rows, at(41)), 1)).toMatchObject({ status: "queued", attempts: 1 });
+    expect(status(plan([pr(1)], placed.rows, at(42)), 1)).toMatchObject({ status: "queued", attempts: 2 });
+  });
+});
+
 describe("triage again on a head that already has a verdict", () => {
   // The route's write on top of a triaged row: queued, queuedAt now, runId kept.
   const requeued: PrTriageRow = { number: 8, headSha: "sha8", status: "queued", attempts: 0, runId: "run-old", firstSeenAt: at(-120).toISOString(), queuedAt: at(-1).toISOString(), updatedAt: at(-1).toISOString() };

apps/hub/src/github/reconcile-plan.ts

@@ -1,6 +1,6 @@
 // Decides, from what the host and the managers' runs say now, which open pull
 // requests to queue again. Pure, so every recovery path is decided in one place.
-import { PR_TRIAGE_STUCK_RUN_MS, type AwaitedStage, type PrTriageRow, type RepoSource } from "@corbits/triage-contracts";
+import { PR_TRIAGE_STUCK_RUN_MS, WAITING_FOR_CAPACITY, type AwaitedStage, type PrTriageRow, type RepoSource } from "@corbits/triage-contracts";
 import { MODEL_NOT_ASKED } from "@corbits/triage-workflows/logic/settled";
 
 /** `labels` only where the host lists them. */
@@ -81,6 +81,8 @@ export type TenantPlanInput = {
   repos: readonly RepoPlanInput[];
   /** Every observed run, by repository, then by `${number}@${headSha}`, whether or not its repository or head is still here. */
   runs: ReadonlyMap<string, ReadonlyMap<string, readonly ObservedRun[]>>;
+  /** By `pullKey`, the pull requests whose last manager every provisioner refused a placement for capacity. */
+  roomless: ReadonlySet<string>;
   now: Date;
   policy: ReconcilePolicy;
 };
@@ -95,6 +97,10 @@ export function runKey(number: number, headSha: string): string {
   return `${number}@${headSha}`;
 }
 
+export function pullKey(repo: string, number: number): string {
+  return `${repo}#${number}`;
+}
+
 export function runKeyNumber(key: string): number {
   return Number(key.slice(0, key.indexOf("@")));
 }
@@ -162,9 +168,11 @@ function sinceStale(row: PrTriageRow, runs: readonly ObservedRun[], policy: Reco
  * only sees runs since it was queued, so a head queued again on top of a
  * verdict stays queued until its new run starts. Runs from before the base
  * branch last moved settle nothing; one of them in flight holds the head
- * until it ends, and the head is stale then.
+ * until it ends, and the head is stale then. A queued head whose last manager
+ * was refused for capacity stays queued, never aging, until a manager deployed
+ * since is not refused.
  */
-function observe(row: PrTriageRow, observed: readonly ObservedRun[], now: Date, policy: ReconcilePolicy): PrTriageRow {
+function observe(row: PrTriageRow, observed: readonly ObservedRun[], roomless: boolean, now: Date, policy: ReconcilePolicy): PrTriageRow {
   const at = now.toISOString();
   const waiting = row.status === "queued" || row.status === "running";
   const current = sinceStale(row, observed, policy);
@@ -185,6 +193,12 @@ function observe(row: PrTriageRow, observed: readonly ObservedRun[], now: Date,
     const failed = withStatus(row, { status: "failed", run: ended, error: failure(ended, isStuck(ended, now, policy)) }, at);
     return row.status === "triaged" && ended.unsettled === MODEL_NOT_ASKED ? { ...failed, attempts: 0 } : failed;
   }
+  if (row.status === "queued" && roomless) return withStatus(row, { status: "queued", error: WAITING_FOR_CAPACITY }, at);
+  if (row.status === "queued" && row.error === WAITING_FOR_CAPACITY) {
+    // A manager deployed since the refusal was not refused, so the head waits for it from now as any queued head does.
+    const { error: _error, ...rest } = row;
+    return { ...rest, queuedAt: at, updatedAt: at };
+  }
   if (row.status === "queued" && row.queuedAt !== undefined && now.getTime() - ms(row.queuedAt) > policy.unstartedAfterMs) {
     return withStatus(row, { status: "failed", error: NEVER_STARTED }, at);
   }
@@ -208,6 +222,8 @@ function isDue(row: PrTriageRow, pr: OpenPr, runs: readonly ObservedRun[], now:
   if (row.status === "new") return now.getTime() - ms(pr.updatedAt) >= policy.webhookGraceMs;
   // Mail that reached no listener says nothing about the head, so it goes out again on the first pass a listener is up.
   if (row.status === "failed" && row.unrouted) return true;
+  // A head waiting for capacity asks for a manager again on every pass; the hub's hold on a recent ask paces the deploys.
+  if (row.status === "queued" && row.error === WAITING_FOR_CAPACITY) return true;
   if (row.status !== "failed" || row.attempts >= policy.maxAttempts) return false;
   const last = row.queuedAt === undefined ? -Infinity : ms(row.queuedAt);
   const verdict = runs.find((run) => run.runId === row.runId);
@@ -281,7 +297,7 @@ export function requeue(plan: RepoPlan, row: PrTriageRow, reason: string, now: D
 }
 
 /** Queues due heads across the tenant's repositories, never-triaged ones first, while fewer than `maxInFlight` heads are queued or running, a manager waiting for CI aside. */
-export function planTenant({ repos, runs, now, policy }: TenantPlanInput): Map<string, RepoPlan> {
+export function planTenant({ repos, runs, roomless, now, policy }: TenantPlanInput): Map<string, RepoPlan> {
   const at = now.toISOString();
   const plans = new Map<string, RepoPlan>();
   const due: Due[] = [];
@@ -294,18 +310,22 @@ export function planTenant({ repos, runs, now, policy }: TenantPlanInput): Map<s
       const key = runKey(pr.number, pr.headSha);
       const current = fresh(pr, byHead.get(key), at);
       const runsOfHead = observed?.get(key) ?? [];
-      const row = uncapped(observe(current, runsOfHead, now, policy), now, policy);
+      const row = uncapped(observe(current, runsOfHead, roomless.has(pullKey(name, pr.number)), now, policy), now, policy);
       if (inFlight(row) && !parked(runsOfHead)) busy.add(`${name} ${key}`);
       if (isDue(row, pr, runsOfHead, now, policy)) due.push({ repo: name, index: next.length, row });
       next.push(row);
     }
     plans.set(name, { rows: next, enqueue: [] });
   }
-  for (const { repo, index, row } of due.sort(byWait).slice(0, Math.max(policy.maxInFlight - busy.size, 0))) {
+  // A head waiting for capacity already holds its slot, so it asks again whatever the others leave free.
+  const holding = due.filter((entry) => inFlight(entry.row));
+  const next = due.filter((entry) => !inFlight(entry.row)).sort(byWait).slice(0, Math.max(policy.maxInFlight - busy.size, 0));
+  for (const { repo, index, row } of [...holding, ...next]) {
     const plan = plans.get(repo)!;
-    const { runId: _run, error: _error, unrouted, ...rest } = row;
-    const attempts = unrouted ? row.attempts : row.attempts + 1;
-    plan.rows[index] = { ...rest, status: "queued", attempts, queuedAt: at, updatedAt: at };
+    const { runId: _run, error, unrouted, ...rest } = row;
+    const roomless = error === WAITING_FOR_CAPACITY;
+    const attempts = unrouted || roomless ? row.attempts : row.attempts + 1;
+    plan.rows[index] = { ...rest, status: "queued", attempts, queuedAt: at, updatedAt: at, ...(roomless && { error }) };
     const reason = row.error ?? "never triaged";
     const noWait = decideNow(attempts, policy);
     plan.enqueue.push({ number: row.number, reason, ...(noWait && { noWait }) });