Diffinery

codician/diffinerypublic

Open-source, agent-first GitHub and GitHub Actions alternative that runs on your own Cloudflare account: git hosting, pull requests, code review and drop-in GitHub Actions YAML on Workers, Durable Objects, R2 and Containers.

A run whose jobs all finished always finishes and frees its concurrency group (#236)

* A run whose jobs all finished always finishes and frees its concurrency group
test-bot authored 1 parent 12fa3c8commit 92a47952f2e99bfac718d840316495720dc8cd57Browse files
8 files changed+326−6
Files (8)
modifiedapps/diffinery/docs/actions/overview.md+3−0View file
@@ -113,6 +113,8 @@
113113 5. **Each hosted job gets its own container.** A fresh Cloudflare Container (Ubuntu 24.04, linux/amd64) starts with a job token that is valid only for that job. The runner agent inside fetches the job definition (for a job with `environment:`, this is when the environment's branch policy is checked and its secrets and variables are added), runs the steps in order, streams the log back as it goes, and reports each step's result and the job's conclusion. The container stops when the job ends. Jobs for self-hosted runners wait until a `diffinery runner` with matching labels claims them instead.
114114 6. **The run finishes** when every job is done. Its conclusion is `failure` if any job failed or timed out (except jobs with `continue-on-error: true`), otherwise `cancelled` if the run was cancelled or any job was cancelled, otherwise `skipped` if every job was skipped, otherwise `success`. Skipped jobs don't make a run fail. As on GitHub, a `workflow_run` job with `if: github.event.workflow_run.conclusion == 'success'` is skipped after a run whose jobs were all skipped.
115115
116+ If the orchestration loses track of a run whose jobs have all finished (for example, a job's completion signal lost while the platform updates), Diffinery notices 5 minutes after the last job finished and signals the orchestration again. A run still unfinished 30 minutes after its last job, or whose orchestration has stopped, is finished by Diffinery itself, which frees its concurrency group. Jobs the run never started make it fail, and a `Set up workflow` job names them: `The run's workflow stopped after its jobs finished, so a job never started (deploy). Re-run the workflow.`
117+
116118 ### Timeouts
117119
118120 | What | Default | Maximum |
@@ -441,6 +443,7 @@
441443 | Symptom | Cause and fix |
442444 | --- | --- |
443445 | Nothing runs after a push | Check that the file is directly in `.github/workflows/` or `.diffinery/workflows/` and ends in `.yml` or `.yaml`, and that the `on:` filters match the branch, tag or paths. Run `diffinery workflow validate`. |
446+| A run stays in progress though all its jobs finished, and newer runs in its concurrency group wait with no jobs | The run's orchestration lost track of it. Diffinery repairs it by itself: within 5 minutes of its last job the orchestration is signalled again, and by 30 minutes the run is finished, freeing the group. Nothing to do; re-run the workflow if it names jobs that never started. |
444447 | A run titled `Invalid workflow file: ...` appears | A workflow file in the pushed commit has errors. Its one job, **Invalid workflow file**, lists every error with its path and line. Pushes and merge groups create these runs; other events skip invalid files silently. |
445448 | A pull request's `pull_request` workflows don't run | The pull request conflicts with its base, so there is no test merge commit. Resolve the conflicts. `pull_request_target` workflows still run. |
446449 | `No Diffinery runner matches runs-on: <labels>. Linux (ubuntu-*) runners are hosted; anything else needs runs-on: [self-hosted, ...] and a registered runner.` | Hosted runners accept `ubuntu-*`, `linux`, `diffinery`, `diffinery-*`, `x64` and `amd64`. For other platforms, add `self-hosted` and start a runner with matching labels. See [Runners](/docs/actions/runners). |
modifiedapps/diffinery/docs/reference/limits.md+1−0View file
@@ -215,6 +215,7 @@
215215 | Schedule granularity | 5 minutes | Scheduled workflows are checked every 5 minutes. A cron that fires more often still runs at most once per 5-minute window. Only workflows on the default branch of repositories that aren't archived are scheduled. |
216216 | Schedule catch-up | 60 minutes | A check that comes late, or fails, is caught up by the next one, up to an hour back. Each schedule starts once per check, however many of its minutes the check covers. |
217217 | Running job whose runner stops responding | Ended 10 minutes after its `timeout-minutes` | Counted from when its runner fetched the job. The runner ends a job at its timeout itself; Diffinery only steps in when it hasn't reported by then, stops its container and fails it as timed out. |
218+| Run whose jobs have all finished but that hasn't finished | Repaired 5 minutes after its last job, finished by 30 minutes | Diffinery signals the run's orchestration again after 5 minutes, and finishes the run itself after 30 (or after 5, when the orchestration has stopped), so it no longer holds its concurrency group. Jobs it never started fail the run and are named in a `Set up workflow` job. |
218219 | Starting a job | 3 retries | Starting a job's container is retried 3 times with exponential backoff from 10 seconds before the job fails with "Could not start a runner". |
219220 | Hosted runner sizes | large: `standard-4` (4 vCPU, 12 GiB of memory, 20 GB of disk); medium: `standard-3` (2 vCPU, 8 GiB, 16 GB); small: `standard-1` (1/2 vCPU, 4 GiB, 8 GB); tiny: `basic` (1/4 vCPU, 1 GiB, 4 GB) | Each hosted job runs in its own container. A job runs on large unless `runs-on` names a size or its earlier runs show a smaller one fits with headroom ([Runner sizes](/docs/actions/runners#runner-sizes)). |
220221 | Hosted job containers at once | 100 per size | `max_instances` of each runner container class. |
modifiedapps/diffinery/src/repo/RepoDO.ts+4−3View file
@@ -324,7 +324,7 @@
324324 * fires at the earliest running job's deadline (Actions.enforceJobDeadlines).
325325 */
326326 armAlarm(at: number | null): void {
327- const next = this.earliest(at, this.actions.hasAccountHeldJobs() ? Date.now() + HELD_RETRY_MS : null, this.actions.nextJobDeadline());
327+ const next = this.earliest(at, this.actions.hasAccountHeldJobs() ? Date.now() + HELD_RETRY_MS : null, this.actions.nextJobDeadline(), this.actions.nextStuckRunCheck());
328328 this.ctx.waitUntil(next === null ? this.ctx.storage.deleteAlarm() : this.ctx.storage.setAlarm(next));
329329 }
330330
@@ -349,8 +349,9 @@
349349 await this.queue.process();
350350 if (this.actions.hasAccountHeldJobs()) await this.actions.releaseAccountHeld();
351351 await this.actions.enforceJobDeadlines();
352- // Wake up again for held jobs and running jobs' deadlines, keeping the queue's own alarm if it set an earlier one.
353- const want = this.earliest(this.actions.hasAccountHeldJobs() ? Date.now() + HELD_RETRY_MS : null, this.actions.nextJobDeadline());
352+ await this.actions.repairStuckRuns();
353+ // Wake up again for held jobs, running jobs' deadlines and stuck runs, keeping the queue's own alarm if it set an earlier one.
354+ const want = this.earliest(this.actions.hasAccountHeldJobs() ? Date.now() + HELD_RETRY_MS : null, this.actions.nextJobDeadline(), this.actions.nextStuckRunCheck());
354355 if (want !== null) {
355356 const current = await this.ctx.storage.getAlarm();
356357 if (current === null || current <= Date.now() || current > want) await this.ctx.storage.setAlarm(want);
modifiedapps/diffinery/src/repo/actions.ts+102−3View file
@@ -50,6 +50,22 @@
5050 * or vanished, which would otherwise hold their container, quota slot and minutes for up to 6 hours.
5151 */
5252 export const JOB_DEADLINE_GRACE_MS = 10 * 60_000;
53+/**
54+ * A run finishes when its workflow (runner/workflow.ts) has had every job's completion event. An event
55+ * lost while the platform updates (completeJob's sendEvent failing during a deploy) leaves the workflow
56+ * waiting up to 6 hours, and a workflow instance that failed never finishes the run: either way the run
57+ * stays in progress and holds its concurrency group. Once every job of a run has been finished this long,
58+ * Diffinery sends the jobs' completion events again (repairStuckRuns), so the workflow carries on with
59+ * the rest of the graph.
60+ */
61+export const STUCK_RUN_MS = 5 * 60_000;
62+/**
63+ * A run still stuck this long after its last job finished, though its workflow got the events again, is
64+ * finished by Diffinery itself (a run whose workflow no longer runs, after STUCK_RUN_MS).
65+ */
66+export const STUCK_RUN_FINISH_MS = 30 * 60_000;
67+/** Workflow instance states in which the run's workflow can still finish the run itself. */
68+const LIVE_WORKFLOW = new Set(["queued", "running", "paused", "waiting", "waitingForPause"]);
5369 /** timeout-minutes when a job row has none (GitHub's default). */
5470 const DEFAULT_TIMEOUT_MINUTES = 360;
5571 /** A job token never works longer than this after the job started (timeouts are at most 360 minutes). */
@@ -887,9 +903,10 @@
887903 const { order, errors } = jobOrder(exp.workflow);
888904 if (errors.length) return { jobs: [], error: errors.map((e) => e.message).join("; "), concurrencyBlocked: false };
889905 if (order.flat().length > MAX_RUN_JOBS) return { jobs: [], error: `A run may have at most ${MAX_RUN_JOBS} jobs, those of called workflows included; this one has ${order.flat().length}.`, concurrencyBlocked: false };
890- const blocked =
891- !!run.concurrency_group &&
892- this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM runs WHERE concurrency_group = ? AND status = 'in_progress' AND id != ?", run.concurrency_group, runId).one().n > 0;
906+ const holders = () => (run.concurrency_group ? this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM runs WHERE concurrency_group = ? AND status = 'in_progress' AND id != ?", run.concurrency_group, runId).one().n : 0);
907+ // A run holding the group may be stuck though its jobs all finished (STUCK_RUN_MS): repair it rather than wait behind it.
908+ if (holders()) await this.repairStuckRuns();
909+ const blocked = holders() > 0;
893910 if (!blocked) this.sql.exec("UPDATE runs SET status = 'in_progress', updated_at = ? WHERE id = ? AND status = 'queued'", now(), runId);
894911 return { jobs: order.flat().map((key) => ({ key, needs: exp.workflow.jobs[key]!.needs })), concurrencyBlocked: blocked };
895912 }
@@ -1037,6 +1054,8 @@
10371054 */
10381055 async startJobKey(runId: string, jobKey: string): Promise<{ jobIds: string[] }> {
10391056 const run = this.runRow(runId);
1057+ // Finished by repairStuckRuns while its workflow was stuck: nothing more starts.
1058+ if (run.status === "completed") return { jobIds: [] };
10401059 // A partial re-run copies what it keeps from the earlier attempt: the whole job, or some of its matrix legs.
10411060 const carry = run.rerun_carry ? (JSON.parse(run.rerun_carry) as RerunCarry)[jobKey] : undefined;
10421061 if (carry && "all" in carry) {
@@ -1693,6 +1712,82 @@
16931712 * Ends running jobs past their deadline: their runner never reported, though it stops a job at its
16941713 * timeout. Its container is stopped and the job fails as timed out, freeing the slot it held.
16951714 */
1715+ /**
1716+ * Unfinished runs whose jobs have all finished, with when the last one did or the run was created,
1717+ * whichever is later (ms since the epoch): a re-run's carried jobs keep their earlier finish times.
1718+ */
1719+ private finishedJobsRuns(): { id: string; last: number }[] {
1720+ return this.sql
1721+ .exec<{ id: string; last: string | null }>(
1722+ `SELECT r.id, max(r.created_at, coalesce((SELECT max(completed_at) FROM jobs WHERE run_id = r.id), r.created_at)) AS last FROM runs r
1723+ WHERE r.status != 'completed' AND EXISTS (SELECT 1 FROM jobs WHERE run_id = r.id)
1724+ AND NOT EXISTS (SELECT 1 FROM jobs WHERE run_id = r.id AND status != 'completed')`,
1725+ )
1726+ .toArray()
1727+ .map((r) => ({ id: r.id, last: Date.parse(r.last ?? "") || 0 }));
1728+ }
1729+
1730+ /** When repairStuckRuns next has something to do, in ms since the epoch. */
1731+ nextStuckRunCheck(at = Date.now()): number | null {
1732+ // Not yet due; due to finish (its workflow got the events again); or due but not repaired yet (its workflow's state was unknown): a minute later.
1733+ const due = this.finishedJobsRuns().map((r) => (at < r.last + STUCK_RUN_MS ? r.last + STUCK_RUN_MS : this.nudged.has(r.id) ? r.last + STUCK_RUN_FINISH_MS : at + 60_000));
1734+ return due.length ? Math.max(at, Math.min(...due)) : null;
1735+ }
1736+
1737+ /**
1738+ * Repairs runs stuck in progress though all their jobs finished (see STUCK_RUN_MS): after STUCK_RUN_MS
1739+ * their workflow gets the jobs' completion events again, and after STUCK_RUN_FINISH_MS, or after
1740+ * STUCK_RUN_MS when the workflow no longer runs, Diffinery finishes the run, which frees its concurrency group.
1741+ */
1742+ async repairStuckRuns(at = Date.now()): Promise<void> {
1743+ for (const r of this.finishedJobsRuns()) {
1744+ if (at < r.last + STUCK_RUN_MS) continue;
1745+ let instance: Awaited<ReturnType<Env["RUN_WORKFLOW"]["get"]>> | null = null;
1746+ let state: string | null = null;
1747+ try {
1748+ instance = await this.env.RUN_WORKFLOW.get(r.id);
1749+ state = (await instance.status()).status;
1750+ } catch (e) {
1751+ console.warn("run workflow status failed", r.id, e);
1752+ }
1753+ // Can't tell whether its workflow still runs: look again later (nextStuckRunCheck), up to STUCK_RUN_FINISH_MS.
1754+ if (state === null && at < r.last + STUCK_RUN_FINISH_MS) continue;
1755+ const live = state !== null && LIVE_WORKFLOW.has(state);
1756+ if (instance && live && !this.nudged.has(r.id)) {
1757+ this.nudged.add(r.id);
1758+ console.warn("run stuck after its jobs finished: sending their completion events again", r.id, state);
1759+ for (const j of this.sql.exec<{ id: string; conclusion: string | null }>("SELECT id, conclusion FROM jobs WHERE run_id = ?", r.id).toArray()) {
1760+ await instance.sendEvent({ type: `job-${j.id}`, payload: { conclusion: j.conclusion } }).catch((e: unknown) => console.warn("resend job event failed", j.id, e));
1761+ }
1762+ }
1763+ if (!live || at >= r.last + STUCK_RUN_FINISH_MS) await this.finishStuckRun(r.id, state ?? "unknown");
1764+ }
1765+ }
1766+
1767+ /** Finishes a stuck run (repairStuckRuns). Jobs its workflow never started make it fail and are named, so it doesn't pass for a success. */
1768+ private async finishStuckRun(runId: string, state: string): Promise<void> {
1769+ console.warn("run stuck after its jobs finished: finishing it", runId, state);
1770+ const run = this.runRow(runId);
1771+ if (run.conclusion !== "cancelled") {
1772+ let missing: string[] = [];
1773+ try {
1774+ const started = new Set(this.sql.exec<{ job_key: string }>("SELECT DISTINCT job_key FROM jobs WHERE run_id = ?", runId).toArray().map((j) => j.job_key));
1775+ const exp = await this.expanded(run);
1776+ missing = jobOrder(exp.workflow).order.flat().filter((k) => !started.has(k)).map((k) => exp.workflow.jobs[k]?.name ?? k);
1777+ } catch (e) {
1778+ console.warn("stuck run plan failed", runId, e);
1779+ }
1780+ if (missing.length) {
1781+ const shown = missing.slice(0, 5).join(", ") + (missing.length > 5 ? ` and ${missing.length - 5} more` : "");
1782+ await this.plannedError(runId, `The run's workflow stopped after its jobs finished, so ${missing.length === 1 ? "a job" : `${missing.length} jobs`} never started (${shown}). Re-run the workflow.`);
1783+ }
1784+ }
1785+ await this.finishRun(runId);
1786+ }
1787+
1788+ /** Runs whose completion events were sent again (repairStuckRuns), so each run gets them once per object start. */
1789+ private nudged = new Set<string>();
1790+
16961791 async enforceJobDeadlines(at = Date.now()): Promise<void> {
16971792 for (const j of this.sql.exec<JobRow>("SELECT * FROM jobs WHERE status = 'in_progress' AND started_at IS NOT NULL ORDER BY rowid").toArray()) {
16981793 const d = this.jobDeadline(j);
@@ -1831,6 +1926,8 @@
18311926
18321927 async finishRun(runId: string): Promise<RunInfo> {
18331928 const run = this.runRow(runId);
1929+ // Already finished by repairStuckRuns: the late workflow changes nothing and sends no second completed event.
1930+ if (run.status === "completed") return toRun(run);
18341931 const jobs = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ?", runId).toArray();
18351932 let conclusion: RunRow["conclusion"] = run.conclusion === "cancelled" ? "cancelled" : "success";
18361933 if (conclusion !== "cancelled") {
@@ -2375,6 +2472,8 @@
23752472 job.labels ? null : cleanUsage(completion.usage),
23762473 jobId,
23772474 );
2475+ // The run's last job: check later that the workflow finished the run (repairStuckRuns).
2476+ if (!this.sql.exec("SELECT 1 FROM jobs WHERE run_id = ? AND status != 'completed' LIMIT 1", run.id).toArray().length) this.repo.alarmBy(Date.parse(t) + STUCK_RUN_MS);
23782477 this.repo.deploys.jobCompleted(jobId, conclusion);
23792478 const m = this.repo.meta;
23802479 this.repo.upsertCheck({
modifiedapps/diffinery/test/unit/actions-runtime.test.ts+6−0View file
@@ -703,6 +703,7 @@
703703 broadcast: () => {},
704704 deploys: { jobCompleted: () => {} },
705705 runtime: { sweepJob: async () => {} },
706+ alarmBy: () => {},
706707 } as unknown as RepoDO;
707708 const env = {
708709 PUBLIC_URL: "https://d.test",
@@ -740,6 +741,7 @@
740741 broadcast: () => {},
741742 deploys: { jobCompleted: () => {} },
742743 runtime: { sweepJob: async () => {} },
744+ alarmBy: () => {},
743745 } as unknown as RepoDO;
744746 const DB = testD1();
745747 DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString());
@@ -841,6 +843,7 @@
841843 broadcast: () => {},
842844 deploys: { jobCompleted: () => {} },
843845 runtime: { sweepJob: async () => {} },
846+ alarmBy: () => {},
844847 } as unknown as RepoDO;
845848 const DB = testD1();
846849 DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString());
@@ -897,6 +900,7 @@
897900 broadcast: () => {},
898901 deploys: { jobCompleted: () => {} },
899902 runtime: { sweepJob: async () => {} },
903+ alarmBy: () => {},
900904 } as unknown as RepoDO;
901905 const DB = testD1();
902906 DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString());
@@ -955,6 +959,7 @@
955959 broadcast: () => {},
956960 deploys: { jobCompleted: () => {} },
957961 runtime: { sweepJob: async () => {} },
962+ alarmBy: () => {},
958963 } as unknown as RepoDO;
959964 const env = {
960965 PUBLIC_URL: "https://d.test",
@@ -1021,6 +1026,7 @@
10211026 deploys: { jobCompleted: () => {} },
10221027 runtime: { sweepJob: async (id: string, o?: { restart?: boolean }) => void swept.push([id, !!o?.restart]), grantFor: () => ({ role: "read", permissions: {} }) },
10231028 policy: { settings: () => ({ defaultPermissions: "write" }), isForkPull: () => false },
1029+ alarmBy: () => {},
10241030 } as unknown as RepoDO;
10251031 const DB = testD1();
10261032 DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString());
modifiedapps/diffinery/test/unit/rerun.test.ts+1−0View file
@@ -92,6 +92,7 @@
9292 deploys: { jobCompleted: () => {}, protection: () => null },
9393 runtime: { sweepJob: async () => {} },
9494 policy: { agentOrigin: () => false },
95+ alarmBy: () => {},
9596 } as unknown as RepoDO;
9697 const env = {
9798 DB,
modifiedapps/diffinery/test/unit/runner-sizes.test.ts+1−0View file
@@ -230,6 +230,7 @@
230230 deploys: { jobCompleted: () => {}, protection: () => null },
231231 runtime: { sweepJob: async () => {} },
232232 policy: { agentOrigin: () => false },
233+ alarmBy: () => {},
233234 } as unknown as RepoDO;
234235 const ns = (binding: string) => ({
235236 idFromName: (n: string) => n,
addedapps/diffinery/test/unit/stuck-runs.test.ts+208−0View file
@@ -1,0 +1,208 @@
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+});