| 1 | import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep } from "cloudflare:workers"; |
| 2 | import type { Env } from "../env.js"; |
| 3 | import { graphFromLevels, runGraph, type JobNode } from "../repo/job-graph.js"; |
| 4 | import { MAX_PLACEMENT_TRIES, MAX_RUNNER_RESTARTS, QUICK_RETRIES } from "../repo/actions.js"; |
| 5 | import { plainSteps } from "../lib/rpc.js"; |
| 6 | |
| 7 | export interface RunParams { |
| 8 | repoId: string; |
| 9 | runId: string; |
| 10 | } |
| 11 | |
| 12 | /** |
| 13 | * Durable orchestration of one workflow run: walks the job graph, starting each |
| 14 | * job's containers as soon as the jobs it needs have finished, and waits for the |
| 15 | * runner to report completion. Retries, sleeps and waits survive restarts of the |
| 16 | * platform. Each job costs a fixed number of steps (start, then a status check |
| 17 | * and one wait per matrix leg), however long it runs, plus a few per wake while |
| 18 | * its environment's reviewers or wait timer hold it. |
| 19 | */ |
| 20 | export class RunWorkflow extends WorkflowEntrypoint<Env, RunParams> { |
| 21 | override async run(event: WorkflowEvent<RunParams>, rpcStep: WorkflowStep): Promise<unknown> { |
| 22 | const step = plainSteps(rpcStep); |
| 23 | const { repoId, runId } = event.payload; |
| 24 | const repo = () => this.env.REPO.get(this.env.REPO.idFromName(repoId)); |
| 25 | |
| 26 | let plan = await step.do("plan", () => repo().runPlan(runId)); |
| 27 | // Concurrency groups without cancel-in-progress: wait for the running run to finish. |
| 28 | for (let i = 0; plan.concurrencyBlocked && i < 360; i++) { |
| 29 | await step.sleep(`concurrency-wait-${i}`, "1 minute"); |
| 30 | plan = await step.do(`plan-${i}`, () => repo().runPlan(runId)); |
| 31 | } |
| 32 | if (plan.error) { |
| 33 | await step.do("record-error", () => repo().plannedRunError(runId, plan.error!)); |
| 34 | return step.do("finish", () => repo().finishRun(runId)); |
| 35 | } |
| 36 | |
| 37 | const waitFor = async (jobId: string) => { |
| 38 | let status = await step.do(`job ${jobId} info`, () => repo().jobStatus(jobId)); |
| 39 | // Held by its environment's protection rules: wait for a reviewer's decision or the wait timer, |
| 40 | // for up to 30 days (repo/actions.ts jobGate fails it after that). Each wake re-checks the rules. |
| 41 | for (let i = 0; status === "approval" && i < 100; i++) { |
| 42 | const gate = await step.do(`gate ${jobId} ${i}`, () => repo().jobGate(jobId)); |
| 43 | status = gate.status; |
| 44 | if (status !== "approval") break; |
| 45 | const until = Date.parse(gate.until ?? "") || Date.now() + 60_000; |
| 46 | if (gate.wait === "timer") await step.sleepUntil(`timer ${jobId} ${i}`, until); |
| 47 | else { |
| 48 | try { |
| 49 | await step.waitForEvent(`approval ${jobId} ${i}`, { type: `gate-${jobId}`, timeout: Math.max(1000, until - Date.now()) }); |
| 50 | } catch { |
| 51 | // The deadline passed: the next gate check fails the job. |
| 52 | } |
| 53 | } |
| 54 | } |
| 55 | if (status === "completed" || status === "approval") return; |
| 56 | // A job whose container got no machine, or whose runner a rollout replaced before its steps ran |
| 57 | // (repo/actions.ts runnerReplaced), comes back with `retryInSeconds`: pause, start it again |
| 58 | // and keep waiting. The first round keeps its step names, so runs already in flight replay. |
| 59 | for (let i = 0; ; i++) { |
| 60 | const n = i ? ` ${i}` : ""; |
| 61 | if (i && (await step.do(`job ${jobId} info${n}`, () => repo().jobStatus(jobId))) === "completed") return; |
| 62 | let event: { payload?: { retryInSeconds?: number } }; |
| 63 | try { |
| 64 | event = await step.waitForEvent(`wait ${jobId}${n}`, { type: `job-${jobId}`, timeout: "6 hours" }); |
| 65 | } catch { |
| 66 | await step.do(`timeout ${jobId}${n}`, () => repo().jobTimedOut(jobId)); |
| 67 | return; |
| 68 | } |
| 69 | const pause = event.payload?.retryInSeconds; |
| 70 | if (!pause || i >= MAX_PLACEMENT_TRIES + MAX_RUNNER_RESTARTS + QUICK_RETRIES) return; |
| 71 | await step.sleep(`pause ${jobId} ${i + 1}`, `${pause} seconds`); |
| 72 | await step.do(`relaunch ${jobId} ${i + 1}`, () => repo().relaunchJob(jobId)); |
| 73 | } |
| 74 | }; |
| 75 | // Runs planned before plans carried a graph replay with their recorded levels. |
| 76 | const legacy = plan as { jobs?: JobNode[]; levels?: string[][] }; |
| 77 | await runGraph(legacy.jobs ?? graphFromLevels(legacy.levels ?? []), async (key) => { |
| 78 | const started = await step.do(`start ${key}`, { retries: { limit: 3, delay: "10 seconds", backoff: "exponential" } }, () => repo().startJobKey(runId, key)); |
| 79 | await Promise.all(started.jobIds.map(waitFor)); |
| 80 | }); |
| 81 | return step.do("finish", () => repo().finishRun(runId)); |
| 82 | } |
| 83 | } |