| 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 } from "../repo/actions.js"; |
| 5 | |
| 6 | export interface RunParams { |
| 7 | repoId: string; |
| 8 | runId: string; |
| 9 | } |
| 10 | |
| 11 | /** |
| 12 | * Durable orchestration of one workflow run: walks the job graph, starting each |
| 13 | * job's containers as soon as the jobs it needs have finished, and waits for the |
| 14 | * runner to report completion. Retries, sleeps and waits survive restarts of the |
| 15 | * platform. Each job costs a fixed number of steps (start, then a status check |
| 16 | * and one wait per matrix leg), however long it runs. |
| 17 | */ |
| 18 | export class RunWorkflow extends WorkflowEntrypoint<Env, RunParams> { |
| 19 | override async run(event: WorkflowEvent<RunParams>, step: WorkflowStep): Promise<unknown> { |
| 20 | const { repoId, runId } = event.payload; |
| 21 | const repo = () => this.env.REPO.get(this.env.REPO.idFromName(repoId)); |
| 22 | |
| 23 | let plan = await step.do("plan", () => repo().runPlan(runId)); |
| 24 | // Concurrency groups without cancel-in-progress: wait for the running run to finish. |
| 25 | for (let i = 0; plan.concurrencyBlocked && i < 360; i++) { |
| 26 | await step.sleep(`concurrency-wait-${i}`, "1 minute"); |
| 27 | plan = await step.do(`plan-${i}`, () => repo().runPlan(runId)); |
| 28 | } |
| 29 | if (plan.error) { |
| 30 | await step.do("record-error", () => repo().plannedRunError(runId, plan.error!)); |
| 31 | return step.do("finish", () => repo().finishRun(runId)); |
| 32 | } |
| 33 | |
| 34 | const waitFor = async (jobId: string) => { |
| 35 | // A job whose container got no machine comes back with `retryInSeconds`: pause, start it again |
| 36 | // and keep waiting. The first round keeps its step names, so runs already in flight replay. |
| 37 | for (let i = 0; ; i++) { |
| 38 | const n = i ? ` ${i}` : ""; |
| 39 | const status = await step.do(`job ${jobId} info${n}`, () => repo().jobStatus(jobId)); |
| 40 | if (status === "completed") return; |
| 41 | let event: { payload?: { retryInSeconds?: number } }; |
| 42 | try { |
| 43 | event = await step.waitForEvent(`wait ${jobId}${n}`, { type: `job-${jobId}`, timeout: "6 hours" }); |
| 44 | } catch { |
| 45 | await step.do(`timeout ${jobId}${n}`, () => repo().jobTimedOut(jobId)); |
| 46 | return; |
| 47 | } |
| 48 | const pause = event.payload?.retryInSeconds; |
| 49 | if (!pause || i >= MAX_PLACEMENT_TRIES) return; |
| 50 | await step.sleep(`pause ${jobId} ${i + 1}`, `${pause} seconds`); |
| 51 | await step.do(`relaunch ${jobId} ${i + 1}`, () => repo().relaunchJob(jobId)); |
| 52 | } |
| 53 | }; |
| 54 | // Runs planned before plans carried a graph replay with their recorded levels. |
| 55 | const legacy = plan as { jobs?: JobNode[]; levels?: string[][] }; |
| 56 | await runGraph(legacy.jobs ?? graphFromLevels(legacy.levels ?? []), async (key) => { |
| 57 | const started = await step.do(`start ${key}`, { retries: { limit: 3, delay: "10 seconds", backoff: "exponential" } }, () => repo().startJobKey(runId, key)); |
| 58 | await Promise.all(started.jobIds.map(waitFor)); |
| 59 | }); |
| 60 | return step.do("finish", () => repo().finishRun(runId)); |
| 61 | } |
| 62 | } |