diff --git a/apps/dispatcher/src/check-name.test.ts b/apps/dispatcher/src/check-name.test.ts index a8d0dd5..43dc951 100644 --- a/apps/dispatcher/src/check-name.test.ts +++ b/apps/dispatcher/src/check-name.test.ts @@ -5,7 +5,7 @@ // `checkLabel` produces a distinct, separately-requirable name. import { describe, expect, it } from "vitest"; -import { checkRunNameFor } from "./check-name"; +import { checkRunNameFor, checkRunTitleFor } from "./check-name"; describe("checkRunNameFor", () => { it("names a labelless dispatch exactly as before — `flare-dispatch/`", () => { @@ -74,3 +74,15 @@ describe("checkRunNameFor", () => { expect(checkRunNameFor("check", { checkLabel: label })).toBe(`flare-dispatch/check:${label}`); }); }); + +describe("checkRunTitleFor", () => { + it("is the bare check name on attempt 1", () => { + expect(checkRunTitleFor("flare-dispatch/check", 1)).toBe("flare-dispatch/check"); + }); + + it("names the attempt on a re-run", () => { + expect(checkRunTitleFor("flare-dispatch/check:codegen", 3)).toBe( + "flare-dispatch/check:codegen (attempt 3)", + ); + }); +}); diff --git a/apps/dispatcher/src/check-name.ts b/apps/dispatcher/src/check-name.ts index 5801086..eb308ab 100644 --- a/apps/dispatcher/src/check-name.ts +++ b/apps/dispatcher/src/check-name.ts @@ -68,3 +68,12 @@ export const checkRunNameFor = (run: string, inputs: unknown): string => { const label = readCheckLabel(inputs); return label === undefined ? `flare-dispatch/${run}` : `flare-dispatch/${run}:${label}`; }; + +/** + * The check-run output TITLE for an execution: the check name, plus + * ` (attempt N)` on a re-run. The NAME stays fixed across attempts — branch + * protection requires checks by name, and GitHub counts the latest check-run + * of a name — so the attempt lives in the title a reviewer reads instead. + */ +export const checkRunTitleFor = (checkRunName: string, attempt: number): string => + attempt > 1 ? `${checkRunName} (attempt ${attempt})` : checkRunName; diff --git a/apps/dispatcher/src/executions-read.ts b/apps/dispatcher/src/executions-read.ts index e3b4342..2a90b79 100644 --- a/apps/dispatcher/src/executions-read.ts +++ b/apps/dispatcher/src/executions-read.ts @@ -24,6 +24,11 @@ export type ExecutionRow = { readonly input_json: string; readonly summary_json: string | null; readonly check_run_id: number | null; + // --- Attempt lineage (infra/migrations/0007) -------------------------------- + /** 1 for a dispatched execution, N for its (N-1)th check-run re-run. */ + readonly attempt: number; + /** The id of attempt 1 of this row's family; NULL on attempt 1 itself. */ + readonly retry_of: string | null; // --- Cost rollup (infra/migrations/0005; written at finishExecution) -------- // All nullable: pre-0005 rows, still-running executions, and deploys without // the cost path leave these NULL. @@ -121,6 +126,50 @@ export const listExecutions = async ( export const getExecution = async (db: D1Database, id: string): Promise => db.prepare("SELECT * FROM executions WHERE id = ?").bind(id).first(); +/** + * The execution that posted a check-run, by the GitHub check-run id — or + * `null` when no execution on this deploy posted it. Scoped to `repo` as well: + * the repo named by the webhook is the one the re-run is authorized against, + * so a row from any other repo must not answer. + */ +export const getExecutionByCheckRun = async ( + db: D1Database, + repo: string, + checkRunId: number, +): Promise => + db + .prepare("SELECT * FROM executions WHERE check_run_id = ? AND repo = ?") + .bind(checkRunId, repo) + .first(); + +/** + * Every attempt of one execution family: the root (attempt 1) plus each + * re-run pointing at it through `retry_of`. Unordered; empty when the root id + * names no row. + */ +export const getAttemptFamily = async (db: D1Database, rootId: string): Promise => { + const root = await getExecution(db, rootId); + if (root === null) return []; + const { results } = await db + .prepare("SELECT * FROM executions WHERE retry_of = ?") + .bind(rootId) + .all(); + return [root, ...(results ?? [])]; +}; + +/** Every execution recorded against one commit of one repo. */ +export const listExecutionsAtSha = async ( + db: D1Database, + repo: string, + sha: string, +): Promise => { + const { results } = await db + .prepare("SELECT * FROM executions WHERE repo = ? AND sha = ?") + .bind(repo, sha) + .all(); + return results ?? []; +}; + /** Fetch an execution's steps, ordered by start time then name. */ export const getSteps = async (db: D1Database, executionId: string): Promise => { const { results } = await db diff --git a/apps/dispatcher/src/instantiate.ts b/apps/dispatcher/src/instantiate.ts index c8f36e7..e2aa049 100644 --- a/apps/dispatcher/src/instantiate.ts +++ b/apps/dispatcher/src/instantiate.ts @@ -2,7 +2,8 @@ // // Extracted from `routes/dispatch.ts` so EVERY trigger surface that ends in a // `RUNS_WORKFLOW.create` — Action mode (`routes/dispatch.ts`), Schedule mode -// (`routes/scheduled.ts`), and signal ingress (`routes/signals-webhook.ts`) — +// (`routes/scheduled.ts`), signal ingress (`routes/signals-webhook.ts`), and +// check-run re-runs (`rerequest.ts`) — // drives the exact same execution path: // // 1. receiver-level dedup short-circuit on `IDEMPOTENCY_KV` (when bound), @@ -43,6 +44,10 @@ export interface InstantiateArgs { readonly notify?: { readonly emails: readonly string[] }; /** Absolute origin so the run's artifact URLs come back absolute. */ readonly origin: string; + /** Attempt number of a check-run re-run (rerequest.ts); absent → 1. */ + readonly attempt?: number; + /** The id of attempt 1 of the family a re-run retries. */ + readonly retryOf?: string; } /** The outcome of an instantiation — the bits the 202 response carries. */ @@ -99,6 +104,8 @@ export const instantiateRun = async ( ? { notify: { emails: args.notify.emails } } : {}), origin: args.origin, + ...(args.attempt !== undefined ? { attempt: args.attempt } : {}), + ...(args.retryOf !== undefined ? { retryOf: args.retryOf } : {}), }; // CF Workflows rejects a `create({id})` whose id was seen before with diff --git a/apps/dispatcher/src/rerequest.test.ts b/apps/dispatcher/src/rerequest.test.ts new file mode 100644 index 0000000..cb65f1a --- /dev/null +++ b/apps/dispatcher/src/rerequest.test.ts @@ -0,0 +1,302 @@ +// FlareDispatch Dispatcher — check-run re-run acceptance tests. +// +// Drives `POST /v1/webhooks/github` through the router with signed +// `check_run.rerequested` / `check_suite.rerequested` deliveries against a +// seeded executions table, and asserts what reaches `RUNS_WORKFLOW.create`. + +import { describe, expect, it } from "vitest"; +import { sign } from "./hmac"; +import { MAX_ATTEMPTS, retryExecutionId } from "./rerequest"; +import { handleRequest } from "./router"; +import { makeFakeD1, makeFakeEnv, makeFakeKv, makeFakeR2, makeFakeWorkflow } from "./test-helpers"; + +const WEBHOOK_SECRET = "github-webhook-secret-please-rotate"; +const APP_ID = "4242"; +const REPO = "owner/test-repo"; +const SHA = "0123456789abcdef0123456789abcdef01234567"; +const ROOT = "check_owner_test-repo_0123456789ab"; + +/** A recorded `executions` row, snake_case like D1 returns it. */ +const row = (over: Record): Record => ({ + id: ROOT, + run: "check", + repo: REPO, + ref: "refs/heads/main", + sha: SHA, + status: "failure", + started_at: 1_000, + completed_at: 2_000, + parent_execution_id: null, + input_json: JSON.stringify({ repo: REPO, sha: SHA, install: false, secrets: [] }), + summary_json: null, + check_run_id: 555, + attempt: 1, + retry_of: null, + ...over, +}); + +const checkRunPayload = (over: { checkRunId?: number; appId?: number } = {}) => ({ + action: "rerequested", + check_run: { + id: over.checkRunId ?? 555, + head_sha: SHA, + name: "flare-dispatch/check", + app: { id: over.appId ?? Number(APP_ID) }, + }, + repository: { full_name: REPO }, + installation: { id: 99999 }, +}); + +const deliver = async ( + env: ReturnType, + event: string, + payload: unknown, + deliveryId = crypto.randomUUID(), +): Promise<{ status: number; body: Record }> => { + const bodyText = JSON.stringify(payload); + const res = await handleRequest( + new Request("https://dispatcher.example/v1/webhooks/github", { + method: "POST", + headers: { + "content-type": "application/json", + "X-GitHub-Event": event, + "X-GitHub-Delivery": deliveryId, + "X-Hub-Signature-256": await sign(WEBHOOK_SECRET, new TextEncoder().encode(bodyText)), + }, + body: bodyText, + }), + env, + ); + return { status: res.status, body: (await res.json()) as Record }; +}; + +const fixture = (opts: { + executions: Record[]; + /** Workflow instance status per id; unlisted ids have no instance. */ + instances?: Record; +}) => { + const workflow = makeFakeWorkflow({ + instanceStatus: (id) => opts.instances?.[id], + }); + const metadata = makeFakeD1({ executions: opts.executions }); + const env = makeFakeEnv({ + hmacSecret: "unused", + workflow, + storage: makeFakeR2(), + idempotencyKv: makeFakeKv().binding, + githubWebhookSecret: WEBHOOK_SECRET, + githubAppId: APP_ID, + metadata, + }); + return { env, workflow }; +}; + +describe("check_run.rerequested", () => { + it("re-dispatches the execution that posted the check as attempt 2", async () => { + const { env, workflow } = fixture({ + executions: [row({})], + instances: { [ROOT]: "errored" }, + }); + + const { status, body } = await deliver(env, "check_run", checkRunPayload()); + + expect(status).toBe(202); + const expectedId = retryExecutionId(ROOT, 2); + expect(body["rerun"]).toEqual([ + { kind: "dispatched", run: "check", executionId: expectedId, attempt: 2, retryOf: ROOT }, + ]); + expect(workflow.calls).toHaveLength(1); + expect(workflow.calls[0]).toEqual({ + id: expectedId, + params: { + executionId: expectedId, + run: "check", + github: { + repo: REPO, + ref: "refs/heads/main", + sha: SHA, + installation_id: 99999, + }, + inputs: { repo: REPO, sha: SHA, install: false, secrets: [] }, + origin: "https://dispatcher.example", + attempt: 2, + retryOf: ROOT, + }, + }); + }); + + it("numbers past every recorded attempt, whichever check-run was clicked", async () => { + // The OLD (attempt-1) check-run is re-run after attempt 2 already failed. + const second = retryExecutionId(ROOT, 2); + const { env, workflow } = fixture({ + executions: [row({}), row({ id: second, attempt: 2, retry_of: ROOT, check_run_id: 556 })], + instances: { [ROOT]: "complete", [second]: "complete" }, + }); + + const { body } = await deliver(env, "check_run", checkRunPayload({ checkRunId: 555 })); + + expect(body["rerun"]).toMatchObject([{ kind: "dispatched", attempt: 3, retryOf: ROOT }]); + expect(workflow.calls.map((c) => c.id)).toEqual([retryExecutionId(ROOT, 3)]); + }); + + it("refuses while an attempt is live — a click storm dispatches once", async () => { + const second = retryExecutionId(ROOT, 2); + const { env, workflow } = fixture({ + executions: [row({})], + instances: { [ROOT]: "complete" }, + }); + + // First click dispatches attempt 2; the platform now reports it queued. + await deliver(env, "check_run", checkRunPayload()); + const statuses = { [ROOT]: "complete", [second]: "queued" }; + const stormEnv = { + ...env, + RUNS_WORKFLOW: makeFakeWorkflow({ instanceStatus: (id) => statuses[id] }).binding, + }; + const results = await Promise.all( + [1, 2, 3].map(() => deliver(stormEnv, "check_run", checkRunPayload())), + ); + + expect(workflow.calls).toHaveLength(1); + for (const { body } of results) { + expect(body["rerun"]).toEqual([ + { kind: "refused", reason: "in_progress", executionId: second }, + ]); + } + }); + + it("refuses while the recorded execution's Workflow is still running", async () => { + const { env, workflow } = fixture({ + executions: [row({ status: "running", completed_at: null })], + instances: { [ROOT]: "running" }, + }); + + const { body } = await deliver(env, "check_run", checkRunPayload()); + + expect(body["rerun"]).toEqual([{ kind: "refused", reason: "in_progress", executionId: ROOT }]); + expect(workflow.calls).toHaveLength(0); + }); + + it("retries a row stuck at `running` whose Workflow died with its container", async () => { + const { env, workflow } = fixture({ + executions: [row({ status: "running", completed_at: null })], + instances: { [ROOT]: "errored" }, + }); + + const { body } = await deliver(env, "check_run", checkRunPayload()); + + expect(body["rerun"]).toMatchObject([{ kind: "dispatched", attempt: 2 }]); + expect(workflow.calls).toHaveLength(1); + }); + + it("steps past an attempt the platform accepted but that never recorded a row", async () => { + const second = retryExecutionId(ROOT, 2); + const { env, workflow } = fixture({ + executions: [row({})], + instances: { [ROOT]: "complete", [second]: "errored" }, + }); + + const { body } = await deliver(env, "check_run", checkRunPayload()); + + expect(body["rerun"]).toMatchObject([{ kind: "dispatched", attempt: 3 }]); + expect(workflow.calls.map((c) => c.id)).toEqual([retryExecutionId(ROOT, 3)]); + }); + + it(`refuses after ${MAX_ATTEMPTS} attempts`, async () => { + const last = retryExecutionId(ROOT, MAX_ATTEMPTS); + const { env, workflow } = fixture({ + executions: [row({}), row({ id: last, attempt: MAX_ATTEMPTS, retry_of: ROOT })], + }); + + const { body } = await deliver(env, "check_run", checkRunPayload()); + + expect(body["rerun"]).toEqual([ + { kind: "refused", reason: "attempts_exhausted", executionId: last }, + ]); + expect(workflow.calls).toHaveLength(0); + }); + + it("refuses a check-run another App posted", async () => { + const { env, workflow } = fixture({ executions: [row({})] }); + + const { body } = await deliver(env, "check_run", checkRunPayload({ appId: 7 })); + + expect(body["rerun"]).toEqual([{ kind: "refused", reason: "foreign_app" }]); + expect(workflow.calls).toHaveLength(0); + }); + + it("refuses a check-run no execution on this deploy posted", async () => { + const { env, workflow } = fixture({ executions: [row({})] }); + + const { body } = await deliver(env, "check_run", checkRunPayload({ checkRunId: 1 })); + + expect(body["rerun"]).toEqual([{ kind: "refused", reason: "unknown_check_run" }]); + expect(workflow.calls).toHaveLength(0); + }); + + it("refuses when the recorded inputs no longer decode against the run", async () => { + const { env, workflow } = fixture({ + executions: [row({ input_json: JSON.stringify({ repo: "not a repo" }) })], + }); + + const { body } = await deliver(env, "check_run", checkRunPayload()); + + expect(body["rerun"]).toEqual([ + { kind: "refused", reason: "inputs_unreplayable", executionId: ROOT }, + ]); + expect(workflow.calls).toHaveLength(0); + }); + + it("leaves other check_run actions to the trigger fan-out", async () => { + const { env, workflow } = fixture({ executions: [row({})] }); + + const { body } = await deliver(env, "check_run", { ...checkRunPayload(), action: "completed" }); + + expect(body["rerun"]).toBeUndefined(); + expect(body["dispatched"]).toEqual([]); + expect(workflow.calls).toHaveLength(0); + }); +}); + +describe("check_suite.rerequested", () => { + const suitePayload = { + action: "rerequested", + check_suite: { head_sha: SHA, app: { id: Number(APP_ID) } }, + repository: { full_name: REPO }, + installation: { id: 99999 }, + }; + + it("re-runs each check at the commit whose latest attempt did not pass", async () => { + const lint = "check_lint_owner_test-repo_0123456789ab"; + const deploy = "worker-deploy_owner_test-repo_0123456789ab"; + const { env, workflow } = fixture({ + executions: [ + row({}), + row({ + id: lint, + check_run_id: 600, + input_json: JSON.stringify({ repo: REPO, sha: SHA, checkLabel: "lint" }), + }), + row({ id: deploy, run: "worker-deploy", status: "success", check_run_id: 700 }), + // A spawned child reports through its parent — never re-run on its own. + row({ id: "child", parent_execution_id: ROOT, check_run_id: 800 }), + // An uncredentialed execution posted no check. + row({ id: "silent", check_run_id: "noop" }), + ], + }); + + const { body } = await deliver(env, "check_suite", suitePayload); + + expect(body["rerun"]).toEqual( + expect.arrayContaining([ + expect.objectContaining({ kind: "dispatched", retryOf: ROOT, attempt: 2 }), + expect.objectContaining({ kind: "dispatched", retryOf: lint, attempt: 2 }), + { kind: "succeeded", executionId: deploy }, + ]), + ); + expect((body["rerun"] as unknown[]).length).toBe(3); + expect(workflow.calls.map((c) => c.id).sort()).toEqual( + [retryExecutionId(ROOT, 2), retryExecutionId(lint, 2)].sort(), + ); + }); +}); diff --git a/apps/dispatcher/src/rerequest.ts b/apps/dispatcher/src/rerequest.ts new file mode 100644 index 0000000..e932f0e --- /dev/null +++ b/apps/dispatcher/src/rerequest.ts @@ -0,0 +1,325 @@ +// FlareDispatch Dispatcher — re-run a check from GitHub's "Re-run" button. +// +// GitHub sends `check_run.rerequested` when someone clicks "Re-run" on a +// check-run this App posted, and `check_suite.rerequested` for "Re-run all +// checks" on its suite. Both re-dispatch work this deploy already ran, so a +// check that went red because the platform killed its container is retried +// with one click instead of a new commit. +// +// --- A re-run is a new execution --------------------------------------------- +// +// The retried execution's id cannot be reused: Cloudflare Workflows refuses +// `create({ id })` for every id it has seen, terminated or not. So attempt N +// gets its own deterministic id, `:attempt-` through `toInstanceId`, +// and carries `attempt` + `retryOf` (the root's id) into the Workflow, which +// records them on the `executions` row (infra/migrations/0007) and shows the +// attempt in the check-run title. The check-run NAME is unchanged, so the new +// check-run supersedes the red one under the same branch-protection rule. +// +// Everything the retry needs comes from the prior execution's own row — run, +// repo, ref, sha, and the decoded inputs in `input_json` — plus the +// installation id on the webhook. Metadata that row does not hold (the Slack +// origin, completion-notify recipients) is not replayed: a re-run reports on +// the check-run only. +// +// --- Guards ------------------------------------------------------------------ +// +// * foreign_app — the check-run / suite belongs to another GitHub App. +// * unknown_check_run — no execution on this deploy posted that check-run. +// * in_progress — an attempt of the same family is still live, per the +// Workflow instance's own status (a D1 `running` row +// whose instance errored does not block). This is the +// click-storm guard: N clicks while attempt 2 runs +// dispatch nothing more. +// * attempts_exhausted — the family reached MAX_ATTEMPTS. +// * run_not_registered / inputs_unreplayable — the run left the registry, or +// its recorded inputs no longer decode against it. +// +// A re-run deliberately bypasses the run's cooldown: cooldown caps what a push +// storm can dispatch, and a re-run is one explicit request on a finished +// check, bounded by `in_progress` and `MAX_ATTEMPTS` instead. + +import { Either, Option, Schema } from "effect"; +import type { Env } from "./env"; +import { + type ExecutionRow, + getAttemptFamily, + getExecutionByCheckRun, + listExecutionsAtSha, +} from "./executions-read"; +import { toInstanceId } from "./instance-id"; +import { instantiateRun } from "./instantiate"; +import { lookupRun } from "./registry"; + +/** Attempts per execution family, the first dispatch included. */ +export const MAX_ATTEMPTS = 5; + +/** Workflow instance states that mean an attempt is still live. */ +const ACTIVE_STATES: ReadonlySet = new Set([ + "queued", + "running", + "paused", + "waiting", + "waitingForPause", +]); + +/** D1 execution statuses that are final without asking the platform. */ +const TERMINAL_ROW_STATES: ReadonlySet = new Set([ + "success", + "failure", + "skipped", + "cancelled", +]); + +const App = Schema.Struct({ id: Schema.Number }); +const Repository = Schema.Struct({ full_name: Schema.String }); +const Installation = Schema.Struct({ id: Schema.Number }); + +const CheckRunRerequested = Schema.Struct({ + action: Schema.Literal("rerequested"), + check_run: Schema.Struct({ + id: Schema.Number, + head_sha: Schema.String, + app: Schema.optional(App), + }), + repository: Repository, + installation: Schema.optional(Installation), +}); + +const CheckSuiteRerequested = Schema.Struct({ + action: Schema.Literal("rerequested"), + check_suite: Schema.Struct({ + head_sha: Schema.String, + app: Schema.optional(App), + }), + repository: Repository, + installation: Schema.optional(Installation), +}); + +/** A decoded re-run request — one check-run, or every check of a commit. */ +export type Rerequest = + | { + readonly kind: "check_run"; + readonly repo: string; + readonly checkRunId: number; + readonly appId?: number; + readonly installationId?: number; + } + | { + readonly kind: "check_suite"; + readonly repo: string; + readonly headSha: string; + readonly appId?: number; + readonly installationId?: number; + }; + +/** + * Decode a webhook into a re-run request, or `undefined` when it is not one + * (any other event or action, or a `rerequested` body missing the fields a + * re-run needs). + */ +export const decodeRerequest = (event: string, payload: unknown): Rerequest | undefined => { + if (event === "check_run") { + return Option.getOrUndefined( + Option.map(Schema.decodeUnknownOption(CheckRunRerequested)(payload), (p) => ({ + kind: "check_run" as const, + repo: p.repository.full_name, + checkRunId: p.check_run.id, + ...(p.check_run.app !== undefined ? { appId: p.check_run.app.id } : {}), + ...(p.installation !== undefined ? { installationId: p.installation.id } : {}), + })), + ); + } + if (event === "check_suite") { + return Option.getOrUndefined( + Option.map(Schema.decodeUnknownOption(CheckSuiteRerequested)(payload), (p) => ({ + kind: "check_suite" as const, + repo: p.repository.full_name, + headSha: p.check_suite.head_sha, + ...(p.check_suite.app !== undefined ? { appId: p.check_suite.app.id } : {}), + ...(p.installation !== undefined ? { installationId: p.installation.id } : {}), + })), + ); + } + return undefined; +}; + +/** Why a re-run dispatched nothing. */ +export type RerunRefusal = + | "foreign_app" + | "unknown_check_run" + | "in_progress" + | "attempts_exhausted" + | "run_not_registered" + | "inputs_unreplayable"; + +/** What one re-run request did for one execution family. */ +export type RerunOutcome = + | { + readonly kind: "dispatched"; + readonly run: string; + readonly executionId: string; + readonly attempt: number; + readonly retryOf: string; + } + | { + readonly kind: "refused"; + readonly reason: RerunRefusal; + /** The execution the refusal is about, when there is one. */ + readonly executionId?: string; + } + | { + /** `check_suite` only: the family's latest attempt already succeeded. */ + readonly kind: "succeeded"; + readonly executionId: string; + }; + +/** The Workflow instance id of attempt `attempt` of the family rooted at `rootId`. */ +export const retryExecutionId = (rootId: string, attempt: number): string => + toInstanceId(`${rootId}:attempt-${attempt}`); + +/** The id of attempt 1 of the family `row` belongs to. */ +const rootOf = (row: ExecutionRow): string => row.retry_of ?? row.id; + +/** A family's latest attempt — the highest `attempt` number. */ +const latestOf = (family: readonly ExecutionRow[]): ExecutionRow => + family.reduce((a, b) => ((b.attempt ?? 1) > (a.attempt ?? 1) ? b : a)); + +/** + * The Workflow instance's status, or `undefined` when the platform knows no + * instance by that id (never created, or past retention). The binding's `get` + * is async on the platform and rejects for an unknown id. + */ +const instanceStatus = async (env: Env, id: string): Promise => { + try { + const instance = await env.RUNS_WORKFLOW.get(id); + return (await instance.status()).status; + } catch { + return undefined; + } +}; + +const isActive = (status: string | undefined): boolean => + status !== undefined && ACTIVE_STATES.has(status); + +/** + * Re-dispatch the next attempt of one execution family. `family` holds every + * recorded attempt (attempt 1 first-class among them). + */ +const rerunFamily = async ( + env: Env, + family: readonly ExecutionRow[], + installationId: number | undefined, + origin: string, +): Promise => { + const latest = latestOf(family); + const rootId = rootOf(latest); + + // A row whose D1 status is not final may still be live — ask the platform, + // which is authoritative: a container kill that took the Workflow down + // leaves the row `running` forever, and that must not block the retry. + for (const row of family) { + if (TERMINAL_ROW_STATES.has(row.status)) continue; + if (isActive(await instanceStatus(env, row.id))) { + return { kind: "refused", reason: "in_progress", executionId: row.id }; + } + } + + const run = lookupRun(latest.run); + if (run === undefined) { + return { kind: "refused", reason: "run_not_registered", executionId: latest.id }; + } + const inputs: unknown = (() => { + try { + return JSON.parse(latest.input_json); + } catch { + return undefined; + } + })(); + if (Either.isLeft(Schema.decodeUnknownEither(run.inputs)(inputs))) { + return { kind: "refused", reason: "inputs_unreplayable", executionId: latest.id }; + } + + // The next free attempt number. An instance that exists with no D1 row is + // an attempt the platform accepted but that never recorded itself: live → + // someone else's click already dispatched it; finished → it died before + // `startExecution`, so step past it rather than collapse onto a dead id. + for (let attempt = (latest.attempt ?? 1) + 1; attempt <= MAX_ATTEMPTS; attempt++) { + const executionId = retryExecutionId(rootId, attempt); + const status = await instanceStatus(env, executionId); + if (isActive(status)) { + return { kind: "refused", reason: "in_progress", executionId }; + } + if (status !== undefined) continue; + + await instantiateRun(env, { + executionId, + run: latest.run, + github: { + repo: latest.repo, + ref: latest.ref, + sha: latest.sha, + ...(installationId !== undefined ? { installation_id: installationId } : {}), + }, + inputs, + origin, + attempt, + retryOf: rootId, + }); + return { kind: "dispatched", run: latest.run, executionId, attempt, retryOf: rootId }; + } + return { kind: "refused", reason: "attempts_exhausted", executionId: latest.id }; +}; + +/** True when the request names an App other than this deploy's. */ +const isForeignApp = (env: Env, appId: number | undefined): boolean => + appId !== undefined && env.GITHUB_APP_ID !== undefined && String(appId) !== env.GITHUB_APP_ID; + +/** + * Handle a decoded re-run request: one outcome for a `check_run` re-run, one + * per check-posting execution family at the commit for a `check_suite` re-run. + */ +export const handleRerequest = async ( + env: Env, + request: Rerequest, + origin: string, +): Promise => { + if (isForeignApp(env, request.appId)) { + return [{ kind: "refused", reason: "foreign_app" }]; + } + + if (request.kind === "check_run") { + const prior = await getExecutionByCheckRun(env.RUNS_METADATA, request.repo, request.checkRunId); + if (prior === null) return [{ kind: "refused", reason: "unknown_check_run" }]; + const family = await getAttemptFamily(env.RUNS_METADATA, rootOf(prior)); + return [ + await rerunFamily(env, family.length > 0 ? family : [prior], request.installationId, origin), + ]; + } + + // `check_suite`: every top-level execution that posted a check at this + // commit, grouped into families. A family whose latest attempt succeeded is + // left alone — "re-run all" is for the checks that did not pass, and + // re-running a green review or deploy would repeat its side effects. A + // numeric `check_run_id` is what "posted a check" means: an uncredentialed + // execution records the no-op sentinel (`"noop"`) there, and a spawned child + // reports through its parent. + const rows = (await listExecutionsAtSha(env.RUNS_METADATA, request.repo, request.headSha)).filter( + (row) => typeof row.check_run_id === "number" && row.parent_execution_id == null, + ); + const families = new Map(); + for (const row of rows) { + const root = rootOf(row); + families.set(root, [...(families.get(root) ?? []), row]); + } + const outcomes: RerunOutcome[] = []; + for (const family of families.values()) { + const latest = latestOf(family); + outcomes.push( + latest.status === "success" + ? { kind: "succeeded", executionId: latest.id } + : await rerunFamily(env, family, request.installationId, origin), + ); + } + return outcomes; +}; diff --git a/apps/dispatcher/src/routes/executions.ts b/apps/dispatcher/src/routes/executions.ts index e12a313..d93b21d 100644 --- a/apps/dispatcher/src/routes/executions.ts +++ b/apps/dispatcher/src/routes/executions.ts @@ -55,6 +55,9 @@ const executionView = (row: ExecutionRow, links: { logsUrl?: string; dashboardUr completedAt: row.completed_at, ...(row.parent_execution_id !== null ? { parentExecutionId: row.parent_execution_id } : {}), ...(row.check_run_id !== null ? { checkRunId: row.check_run_id } : {}), + // `?? 1`: a row read before migration 0007 has no column at all. + attempt: row.attempt ?? 1, + ...(row.retry_of != null ? { retryOf: row.retry_of } : {}), ...(links.logsUrl !== undefined ? { logsUrl: links.logsUrl } : {}), ...(links.dashboardUrl !== undefined ? { dashboardUrl: links.dashboardUrl } : {}), }); diff --git a/apps/dispatcher/src/routes/webhook.ts b/apps/dispatcher/src/routes/webhook.ts index 12ad78e..537ad4d 100644 --- a/apps/dispatcher/src/routes/webhook.ts +++ b/apps/dispatcher/src/routes/webhook.ts @@ -37,6 +37,7 @@ import { verify } from "../hmac"; import { toInstanceId } from "../instance-id"; import { triggersByEvent } from "../registry"; import { resolveReleaseApproval } from "../release-approval"; +import { decodeRerequest, handleRerequest } from "../rerequest"; import { signalWorkflow } from "../signal-workflow"; import type { Env } from "../env"; @@ -187,6 +188,26 @@ export const handleGithubWebhook = async (request: Request, env: Env): Promise; + /** + * The status `get(id).status()` reports per instance id; `undefined` makes + * the status read reject the way the platform does for an id it has no + * instance for. Omitted → every id reports `running`. + */ + instanceStatus?: (id: string) => string | undefined; } = {}, ): FakeWorkflow => { const calls: WorkflowCreateCall[] = []; const events: WorkflowSendEventCall[] = []; const reject = opts.rejectSendEventFor ?? new Set(); const alreadyExists = opts.throwAlreadyExistsFor ?? new Set(); + const statusOf = async (id: string) => { + const status = opts.instanceStatus === undefined ? "running" : opts.instanceStatus(id); + if (status === undefined) throw new Error(`instance.not_found: ${id}`); + return { status }; + }; const binding = { create: async (options?: { id?: string; params?: unknown }) => { const id = options?.id ?? ""; @@ -62,7 +73,7 @@ export const makeFakeWorkflow = ( }, get: (id: string) => ({ id, - status: async () => ({ status: "running" }), + status: () => statusOf(id), sendEvent: async (e: { type: string; payload: unknown }) => { if (reject.has(id)) { throw new Error(`unknown_instance: ${id}`); @@ -245,6 +256,8 @@ export const makeFakeEnv = (opts: { idempotencyKv?: KVNamespace; configKv?: KVNamespace; githubWebhookSecret?: string; + /** `GITHUB_APP_ID` — the App a check-run re-run must belong to. */ + githubAppId?: string; adminToken?: string; logLinkSecret?: string; metadata?: FakeD1; @@ -270,6 +283,7 @@ export const makeFakeEnv = (opts: { ...(opts.githubWebhookSecret !== undefined ? { GITHUB_WEBHOOK_SECRET: opts.githubWebhookSecret } : {}), + ...(opts.githubAppId !== undefined ? { GITHUB_APP_ID: opts.githubAppId } : {}), ...(opts.adminToken !== undefined ? { ADMIN_TOKEN: opts.adminToken } : {}), ...(opts.logLinkSecret !== undefined ? { LOG_LINK_SECRET: opts.logLinkSecret } : {}), ...(opts.publicOrigin !== undefined ? { PUBLIC_ORIGIN: opts.publicOrigin } : {}), diff --git a/apps/dispatcher/src/workflow.ts b/apps/dispatcher/src/workflow.ts index 66585e1..8e7dfef 100644 --- a/apps/dispatcher/src/workflow.ts +++ b/apps/dispatcher/src/workflow.ts @@ -85,7 +85,7 @@ import { queuedSummary } from "./admission-summary"; import { appendFailureSummary, failureSummaryMd, runSkippedReason } from "./failure-summary"; import { renderResultEmail } from "./notify"; import { workflowDashboardUrl } from "./dashboard-url"; -import { checkRunNameFor } from "./check-name"; +import { checkRunNameFor, checkRunTitleFor } from "./check-name"; import { buildLogsUrl, resolveLogLinkSecret, signLogToken } from "./log-token"; import { logLinksSuffix, startedSummary } from "./summary-links"; import { resolveMailboxLinkSecret, signMailboxToken } from "./mailbox-token"; @@ -167,6 +167,15 @@ const DispatchPayload = Schema.Struct({ * Distinct from `origin` above, which is the dispatcher's own public URL. */ source: Schema.optional(DispatchSource), + /** + * Which attempt of its family this execution is — absent (→ 1) on every + * dispatch path but a check-run re-run, which sets the next number + * (rerequest.ts). Persisted to `executions.attempt` and shown in the + * check-run title. + */ + attempt: Schema.optional(Schema.Int.pipe(Schema.greaterThanOrEqualTo(1))), + /** The id of attempt 1 of the family a re-run retries — `executions.retry_of`. */ + retryOf: Schema.optional(Schema.String), }); type DispatchPayload = Schema.Schema.Type; @@ -351,6 +360,10 @@ export class RunWorkflow extends WorkflowEntrypoint { // separately-requirable checks on the same commit (check-name.ts). Every // run without that input names identically to before. const checkRunName = checkRunNameFor(payload.run, payload.inputs); + // The name stays fixed across re-runs (branch protection keys on it); the + // output title carries the attempt so a re-run reads as one. + const attempt = payload.attempt ?? 1; + const checkRunTitle = checkRunTitleFor(checkRunName, attempt); // The Cloudflare Workflows instance page for this execution — the "Details" // link on the GitHub check-run + a markdown link in its summary. `undefined` // when CLOUDFLARE_ACCOUNT_ID is unset (BYOC default): the check-run renders @@ -601,6 +614,8 @@ export class RunWorkflow extends WorkflowEntrypoint { ...(payload.parentExecutionId !== undefined ? { parentExecutionId: payload.parentExecutionId } : {}), + attempt, + ...(payload.retryOf !== undefined ? { retryOf: payload.retryOf } : {}), }); // Open the check-run (`in_progress`). With no App config this resolves @@ -611,7 +626,7 @@ export class RunWorkflow extends WorkflowEntrypoint { name: checkRunName, ...(checkDetailsUrl !== undefined ? { detailsUrl: checkDetailsUrl } : {}), output: { - title: checkRunName, + title: checkRunTitle, // Both log links from the very first render (summary-links.ts) — a // reviewer watching an in-progress check reaches the viewer without // waiting for the verdict update. @@ -721,7 +736,7 @@ export class RunWorkflow extends WorkflowEntrypoint { checkRunId, ...(checkDetailsUrl !== undefined ? { detailsUrl: checkDetailsUrl } : {}), output: { - title: checkRunName, + title: checkRunTitle, summary: queuedSummary( decision.position, decision.poolBusy, @@ -1029,7 +1044,7 @@ export class RunWorkflow extends WorkflowEntrypoint { conclusion: status === "skipped" ? "neutral" : status, ...(checkDetailsUrl !== undefined ? { detailsUrl: checkDetailsUrl } : {}), output: { - title: checkRunName, + title: checkRunTitle, summary: skipReason !== undefined ? `⊘ ${payload.run} — skipped: ${skipReason}.${logsSuffix}` diff --git a/infra/migrations/0007_execution_attempt.sql b/infra/migrations/0007_execution_attempt.sql new file mode 100644 index 0000000..9bdcd61 --- /dev/null +++ b/infra/migrations/0007_execution_attempt.sql @@ -0,0 +1,21 @@ +-- Migration number: 0007 execution attempt lineage +-- +-- A check-run re-run (the GitHub "Re-run" button → `check_run.rerequested`) +-- dispatches a FRESH execution of the same run, repo, SHA and inputs. The +-- execution id cannot be reused: Cloudflare Workflows refuses `create({ id })` +-- for any id it has seen, terminated or not. So a re-run is a new row, and +-- these two columns tie it back to the execution it retries: +-- +-- * `attempt` — 1 for a dispatched execution, N for the (N-1)th re-run. +-- Pre-0007 rows are first attempts, so the default is exact. +-- * `retry_of` — the id of attempt 1 (the ROOT of the family), NULL on +-- attempt 1 itself. Every attempt points at the root rather +-- than at its predecessor, so one indexed lookup returns the +-- whole family. +-- +-- The `check_run_id` index serves the re-run lookup: the webhook names the +-- check-run, and the execution that posted it is found by that id. +ALTER TABLE executions ADD COLUMN attempt INTEGER NOT NULL DEFAULT 1; +ALTER TABLE executions ADD COLUMN retry_of TEXT; +CREATE INDEX IF NOT EXISTS executions_retry_of ON executions(retry_of); +CREATE INDEX IF NOT EXISTS executions_check_run ON executions(check_run_id); diff --git a/packages/core/src/fakes/executions-fake.ts b/packages/core/src/fakes/executions-fake.ts index 51b9e25..cb9124e 100644 --- a/packages/core/src/fakes/executions-fake.ts +++ b/packages/core/src/fakes/executions-fake.ts @@ -31,13 +31,15 @@ export const makeExecutionsFake = (): { const state: ExecutionsFakeState = { executions: [], steps: [] }; const service: ExecutionsService = { - startExecution: ({ id, run, startedAt, parentExecutionId }) => + startExecution: ({ id, run, startedAt, parentExecutionId, attempt, retryOf }) => Effect.sync(() => { state.executions.push({ id, run, startedAt, ...(parentExecutionId !== undefined ? { parentExecutionId } : {}), + ...(attempt !== undefined ? { attempt } : {}), + ...(retryOf !== undefined ? { retryOf } : {}), }); }), diff --git a/packages/core/src/services/executions.ts b/packages/core/src/services/executions.ts index 3e23a90..ee7507e 100644 --- a/packages/core/src/services/executions.ts +++ b/packages/core/src/services/executions.ts @@ -54,6 +54,10 @@ export type ExecutionRecord = { * column a fan-out parent reads back to join on its children's outcomes. */ readonly parentExecutionId?: string; + /** 1 for a dispatched execution; N for the (N-1)th re-run of it. Absent → 1. */ + readonly attempt?: number; + /** The id of attempt 1 of this execution's family, when this row is a re-run. */ + readonly retryOf?: string; }; /** @@ -73,6 +77,17 @@ export interface ExecutionsService { * fan-out parent can enumerate its children. Omitted for top-level rows. */ parentExecutionId?: string; + /** + * Which attempt of its family this execution is — persisted to + * `executions.attempt`. Omitted → 1. A check-run re-run dispatches a fresh + * execution with the next attempt number (routes/rerequest.ts). + */ + attempt?: number; + /** + * The id of attempt 1 of the family this execution re-runs — persisted to + * `executions.retry_of`. Omitted on attempt 1. + */ + retryOf?: string; }) => Effect.Effect; /** Mark an `executions` row terminal. */ diff --git a/packages/runtime-cf/src/executions-d1.ts b/packages/runtime-cf/src/executions-d1.ts index 92bae99..afc21ef 100644 --- a/packages/runtime-cf/src/executions-d1.ts +++ b/packages/runtime-cf/src/executions-d1.ts @@ -86,18 +86,21 @@ export const makeD1ExecutionsLive = ( }).pipe(Effect.orDie); const service: ExecutionsService = { - startExecution: ({ id, run: runName, startedAt, parentExecutionId }) => + startExecution: ({ id, run: runName, startedAt, parentExecutionId, attempt, retryOf }) => run("startExecution", () => db // `OR IGNORE`: the PK `id` is deterministic (the executionId), so a // replayed insert on Workflow resume is a no-op, not a PK violation. // `parent_execution_id` is NULL for a top-level execution and the // spawning parent's id for a `spawnChildRun` child — the lineage a - // fan-out parent reads back to join on its children. + // fan-out parent reads back to join on its children. `attempt` / + // `retry_of` tie a check-run re-run to the family it retries + // (infra/migrations/0007) — 1 / NULL for every dispatched execution. .prepare( `INSERT OR IGNORE INTO executions - (id, run, repo, ref, sha, status, started_at, input_json, parent_execution_id) - VALUES (?, ?, ?, ?, ?, 'running', ?, ?, ?)`, + (id, run, repo, ref, sha, status, started_at, input_json, parent_execution_id, + attempt, retry_of) + VALUES (?, ?, ?, ?, ?, 'running', ?, ?, ?, ?, ?)`, ) .bind( id, @@ -108,6 +111,8 @@ export const makeD1ExecutionsLive = ( startedAt, JSON.stringify(ctx.input), parentExecutionId ?? null, + attempt ?? 1, + retryOf ?? null, ) .run(), ), diff --git a/packages/runtime-cf/src/executions-d1.workers.test.ts b/packages/runtime-cf/src/executions-d1.workers.test.ts index ad21a99..b0b1028 100644 --- a/packages/runtime-cf/src/executions-d1.workers.test.ts +++ b/packages/runtime-cf/src/executions-d1.workers.test.ts @@ -108,6 +108,33 @@ describe("D1ExecutionsLive", () => { expect(child?.parent_execution_id).toBe(EXECUTION_ID); }); + it("records attempt 1 / NULL retry_of by default, and the lineage a re-run passes", async () => { + const layer = makeD1ExecutionsLive(bindings.db, CTX); + const rerunId = "01TEST00000000000000000003"; + + await Effect.runPromise( + Effect.gen(function* () { + const executions = yield* Executions; + yield* executions.startExecution({ id: EXECUTION_ID, run: "check", startedAt: 0 }); + yield* executions.startExecution({ + id: rerunId, + run: "check", + startedAt: 1, + attempt: 2, + retryOf: EXECUTION_ID, + }); + }).pipe(Effect.provide(layer)), + ); + + const lineage = async (id: string) => + bindings.db + .prepare(`SELECT attempt, retry_of FROM executions WHERE id = ?`) + .bind(id) + .first<{ attempt: number; retry_of: string | null }>(); + expect(await lineage(EXECUTION_ID)).toEqual({ attempt: 1, retry_of: null }); + expect(await lineage(rerunId)).toEqual({ attempt: 2, retry_of: EXECUTION_ID }); + }); + it("writes one steps row per step, each spanning start → finish", async () => { const layer = makeD1ExecutionsLive(bindings.db, CTX); const stepNames = ["checkout", "exec", "upload-log"]; diff --git a/runs/README.md b/runs/README.md index f66ba0c..bfc33d5 100644 --- a/runs/README.md +++ b/runs/README.md @@ -40,6 +40,35 @@ artifact behind. For `offload-test`, per-stage exec steps into `step-