Skip to content

Commit a317159

Browse files
Merge pull request #644 from corbitsdev/cl-11525-read-verdicts-from-the-managers
refactor(hub): read verdicts and catch-up from the managers
2 parents fc8f1a8 + 180d930 commit a317159

40 files changed

Lines changed: 321 additions & 2077 deletions

‎apps/hub/src/github/active-tenants.postgres.test.ts‎

Lines changed: 0 additions & 37 deletions
This file was deleted.

‎apps/hub/src/github/batch-pairing.test.ts‎

Lines changed: 0 additions & 278 deletions
This file was deleted.

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -375,7 +375,7 @@ test("a paused-label head whose pull request cannot be read is left failed, and
375375
expect(mails).toEqual([]);
376376
expect(state.rows()).toEqual([expect.objectContaining({ status: "failed", error: "delivery failed: Error: 502 Bad Gateway" })]);
377377
const pr = { number: 8, headSha: "sha1", updatedAt: new Date(START).toISOString(), draft: false };
378-
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, workflowVersion: undefined });
378+
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 });
379379
expect(plan.get(REPO)?.enqueue.map((entry) => entry.number)).toEqual([8]);
380380
});
381381

‎apps/hub/src/github/deployment.ts‎

Lines changed: 6 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { and, desc, eq, gt, inArray, isNotNull, isNull, ne, or, sql } from "drizzle-orm";
1+
import { and, desc, eq, inArray, isNotNull, isNull, sql } from "drizzle-orm";
22
import { schema, type DB, type ResolvedSenderPrincipal, type WorkflowRunLaunchSpecStore } from "@intx/db";
33
import { MAIL_ADMISSION_DENIED_CODE } from "@intx/hub-sessions";
44
import type { RunTriggerMaterialize } from "@corbits/webhooks";
@@ -50,12 +50,8 @@ export function systemSenderMaterializer(materialize: RunTriggerMaterialize, sen
5050
};
5151
}
5252

53-
/**
54-
* `cancelling` once cancellation was requested: the run stays live until the lifecycle sweep stops it. `workflowVersion`
55-
* once read from its launch spec, for one pinned from the package registry; `offeringId`, the decision-model offering
56-
* the deployment resolved, once read from its launch spec.
57-
*/
58-
export type LiveDeployment = { runId: string; address: string; createdAt: Date; cancelling: boolean; endedAt?: Date; workflowVersion?: string; offeringId?: string };
53+
/** `cancelling` once cancellation was requested: the run stays live until the lifecycle sweep stops it. */
54+
export type LiveDeployment = { runId: string; address: string; createdAt: Date; cancelling: boolean };
5955

6056
export type TenantDeployment = LiveDeployment & { workflow: string };
6157

@@ -72,26 +68,24 @@ function markedAs(kind: DeploymentKind) {
7268
return kind === "worker" ? eq(marker, "true") : isNull(marker);
7369
}
7470

75-
type DeploymentQuery = { tenantId: string; definitionName?: string; kind: DeploymentKind; endedSince?: Date };
71+
type DeploymentQuery = { tenantId: string; definitionName?: string; kind: DeploymentKind };
7672

7773
/**
7874
* A workflow the portal publishes per person deploys from an asset named for its publisher and release, which names
7975
* its definition too; its frozen projection keeps the workflow's own id, so that is the name a deployment is found by.
8076
*/
8177
export const workflowName = sql<string>`coalesce(${schema.workflowRunLaunchSpec.frozenApprovalBundle} -> 'projection' ->> 'id', ${schema.workflowDefinition.name})`;
8278

83-
/** Matches how the hub lists deployments: a deployment is its anchor run. Newest first; live, or ended since `endedSince`. */
79+
/** Matches how the hub lists deployments: a deployment is its anchor run. Newest first, live only. */
8480
async function deploymentsWhere(db: DB["db"], query: DeploymentQuery): Promise<TenantDeployment[]> {
8581
const { workflowRun, workflowDefinition, workflowRunLaunchSpec, tenant, liveWorkflowRunStatuses } = schema;
86-
const live = inArray(workflowRun.status, [...liveWorkflowRunStatuses]);
8782
const rows = await db
8883
.select({
8984
runId: workflowRun.id,
9085
workflow: workflowName,
9186
domain: tenant.domain,
9287
createdAt: workflowRun.createdAt,
9388
cancellationRequestedAt: workflowRun.cancellationRequestedAt,
94-
endedAt: workflowRun.endedAt,
9589
})
9690
.from(workflowRun)
9791
.innerJoin(workflowDefinition, eq(workflowRun.definitionId, workflowDefinition.id))
@@ -102,7 +96,7 @@ async function deploymentsWhere(db: DB["db"], query: DeploymentQuery): Promise<T
10296
query.definitionName === undefined ? undefined : eq(workflowName, query.definitionName),
10397
isNotNull(workflowRun.anchorRunId),
10498
eq(workflowRun.id, workflowRun.anchorRunId),
105-
query.endedSince === undefined ? live : or(live, gt(workflowRun.endedAt, query.endedSince)),
99+
inArray(workflowRun.status, [...liveWorkflowRunStatuses]),
106100
markedAs(query.kind),
107101
))
108102
.orderBy(desc(workflowRun.createdAt), desc(workflowRun.id));
@@ -111,7 +105,6 @@ async function deploymentsWhere(db: DB["db"], query: DeploymentQuery): Promise<T
111105
address: `${row.runId}@${row.domain}`,
112106
createdAt: row.createdAt,
113107
cancelling: row.cancellationRequestedAt !== null,
114-
...(row.endedAt !== null && { endedAt: row.endedAt }),
115108
workflow: row.workflow,
116109
}));
117110
}
@@ -121,34 +114,6 @@ export function resolveLiveDeployments(db: DB["db"], tenantId: string, definitio
121114
return deploymentsWhere(db, { tenantId, definitionName, kind });
122115
}
123116

124-
/** Workers live or released since `endedSince`: a worker's one body run landed in its log before its anchor was cancelled, so the log outlives the deployment. */
125-
export function recentWorkers(db: DB["db"], tenantId: string, definitionName: string, endedSince: Date): Promise<TenantDeployment[]> {
126-
return deploymentsWhere(db, { tenantId, definitionName, kind: "worker", endedSince });
127-
}
128-
129-
export type ActiveTenant = { id: string; domain: string };
130-
131-
/**
132-
* Tenants with a child run of the definition in flight, or ended or created since `endedSince`: a run inserted already
133-
* completed, as a Bitbucket run is, keeps a null `endedAt` (stock leaves a terminal run routable that way), so its
134-
* creation time is what says it just ended, and its verdict is recorded within a pass.
135-
*/
136-
export async function activeTenants(db: DB["db"], definitionName: string, what: "runs" | "deployments", endedSince?: Date): Promise<ActiveTenant[]> {
137-
const { workflowRun, workflowDefinition, tenant, liveWorkflowRunStatuses } = schema;
138-
const running = eq(workflowRun.status, "running");
139-
const childRuns = endedSince === undefined ? running : or(running, gt(workflowRun.endedAt, endedSince), gt(workflowRun.createdAt, endedSince));
140-
return db
141-
.selectDistinct({ id: tenant.id, domain: tenant.domain })
142-
.from(workflowRun)
143-
.innerJoin(workflowDefinition, eq(workflowRun.definitionId, workflowDefinition.id))
144-
.innerJoin(tenant, eq(workflowRun.tenantId, tenant.id))
145-
.where(and(
146-
eq(workflowDefinition.name, definitionName),
147-
what === "runs" ? ne(workflowRun.id, workflowRun.anchorRunId) : eq(workflowRun.id, workflowRun.anchorRunId),
148-
what === "runs" ? childRuns : inArray(workflowRun.status, [...liveWorkflowRunStatuses]),
149-
));
150-
}
151-
152117
export function newestLive(deployments: readonly LiveDeployment[]): LiveDeployment | undefined {
153118
return deployments.find((live) => !live.cancelling);
154119
}
@@ -201,20 +166,6 @@ function pinFor(bundle: FrozenApprovalBundle): string | undefined {
201166
return pinOf(bundle.closure);
202167
}
203168

204-
export async function withWorkflowVersions<T extends LiveDeployment>(
205-
launchSpecs: Pick<WorkflowRunLaunchSpecStore, "get">,
206-
deployments: readonly T[],
207-
): Promise<T[]> {
208-
const versioned: T[] = [];
209-
for (const deployment of deployments) {
210-
const spec = await launchSpecs.get(deployment.runId);
211-
const workflowVersion = spec ? closureVersions(spec.frozenApprovalBundle)?.workflow.version : undefined;
212-
const offeringId = spec?.defaultSourceOfferingId ?? undefined;
213-
versioned.push({ ...deployment, ...(workflowVersion !== undefined && { workflowVersion }), ...(offeringId !== undefined && { offeringId }) });
214-
}
215-
return versioned;
216-
}
217-
218169
/** The reason the hub gave when it replaced a busy deployment with a fresh copy, which earlier audit entries carry. */
219170
export const ROTATION_REASON = "Replaced by a fresh deployment of the same workflow";
220171

‎apps/hub/src/github/github-read-access.postgres.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -308,7 +308,7 @@ test("PostgreSQL: a code review is sent once per head as one COMMENT review, inl
308308
test("PostgreSQL: every verdict route needs workflow-run read, which holding the GitHub credential does not give", async () => {
309309
const unused = {} as never;
310310
const app = guardedApp(handle.db, sessionOf, grants, {
311-
triageVerdicts: createTriageVerdicts({ ...deps, list, verdicts: unused, observer: unused, ready: unused, versions: unused, openPulls: unused, triageState: unused, now: unused }),
311+
triageVerdicts: createTriageVerdicts({ ...deps, list, verdicts: unused, managerRuns: unused, openPulls: unused, triageState: unused, now: unused }),
312312
triageTimeline: createTriageTimeline({ verdicts: unused }),
313313
triageStats: createTriageStats({ verdicts: unused, outcomes: unused, now: unused, log: unused }),
314314
audit: createAuditLog({ db: unused, artifacts: unused, readCheckPack: unused, runReader: unused, sameSource: unused, authorize: unused, workflows: ["pr-triage"] }),

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

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ import { createGithubPrTriage } from "./pr-triage.js";
1212
import { DEFAULT_RECONCILE_POLICY } from "./reconcile-plan.js";
1313
import { createTriageReconciler } from "./triage-reconciler.js";
1414
import { createTriageStateStore, type TriageStateStore } from "./triage-state-store.js";
15-
import type { ObservedRuns } from "./triage-runs.js";
15+
import type { ObservedRuns } from "./manager-runs.js";
1616

1717
/** No deployment holds the credential in these tests. */
1818
async function noKeyPush(): Promise<void> {}
@@ -39,7 +39,7 @@ function stubDb() {
3939
}
4040

4141
async function noRuns(): Promise<ObservedRuns> {
42-
return { byRepo: new Map(), runCount: 0, verdicts: [], approved: [] };
42+
return { byRepo: new Map() };
4343
}
4444

4545
function route(store: TriageStateStore, deliver: Deliver) {
@@ -67,7 +67,6 @@ function reconciler(store: TriageStateStore, heads: number[], deliver: Deliver,
6767
listener: async () => DEPLOYMENT,
6868
openHeadsFor: async () => async () => heads.map((number) => ({ number, headSha: number === 8 ? "abc123" : `sha${number}`, updatedAt: LONG_AGO, draft: false })),
6969
activity: { stale: () => true },
70-
observeTenant: async () => ({ deployments: [DEPLOYMENT], observed: [await noRuns()], workers: [] }),
7170
managerRuns: noRuns,
7271
headChecks: async () => checks,
7372
verdicts: { latest: async () => [] },

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

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -80,10 +80,10 @@ test("PostgreSQL: the reconciler does not mail a head again while a live manager
8080
const queued: PrTriageRow = { number: 7, headSha: "h1", status: "queued", attempts: 1, firstSeenAt: queuedAt, queuedAt, updatedAt: queuedAt };
8181
function plan(rows: PrTriageRow[], runs: Awaited<ReturnType<typeof managerRuns>>) {
8282
const ci = new Map([[runKey(8, "h1"), "pending" as const]]);
83-
return planTenant({ repos: [{ name: REPO, prs: [open(7, now), open(8, now)], rows, ci }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY, workflowVersion: "1.0.0" }).get(REPO)!;
83+
return planTenant({ repos: [{ name: REPO, prs: [open(7, now), open(8, now)], rows, ci }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!;
8484
}
8585

86-
const running = plan([queued], await managerRuns(TENANT, [pr7, pr8], "1.0.0"));
86+
const running = plan([queued], await managerRuns(TENANT, [pr7, pr8]));
8787
expect(running.rows.find((row) => row.number === 7)).toMatchObject({ status: "running", runId: "run_triaged" });
8888
expect(running.enqueue.map((queuedPr) => queuedPr.number)).not.toContain(7);
8989

@@ -92,9 +92,9 @@ test("PostgreSQL: the reconciler does not mail a head again while a live manager
9292
await take(db, TENANT, held, "run_held");
9393
await done(db, TENANT, held, "run_held", [], { head: "h1", waiting: true });
9494
const firstSeenAt = new Date(now.getTime() - MINUTE).toISOString();
95-
const fresh: PrTriageRow = { number: 8, headSha: "h1", status: "new", attempts: 0, workflowVersion: "1.0.0", firstSeenAt, updatedAt: firstSeenAt };
95+
const fresh: PrTriageRow = { number: 8, headSha: "h1", status: "new", attempts: 0, firstSeenAt, updatedAt: firstSeenAt };
9696

97-
const settled = plan([...running.rows.filter((row) => row.number === 7), fresh], await managerRuns(TENANT, [pr7, pr8], "1.0.0"));
97+
const settled = plan([...running.rows.filter((row) => row.number === 7), fresh], await managerRuns(TENANT, [pr7, pr8]));
9898
expect(settled.rows).toEqual([expect.objectContaining({ number: 7, status: "triaged", runId: "run_triaged" }), expect.objectContaining({ number: 8, status: "waiting", runId: "run_held" })]);
9999
expect(settled.enqueue).toEqual([]);
100100
});
@@ -111,8 +111,8 @@ test("PostgreSQL: on the listener path a head is mailed once while its manager r
111111
let mails = 0;
112112
async function reconcile(afterMs: number): Promise<PrTriageRow> {
113113
now = new Date(start.getTime() + afterMs);
114-
const runs = await managerRuns(TENANT, [pr], undefined);
115-
const plan = planTenant({ repos: [{ name: REPO, prs: [open(7, start)], rows }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY, workflowVersion: undefined }).get(REPO)!;
114+
const runs = await managerRuns(TENANT, [pr]);
115+
const plan = planTenant({ repos: [{ name: REPO, prs: [open(7, start)], rows }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!;
116116
rows = plan.rows;
117117
mails += plan.enqueue.length;
118118
return rows[0]!;
@@ -143,11 +143,11 @@ test("PostgreSQL: a manager that went away mid-pass fails its pass's head, so th
143143
await take(db, TENANT, pr7, "run_gone");
144144
await db.update(schema.workflowRun).set({ status: "failed" }).where(sql`${schema.workflowRun.id} = 'run_gone'`);
145145

146-
const runs = await managerRuns(TENANT, [pr7], undefined);
146+
const runs = await managerRuns(TENANT, [pr7]);
147147
expect(runs.byRepo.get(REPO)?.get(runKey(7, "h1"))).toEqual([expect.objectContaining({ runId: "run_gone", status: "failed" })]);
148148
const queuedAt = new Date(now.getTime() - MINUTE).toISOString();
149149
const queued: PrTriageRow = { number: 7, headSha: "h1", status: "queued", attempts: 1, firstSeenAt: queuedAt, queuedAt, updatedAt: queuedAt };
150-
const plan = planTenant({ repos: [{ name: REPO, prs: [open(7, now)], rows: [queued] }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY, workflowVersion: undefined }).get(REPO)!;
150+
const plan = planTenant({ repos: [{ name: REPO, prs: [open(7, now)], rows: [queued] }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!;
151151
expect(plan.rows).toEqual([expect.objectContaining({ status: "failed", runId: "run_gone", error: "run failed" })]);
152152
expect(plan.enqueue).toEqual([]);
153153
});

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

Lines changed: 10 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -6,37 +6,38 @@ import type { DB } from "@intx/db";
66
import { unsettledBy } from "@corbits/triage-workflows/logic/settled";
77
import { managedPulls, type PullRequestRef } from "../manager-state.js";
88
import { runKey, type ObservedRun } from "./reconcile-plan.js";
9-
import type { ObservedRuns } from "./triage-runs.js";
109
import type { VerdictRecord, VerdictRecordStore } from "./verdict-record-store.js";
1110

12-
/** The runs of `pulls`' managers; `workflowVersion` is the version new mail goes to, which a manager's runs count as. */
13-
export type ManagerRuns = (tenantId: string, pulls: readonly PullRequestRef[], workflowVersion: string | undefined) => Promise<ObservedRuns>;
11+
/** Runs by repository, then by `${number}@${headSha}`. */
12+
export type ObservedRuns = { byRepo: Map<string, Map<string, ObservedRun[]>> };
13+
14+
/** The runs of `pulls`' managers. */
15+
export type ManagerRuns = (tenantId: string, pulls: readonly PullRequestRef[]) => Promise<ObservedRuns>;
1416

1517
function settlesBy(verdict: VerdictRecord): Pick<ObservedRun, "unsettled" | "unconfirmed"> {
1618
const { detail } = verdict;
1719
return unsettledBy({ degraded: verdict.degraded, reason: detail?.reason, headSha: detail?.headSha, state: verdict.state, checks: detail?.checks }, verdict.headSha);
1820
}
1921

2022
export function createManagerRuns(db: DB["db"], verdicts: Pick<VerdictRecordStore, "ofRuns">, now: () => Date): ManagerRuns {
21-
return async function managerRuns(tenantId, asked, workflowVersion) {
23+
return async function managerRuns(tenantId, asked) {
2224
const pulls = await managedPulls(db, tenantId, asked);
2325
const recorded = await verdicts.ofRuns(tenantId, [...new Set(pulls.flatMap((pull) => pull.history.map((pass) => pass.runId)))], asked);
24-
const stamp = workflowVersion === undefined ? {} : { workflowVersion };
2526
const byRepo: ObservedRuns["byRepo"] = new Map();
2627

2728
function observed(repo: string, number: number, head: string, run: ObservedRun): void {
2829
const byHead = byRepo.get(repo) ?? new Map<string, ObservedRun[]>();
2930
byRepo.set(repo, byHead);
3031
const key = runKey(number, head);
3132
const listed = byHead.get(key);
32-
if (listed) listed.push({ ...run, ...stamp });
33-
else byHead.set(key, [{ ...run, ...stamp }]);
33+
if (listed) listed.push(run);
34+
else byHead.set(key, [run]);
3435
}
3536

3637
for (const { pr, history, live, gone } of pulls) {
3738
const { source } = pr;
3839
// The hub checked the run is up, so it is running now and never stuck.
39-
if (live?.head !== undefined) observed(pr.repo, pr.number, live.head, { runId: live.runId, source, status: "running", startedAt: now().toISOString() });
40+
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 }) });
4041
if (gone) observed(pr.repo, pr.number, gone.pass.head, { runId: gone.runId, source, status: "failed", startedAt: gone.pass.takenAt });
4142
// A head's last pass decides whether it still waits for CI; a waiting run is listed first so it outranks an older verdict.
4243
const lastOfHead = new Map(history.map((pass) => [pass.head, pass]));
@@ -47,6 +48,6 @@ export function createManagerRuns(db: DB["db"], verdicts: Pick<VerdictRecordStor
4748
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 }) });
4849
}
4950
}
50-
return { byRepo, runCount: 0, verdicts: [], approved: [] };
51+
return { byRepo };
5152
};
5253
}

0 commit comments

Comments
 (0)