| 1 | import { describe, expect, it } from "vitest"; |
| 2 | import { DatabaseSync } from "node:sqlite"; |
| 3 | import { parseWorkflow } from "@diffinery/workflow-yaml"; |
| 4 | import { Actions, STUCK_RUN_FINISH_MS, STUCK_RUN_MS } from "../../src/repo/actions.js"; |
| 5 | import { REPO_SCHEMA, migrateRepoSchema } from "../../src/repo/schema.js"; |
| 6 | import { testD1 } from "./stubs/d1.js"; |
| 7 | import type { RepoDO } from "../../src/repo/RepoDO.js"; |
| 8 | import type { Env } from "../../src/env.js"; |
| 9 | import type { Caller } from "../../src/repo/types.js"; |
| 10 | |
| 11 | /** The Durable Object's SqlStorage over node:sqlite: exec(query, ...bindings) with toArray() and one(). */ |
| 12 | function sqlStorage() { |
| 13 | const db = new DatabaseSync(":memory:"); |
| 14 | return { |
| 15 | exec(query: string, ...args: unknown[]) { |
| 16 | const params = args.map((a) => (a === undefined ? null : a)) as Array<string | number | null>; |
| 17 | let rows: unknown[] = []; |
| 18 | if (!params.length && query.trim().replace(/;\s*$/, "").includes(";")) db.exec(query); |
| 19 | else rows = db.prepare(query).all(...params); |
| 20 | return { |
| 21 | toArray: () => rows as never[], |
| 22 | one: () => { |
| 23 | if (rows.length !== 1) throw new Error(`expected one row, got ${rows.length}`); |
| 24 | return rows[0] as never; |
| 25 | }, |
| 26 | }; |
| 27 | }, |
| 28 | }; |
| 29 | } |
| 30 | |
| 31 | // Like main's End-to-end: one run at a time per branch, without cancelling the running one. |
| 32 | const WORKFLOW = `name: e2e |
| 33 | on: push |
| 34 | concurrency: |
| 35 | group: e2e-\${{ github.ref }} |
| 36 | jobs: |
| 37 | test: |
| 38 | runs-on: ubuntu-latest |
| 39 | steps: [{ run: echo test }] |
| 40 | deploy: |
| 41 | needs: test |
| 42 | runs-on: ubuntu-latest |
| 43 | steps: [{ run: echo deploy }] |
| 44 | `; |
| 45 | |
| 46 | const ADA: Caller = { id: "u1", login: "ada", kind: "user", role: "write" }; |
| 47 | |
| 48 | /** |
| 49 | * Actions over an in-memory repository, with a stand-in RunWorkflow whose instances report `states[runId]` |
| 50 | * ("waiting" by default) and record the events they get; `lose` drops events, as during a deploy. |
| 51 | */ |
| 52 | function fixture() { |
| 53 | const sql = sqlStorage(); |
| 54 | sql.exec(REPO_SCHEMA); |
| 55 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 56 | const DB = testD1(); |
| 57 | DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString()); |
| 58 | const events: { runId: string; type: string }[] = []; |
| 59 | const completed: string[] = []; |
| 60 | const alarms: number[] = []; |
| 61 | const states: Record<string, string> = {}; |
| 62 | const net = { lose: false }; |
| 63 | const repo = { |
| 64 | sql, |
| 65 | meta: { repoId: "r", owner: "acme", ownerId: "acme", ownerKind: "org", name: "app", defaultBranch: "main" }, |
| 66 | headRef: "refs/heads/main", |
| 67 | getRef: () => null, |
| 68 | upsertCheck: () => {}, |
| 69 | broadcast: () => {}, |
| 70 | emit: (e: { type: string; action?: string; runId?: string }) => void (e.type === "workflow_run" && e.action === "completed" && completed.push(e.runId!)), |
| 71 | deploys: { jobCompleted: () => {}, protection: () => null }, |
| 72 | runtime: { sweepJob: async () => {} }, |
| 73 | policy: { agentOrigin: () => false }, |
| 74 | alarmBy: (at: number) => void alarms.push(at), |
| 75 | } as unknown as RepoDO; |
| 76 | const env = { |
| 77 | DB, |
| 78 | PUBLIC_URL: "https://d.test", |
| 79 | BUCKET: { put: async () => {} }, |
| 80 | RUNNER: { idFromName: (n: string) => n, get: () => ({ startJob: async () => {}, stopJob: async () => {} }) }, |
| 81 | RUN_WORKFLOW: { |
| 82 | create: async () => {}, |
| 83 | get: async (runId: string) => ({ |
| 84 | sendEvent: async (e: { type: string }) => { |
| 85 | if (net.lose) throw new Error("network connection lost"); |
| 86 | events.push({ runId, type: e.type }); |
| 87 | }, |
| 88 | status: async () => ({ status: states[runId] ?? "waiting" }), |
| 89 | }), |
| 90 | }, |
| 91 | } as unknown as Env; |
| 92 | const actions = new Actions(repo, env); |
| 93 | const { workflow } = parseWorkflow(WORKFLOW, ".github/workflows/e2e.yml"); |
| 94 | Object.defineProperty(actions, "loadWorkflows", { value: async () => [{ path: ".github/workflows/e2e.yml", workflow, errors: [] }] }); |
| 95 | const internal = actions as unknown as { completeJob(id: string, c: unknown): Promise<void> }; |
| 96 | const jobs = (runId: string) => sql.exec("SELECT id, job_key, name, status, conclusion, failure FROM jobs WHERE run_id = ? ORDER BY rowid", runId).toArray() as { id: string; job_key: string; name: string; status: string; conclusion: string; failure: string | null }[]; |
| 97 | const push = async () => (await actions.createRun(workflow!, { event: "push", ref: "refs/heads/main", sha: "abc", payload: {}, caller: ADA, title: "push" }))!.id; |
| 98 | /** Plans the run and runs its `test` job to `conclusion`, as the run's workflow and runner would. */ |
| 99 | const runTest = async (runId: string, conclusion = "success") => { |
| 100 | expect((await actions.plan(runId)).concurrencyBlocked).toBe(false); |
| 101 | await actions.startJobKey(runId, "test"); |
| 102 | await internal.completeJob(jobs(runId)[0]!.id, { conclusion, outputs: {} }); |
| 103 | }; |
| 104 | /** Moves the run's finished jobs `ms` into the past. */ |
| 105 | const age = (runId: string, ms: number) => { |
| 106 | const t = new Date(Date.now() - ms).toISOString(); |
| 107 | sql.exec("UPDATE jobs SET completed_at = ? WHERE run_id = ?", t, runId); |
| 108 | sql.exec("UPDATE runs SET created_at = ? WHERE id = ?", t, runId); |
| 109 | }; |
| 110 | return { sql, actions, events, completed, alarms, states, net, jobs, push, runTest, age }; |
| 111 | } |
| 112 | |
| 113 | describe("runs stuck after their jobs finished", () => { |
| 114 | it("a lost job event is sent again, so the run's workflow carries on and the next run in the group starts", async () => { |
| 115 | const f = fixture(); |
| 116 | const a = await f.push(); |
| 117 | f.net.lose = true; |
| 118 | const before = Date.now(); |
| 119 | await f.runTest(a); |
| 120 | f.net.lose = false; |
| 121 | // The run's last job arms a check for later. |
| 122 | expect(f.alarms.at(-1)).toBeGreaterThanOrEqual(before + STUCK_RUN_MS); |
| 123 | // The next push waits behind it: the run is too fresh to call stuck. |
| 124 | const b = await f.push(); |
| 125 | expect((await f.actions.plan(b)).concurrencyBlocked).toBe(true); |
| 126 | expect(f.actions.nextStuckRunCheck()).toBeGreaterThanOrEqual(before + STUCK_RUN_MS); |
| 127 | |
| 128 | // Past STUCK_RUN_MS its workflow, still waiting, gets the job's event again, once. |
| 129 | const later = Date.now() + STUCK_RUN_MS + 1000; |
| 130 | await f.actions.repairStuckRuns(later); |
| 131 | await f.actions.repairStuckRuns(later + 1000); |
| 132 | expect(f.events).toEqual([{ runId: a, type: `job-${f.jobs(a)[0]!.id}` }]); |
| 133 | expect(f.actions.getRun(a).status).toBe("in_progress"); |
| 134 | expect(f.actions.nextStuckRunCheck(later)).toBeGreaterThan(later + STUCK_RUN_FINISH_MS - STUCK_RUN_MS - 2000); |
| 135 | |
| 136 | // The workflow carries on with the rest of the graph and finishes the run, and b can start. |
| 137 | await f.actions.startJobKey(a, "deploy"); |
| 138 | await (f.actions as unknown as { completeJob(id: string, c: unknown): Promise<void> }).completeJob(f.jobs(a)[1]!.id, { conclusion: "success", outputs: {} }); |
| 139 | expect((await f.actions.finishRun(a)).conclusion).toBe("success"); |
| 140 | expect((await f.actions.plan(b)).concurrencyBlocked).toBe(false); |
| 141 | expect(f.completed).toEqual([a]); |
| 142 | }); |
| 143 | |
| 144 | it("a run whose workflow failed is finished, failing and naming the jobs that never started; the late workflow changes nothing", async () => { |
| 145 | const f = fixture(); |
| 146 | const a = await f.push(); |
| 147 | await f.runTest(a); |
| 148 | f.states[a] = "errored"; |
| 149 | await f.actions.repairStuckRuns(Date.now() + STUCK_RUN_MS + 1000); |
| 150 | const run = f.actions.getRun(a); |
| 151 | expect(run).toMatchObject({ status: "completed", conclusion: "failure" }); |
| 152 | const setup = f.jobs(a).find((j) => j.name === "Set up workflow")!; |
| 153 | expect(JSON.parse(setup.failure!).message).toBe("The run's workflow stopped after its jobs finished, so a job never started (deploy). Re-run the workflow."); |
| 154 | expect(f.actions.nextStuckRunCheck()).toBeNull(); |
| 155 | // Should the workflow come back after all, it starts nothing and doesn't finish the run a second time. |
| 156 | expect(await f.actions.startJobKey(a, "deploy")).toEqual({ jobIds: [] }); |
| 157 | expect((await f.actions.finishRun(a)).conclusion).toBe("failure"); |
| 158 | expect(f.completed).toEqual([a]); |
| 159 | }); |
| 160 | |
| 161 | it("a run still stuck STUCK_RUN_FINISH_MS after its jobs finished is finished, though its workflow looks alive", async () => { |
| 162 | const f = fixture(); |
| 163 | const a = await f.push(); |
| 164 | await f.runTest(a, "failure"); |
| 165 | const t = Date.now(); |
| 166 | await f.actions.repairStuckRuns(t + STUCK_RUN_MS + 1000); |
| 167 | expect(f.actions.getRun(a).status).toBe("in_progress"); |
| 168 | await f.actions.repairStuckRuns(t + STUCK_RUN_FINISH_MS + 1000); |
| 169 | // Its failed job fails the run. |
| 170 | expect(f.actions.getRun(a)).toMatchObject({ status: "completed", conclusion: "failure" }); |
| 171 | }); |
| 172 | |
| 173 | it("a waiting run in the group repairs the stuck one when it plans: a cancelled run whose jobs ended long ago frees the group", async () => { |
| 174 | const f = fixture(); |
| 175 | const a = await f.push(); |
| 176 | await f.runTest(a); |
| 177 | // As on main at 14:46: the run was cancelled, its jobs ended, and its workflow never finished it. |
| 178 | f.sql.exec("UPDATE runs SET conclusion = 'cancelled' WHERE id = ?", a); |
| 179 | f.age(a, 2 * 60 * 60_000); |
| 180 | f.states[a] = "waiting"; |
| 181 | const b = await f.push(); |
| 182 | // Long past STUCK_RUN_FINISH_MS: finished at once, cancelled, with nothing added. |
| 183 | expect((await f.actions.plan(b)).concurrencyBlocked).toBe(false); |
| 184 | expect(f.actions.getRun(a)).toMatchObject({ status: "completed", conclusion: "cancelled" }); |
| 185 | expect(f.jobs(a).map((j) => j.name)).toEqual(["test"]); |
| 186 | }); |
| 187 | |
| 188 | it("leaves alone runs with a job still going, and runs that finished", async () => { |
| 189 | const f = fixture(); |
| 190 | const a = await f.push(); |
| 191 | expect((await f.actions.plan(a)).concurrencyBlocked).toBe(false); |
| 192 | await f.actions.startJobKey(a, "test"); |
| 193 | f.age(a, 2 * 60 * 60_000); |
| 194 | await f.actions.repairStuckRuns(); |
| 195 | expect(f.actions.getRun(a).status).toBe("in_progress"); |
| 196 | expect(f.actions.nextStuckRunCheck()).toBeNull(); |
| 197 | expect(f.events).toEqual([]); |
| 198 | |
| 199 | // A re-run that just started, whose kept jobs finished in the earlier attempt hours ago, isn't stuck. |
| 200 | const b = await f.push(); |
| 201 | f.sql.exec("UPDATE jobs SET status = 'completed', conclusion = 'success', completed_at = ? WHERE run_id = ?", new Date().toISOString(), a); |
| 202 | await f.actions.finishRun(a); |
| 203 | expect((await f.actions.plan(b)).concurrencyBlocked).toBe(false); |
| 204 | f.sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status, conclusion, completed_at, carried_from) VALUES ('kept', ?, 'test', 'test', 'completed', 'success', ?, 'old')", b, new Date(Date.now() - 3 * 60 * 60_000).toISOString()); |
| 205 | await f.actions.repairStuckRuns(Date.now() + 1000); |
| 206 | expect(f.actions.getRun(b).status).toBe("in_progress"); |
| 207 | }); |
| 208 | }); |
| 209 | |
| 210 | describe("a concurrency group without cancel-in-progress", () => { |
| 211 | it("keeps one run waiting, like GitHub: a newer run replaces older waiting ones, which finish cancelled at once", async () => { |
| 212 | const f = fixture(); |
| 213 | const a = await f.push(); |
| 214 | expect((await f.actions.plan(a)).concurrencyBlocked).toBe(false); |
| 215 | await f.actions.startJobKey(a, "test"); |
| 216 | const b = await f.push(); |
| 217 | expect((await f.actions.plan(b)).concurrencyBlocked).toBe(true); |
| 218 | const c = await f.push(); |
| 219 | const d = await f.push(); |
| 220 | // b and c are replaced as d comes in; the running a is left alone. |
| 221 | expect(f.actions.getRun(b).conclusion).toBe("cancelled"); |
| 222 | expect(f.actions.getRun(c).conclusion).toBe("cancelled"); |
| 223 | expect(f.actions.getRun(a)).toMatchObject({ status: "in_progress", conclusion: null }); |
| 224 | // A replaced run doesn't wait its turn: its next plan has nothing to run, and its workflow finishes it. |
| 225 | expect(await f.actions.plan(c)).toEqual({ jobs: [], concurrencyBlocked: false }); |
| 226 | expect((await f.actions.finishRun(c)).conclusion).toBe("cancelled"); |
| 227 | expect(f.jobs(c)).toEqual([]); |
| 228 | expect((await f.actions.plan(d)).concurrencyBlocked).toBe(true); |
| 229 | // When a ends, d (the last push) runs next. |
| 230 | await (f.actions as unknown as { completeJob(id: string, c: unknown): Promise<void> }).completeJob(f.jobs(a)[0]!.id, { conclusion: "success", outputs: {} }); |
| 231 | await f.actions.startJobKey(a, "deploy"); |
| 232 | await (f.actions as unknown as { completeJob(id: string, c: unknown): Promise<void> }).completeJob(f.jobs(a)[1]!.id, { conclusion: "success", outputs: {} }); |
| 233 | await f.actions.finishRun(a); |
| 234 | expect((await f.actions.plan(d)).concurrencyBlocked).toBe(false); |
| 235 | expect(f.actions.getRun(d).status).toBe("in_progress"); |
| 236 | }); |
| 237 | |
| 238 | it("clears runs that waited before newer runs replaced older ones: the oldest doesn't take the group ahead of the newest", async () => { |
| 239 | const f = fixture(); |
| 240 | const runs = [await f.push(), await f.push(), await f.push()]; |
| 241 | // As in a deploy group from before this change: every push waiting, none cancelled. |
| 242 | f.sql.exec("UPDATE runs SET status = 'queued', conclusion = NULL"); |
| 243 | // The group is free, and the oldest plans first: it is replaced by the newest, and runs nothing. |
| 244 | expect(await f.actions.plan(runs[0]!)).toEqual({ jobs: [], concurrencyBlocked: false }); |
| 245 | expect(f.actions.getRun(runs[1]!).conclusion).toBe("cancelled"); |
| 246 | expect(f.actions.getRun(runs[2]!)).toMatchObject({ status: "queued", conclusion: null }); |
| 247 | expect((await f.actions.plan(runs[2]!)).concurrencyBlocked).toBe(false); |
| 248 | expect(f.actions.getRun(runs[2]!).status).toBe("in_progress"); |
| 249 | }); |
| 250 | }); |