apps/diffinery/test/unit/actions-runtime.test.ts at 683b739beb0ea3e50cf58e3a579f825990056406 · codician/diffinery · Diffinery

1 import { describe, expect, it } from "vitest";
2 import { BodyTooLarge, STEP_OPS, batchParts, blockKey, chunkKey, compose, finalKey, listStaged, parseBlockList, parseRange, planCompose, putStream, runComposeStep, stagingPrefix, type BlobBucket, type ComposeLimits, type ComposeStep } from "../../src/lib/blob-store.js";
3 import { DatabaseSync } from "node:sqlite";
4 import { Actions, CANCEL_STOP_MS, JOB_DEADLINE_GRACE_MS, MAX_ACTIVE_RUNS, MAX_PLACEMENT_TRIES, MAX_RUNNER_RESTARTS, parseRunnerInfo, placementDelay, QUICK_RETRY_SECONDS, STALE_IMAGE_RETRY_SECONDS, STALE_IMAGE_TRIES } from "../../src/repo/actions.js";
5 import { fleetLimits } from "../../src/lib/quotas.js";
6 import { testD1 } from "./stubs/d1.js";
7 import { DEFAULT_ARTIFACT_QUOTA, Runtime, artifactFileName, artifactObjectKeys } from "../../src/repo/runtime.js";
8 import { REPO_SCHEMA, migrateRepoSchema } from "../../src/repo/schema.js";
9 import type { RepoDO } from "../../src/repo/RepoDO.js";
10 import type { Env } from "../../src/env.js";
11 import { archiveKey, deletePrefixes, legacyArtifactKeys, legacyCacheKeys, logKeys, transientPrefixes } from "../../src/lib/storage-keys.js";
12 import { signBlobUrl, signRuntimeToken, verifyBlobUrl, verifyRuntimeToken } from "../../src/lib/runtime-auth.js";
13 import { b64, fromB64url, sha256 } from "../../src/lib/util.js";
14 import { completePayload, issuePayload, releasePayload } from "../../src/repo/event-payloads.js";
15 import { TokenPolicy } from "../../src/repo/token-policy.js";
16 import type { Caller, RepoMeta } from "../../src/repo/types.js";
17 import { numericId } from "../../src/http/gh.js";
18
19 /** Hosted jobs one account (and so one repository) runs at once by default. */
20 const SHARE = fleetLimits({} as Env).share;
21
22 const MASTER = { MASTER_KEY: b64(new Uint8Array(32).fill(7)) };
23 const OTHER = { MASTER_KEY: b64(new Uint8Array(32).fill(9)) };
24 const soon = () => Math.floor(Date.now() / 1000) + 600;
25
26 describe("runtime tokens", () => {
27   it("carries what the toolkit and buildkit read, and verifies only under the right key", async () => {
28     const token = await signRuntimeToken(MASTER, "https://d.example", { repo: "r1", run: "run1", job: "job1", exp: soon() }, ["refs/heads/feat", "refs/heads/main"]);
29     const claims = JSON.parse(new TextDecoder().decode(fromB64url(token.split(".")[1]!)));
30     // @actions/artifact takes the run and job backend ids from scp.
31     expect(claims.scp).toContain("Actions.Results:run1:job1");
32     // go-actions-cache needs ac (a JSON string), exp and nbf.
33     expect(JSON.parse(claims.ac)).toEqual([
34       { Scope: "refs/heads/feat", Permission: 3 },
35       { Scope: "refs/heads/main", Permission: 1 },
36     ]);
37     expect(claims.nbf).toBeLessThanOrEqual(Math.floor(Date.now() / 1000));
38     expect(await verifyRuntimeToken(MASTER, token)).toMatchObject({ repo: "r1", run: "run1", job: "job1" });
39     expect(await verifyRuntimeToken(OTHER, token)).toBeNull();
40     expect(await verifyRuntimeToken(MASTER, token.slice(0, -2) + "xx")).toBeNull();
41     expect(await verifyRuntimeToken(MASTER, "not.a.token")).toBeNull();
42   });
43
44   it("rejects expired tokens and tampered claims", async () => {
45     const expired = await signRuntimeToken(MASTER, "x", { repo: "r", run: "a", job: "b", exp: Math.floor(Date.now() / 1000) - 5 }, ["refs/heads/main"]);
46     expect(await verifyRuntimeToken(MASTER, expired)).toBeNull();
47     const token = await signRuntimeToken(MASTER, "x", { repo: "r", run: "a", job: "b", exp: soon() }, ["refs/heads/main"]);
48     const [h, p, s] = token.split(".");
49     const forged = JSON.parse(new TextDecoder().decode(fromB64url(p!)));
50     forged.repo = "victim";
51     const p2 = Buffer.from(JSON.stringify(forged)).toString("base64url");
52     expect(await verifyRuntimeToken(MASTER, `${h}.${p2}.${s}`)).toBeNull();
53   });
54 });
55
56 describe("signed blob URLs", () => {
57   it("grant one permission on one blob until they expire", async () => {
58     const ref = { repo: "r1", kind: "artifact" as const, id: 42 };
59     const url = new URL(await signBlobUrl(MASTER, "https://d.example", ref, "r", soon(), "my artifact.zip"));
60     expect(url.pathname).toBe("/_runtime/blob/r1/artifact/42/my%20artifact.zip");
61     expect(await verifyBlobUrl(MASTER, ref, "r", url.searchParams)).toBe(true);
62     expect(await verifyBlobUrl(MASTER, ref, "w", url.searchParams)).toBe(false);
63     expect(await verifyBlobUrl(MASTER, { ...ref, id: 43 }, "r", url.searchParams)).toBe(false);
64     expect(await verifyBlobUrl(MASTER, { ...ref, kind: "cache" }, "r", url.searchParams)).toBe(false);
65     expect(await verifyBlobUrl(OTHER, ref, "r", url.searchParams)).toBe(false);
66     const stale = new URL(await signBlobUrl(MASTER, "https://d.example", ref, "r", Math.floor(Date.now() / 1000) - 1));
67     expect(await verifyBlobUrl(MASTER, ref, "r", stale.searchParams)).toBe(false);
68     url.searchParams.set("se", String(soon() + 100));
69     expect(await verifyBlobUrl(MASTER, ref, "r", url.searchParams)).toBe(false);
70   });
71 });
72
73 describe("blob protocol helpers", () => {
74   it("parses Azure block lists in order", () => {
75     const xml = `<?xml version="1.0" encoding="utf-8"?><BlockList><Latest>AAA=</Latest><Uncommitted>Qg==</Uncommitted><Committed>a&amp;b</Committed></BlockList>`;
76     expect(parseBlockList(xml)).toEqual(["AAA=", "Qg==", "a&b"]);
77   });
78
79   it("parses byte ranges like Azure and HTTP", () => {
80     expect(parseRange(null, 10)).toBeNull();
81     expect(parseRange("bytes=0-", 10)).toEqual({ offset: 0, length: 10 });
82     expect(parseRange("bytes=2-4", 10)).toEqual({ offset: 2, length: 3 });
83     expect(parseRange("bytes=5-100", 10)).toEqual({ offset: 5, length: 5 });
84     expect(parseRange("bytes=-3", 10)).toEqual({ offset: 7, length: 3 });
85     expect(parseRange("bytes=10-", 10)).toBe("invalid");
86     expect(parseRange("bytes=4-2", 10)).toBe("invalid");
87     expect(parseRange("items=0-1", 10)).toBe("invalid");
88   });
89
90   it("keeps staged objects per blob, and transient data outside repos/", () => {
91     const a = { repo: "r", kind: "cache" as const, id: 1 };
92     expect(finalKey(a)).toBe("actions-cache/r/1");
93     expect(finalKey({ ...a, kind: "artifact" })).toBe("artifacts/r/1");
94     expect(blockKey(a, "x/y")).toMatch(/^actions-staging\/r\/cache-1\/b-[A-Za-z0-9_-]+$/);
95     expect(chunkKey(a, 1024)).toBe("actions-staging/r/cache-1/o-0000000000001024");
96     // Old keys are still read, after the new ones.
97     expect(logKeys("r", "j")).toEqual(["logs/r/j.json", "repos/r/logs/j.json"]);
98     expect(legacyCacheKeys("r", "a b")).toEqual(["actions-cache/r/legacy/a%20b", "repos/r/cache/a%20b"]);
99     expect(legacyArtifactKeys("r", "run", "n")).toEqual(["artifacts/r/legacy/run/n", "repos/r/artifacts/run/n"]);
100     // Nothing transient is written under repos/, which has no age rule (git objects, LFS).
101     for (const k of [finalKey(a), finalKey({ ...a, kind: "artifact" }), blockKey(a, "b"), logKeys("r", "j")[0], legacyCacheKeys("r", "k")[0], legacyArtifactKeys("r", "x", "y")[0]]) expect(k.startsWith("repos/")).toBe(false);
102   });
103 });
104
105 /** An in-memory stand-in for the R2 bucket, enforcing R2's multipart rules and counting calls. */
106 function memoryBucket() {
107   const objects = new Map<string, { data: Uint8Array; etag: string }>();
108   const uploads = new Map<string, { key: string; parts: Map<number, Uint8Array> }>();
109   const stats = { multipart: 0, parts: [] as number[], ops: 0 };
110   let n = 0;
111   let listing: { prefix: string; all: Array<{ key: string; size: number }> } | undefined;
112   const bytes = async (v: ReadableStream | ArrayBuffer | ArrayBufferView) =>
113     v instanceof ReadableStream ? new Uint8Array(await new Response(v).arrayBuffer()) : v instanceof ArrayBuffer ? new Uint8Array(v) : new Uint8Array(v.buffer, v.byteOffset, v.byteLength).slice();
114   const store = (key: string, data: Uint8Array) => {
115     const etag = `e${++n}`;
116     objects.set(key, { data, etag });
117     return { size: data.length, etag };
118   };
119   const handle = (uploadId: string) => {
120     const u = uploads.get(uploadId)!;
121     return {
122       uploadId,
123       async uploadPart(p: number, value: ReadableStream | ArrayBuffer | ArrayBufferView) {
124         stats.ops++;
125         u.parts.set(p, await bytes(value));
126         return { partNumber: p, etag: `p${p}` };
127       },
128       async complete(list: Array<{ partNumber: number; etag: string }>) {
129         stats.ops++;
130         const sizes = list.map((p) => u.parts.get(p.partNumber)!.length);
131         sizes.slice(0, -1).forEach((s) => expect(s).toBeGreaterThanOrEqual(5 * 1024 * 1024));
132         sizes.slice(0, -1).forEach((s) => expect(s).toBe(sizes[0]));
133         expect(list.map((p) => p.partNumber)).toEqual(list.map((_, i) => i + 1));
134         stats.parts = sizes;
135         const out = new Uint8Array(sizes.reduce((a, b) => a + b, 0));
136         let at = 0;
137         for (const p of list) {
138           out.set(u.parts.get(p.partNumber)!, at);
139           at += u.parts.get(p.partNumber)!.length;
140         }
141         store(u.key, out);
142         return {};
143       },
144       async abort() {
145         stats.ops++;
146         uploads.delete(uploadId);
147       },
148     };
149   };
150   const bucket: BlobBucket = {
151     async get(key, o) {
152       stats.ops++;
153       const obj = objects.get(key);
154       if (!obj) return null;
155       const data = o?.range ? obj.data.subarray(o.range.offset, o.range.offset + o.range.length) : obj.data;
156       return { body: new Response(data).body!, size: data.length };
157     },
158     async head(key) {
159       stats.ops++;
160       const o = objects.get(key);
161       return o ? { size: o.data.length, etag: o.etag } : null;
162     },
163     async put(key, value) {
164       stats.ops++;
165       return store(key, await bytes(value));
166     },
167     async list({ prefix, cursor, limit }) {
168       stats.ops++;
169       // A listing pages through one snapshot (the cursor is an index into it), so 50k objects aren't filtered again for every page.
170       const start = Number(cursor ?? 0);
171       if (!start || listing?.prefix !== prefix) listing = { prefix, all: [...objects].filter(([k]) => k.startsWith(prefix)).map(([key, v]) => ({ key, size: v.data.length })) };
172       const all = listing.all;
173       const page = all.slice(start, start + (limit ?? 1000));
174       const truncated = start + page.length < all.length;
175       return { objects: page, truncated, cursor: truncated ? String(start + page.length) : undefined };
176     },
177     async delete(keys) {
178       stats.ops++;
179       for (const k of Array.isArray(keys) ? keys : [keys]) objects.delete(k);
180     },
181     async createMultipartUpload(key) {
182       stats.ops++;
183       stats.multipart++;
184       const id = `u${++n}`;
185       uploads.set(id, { key, parts: new Map() });
186       return handle(id);
187     },
188     resumeMultipartUpload(_key, uploadId) {
189       return handle(uploadId);
190     },
191   };
192   return { bucket, objects, stats };
193 }
194
195 const filled = (n: number, seed: number) => {
196   const b = new Uint8Array(n);
197   for (let i = 0; i < n; i++) b[i] = (i * 31 + seed) & 255;
198   return b;
199 };
200
201 describe("compose", () => {
202   const ref = { repo: "r", kind: "artifact" as const, id: 9 };
203   const MiB = 1024 * 1024;
204
205   /**
206    * Composes blocks of `sizes`, running every step as production does (an invocation
207    * of its own), and checks the bytes. Returns the calls the Worker made and the most
208    * any single step made.
209    */
210   async function run(sizes: number[], limits?: ComposeLimits) {
211     const m = memoryBucket();
212     const segs = [];
213     for (const [i, n] of sizes.entries()) {
214       const key = blockKey(ref, `block-${i}`);
215       await m.bucket.put(key, filled(n, i));
216       segs.push({ key, size: n });
217     }
218     const staged = await listStaged(m.bucket, ref);
219     expect(staged.size).toBe(sizes.length);
220     m.stats.ops = 0;
221     let steps = 0;
222     let maxStepOps = 0;
223     const runStep = async (step: ComposeStep) => {
224       steps++;
225       const before = m.stats.ops;
226       const out = await runComposeStep(m.bucket, step);
227       const used = m.stats.ops - before;
228       maxStepOps = Math.max(maxStepOps, used);
229       // the Worker's own calls exclude the steps'
230       m.stats.ops = before;
231       return out;
232     };
233     const total = await compose(m.bucket, finalKey(ref), segs, "application/zip", { scratch: stagingPrefix(ref), run: runStep, concurrency: 1, limits });
234     const out = m.objects.get(finalKey(ref))!.data;
235     const expected = new Uint8Array(total);
236     let at = 0;
237     for (const [i, n] of sizes.entries()) {
238       expected.set(filled(n, i), at);
239       at += n;
240     }
241     expect(out.length).toBe(expected.length);
242     expect(Buffer.from(out).equals(Buffer.from(expected))).toBe(true);
243     // intermediates are removed; only the blocks (deleted by the caller) and the result remain
244     expect([...m.objects.keys()].filter((k) => !staged.has(k) && k !== finalKey(ref))).toEqual([]);
245     return { ...m.stats, workerOps: m.stats.ops, steps, maxStepOps };
246   }
247
248   it("handles empty and single uploads", async () => {
249     expect((await run([])).multipart).toBe(0);
250     expect((await run([123])).multipart).toBe(0);
251   });
252
253   it("streams a few blocks into one object", async () => {
254     const r = await run([1000, 7, 1 << 20, 8 * MiB]);
255     expect(r.multipart).toBe(0);
256     expect(r.steps).toBe(1);
257   });
258
259   it("writes large uploads as equal multipart parts, cutting across blocks", async () => {
260     const small = { stepOps: 400, maxMergeBytes: 4 * MiB };
261     expect((await run([8 * MiB, 8 * MiB, 3 * MiB], small)).parts).toEqual([8 * MiB, 8 * MiB, 3 * MiB]);
262     // Go SDK style 1 MiB blocks are re-cut into 8 MiB parts (ranged reads across block boundaries)
263     expect((await run(Array.from({ length: 19 }, (_, i) => (i === 18 ? 12345 : MiB)), small)).parts).toEqual([8 * MiB, 8 * MiB, 2 * MiB + 12345]);
264     expect((await run([3 * MiB, 7 * MiB, 5 * MiB, 1], small)).parts).toEqual([8 * MiB, 7 * MiB + 1]);
265   });
266
267   it("copies a large upload as multipart parts spread over many steps, not one long put", async () => {
268     // The runner's 32 MiB blocks: one put would copy 101 MiB in a single step (about 15 s); parts of
269     // 8 MiB, two per step, let eight steps copy at once.
270     const r = await run([32 * MiB, 32 * MiB, 32 * MiB, 5 * MiB]);
271     expect(r.multipart).toBe(1);
272     expect(r.parts).toEqual([...Array.from({ length: 12 }, () => 8 * MiB), 5 * MiB]);
273     expect(r.steps).toBe(7);
274     // Up to 16 MiB still goes in one put.
275     expect((await run([16 * MiB])).multipart).toBe(0);
276     expect((await run([16 * MiB, 1])).multipart).toBe(1);
277   });
278
279   it("fails a step whose staged block went missing, instead of writing a short part", async () => {
280     const m = memoryBucket();
281     await m.bucket.put("s/a", filled(1000, 1));
282     const step: ComposeStep = { kind: "merge", target: "s/out", slices: [{ key: "s/a", offset: 0, length: 1000, whole: true }, { key: "s/gone", offset: 0, length: 10, whole: true }], length: 1010 };
283     await expect(runComposeStep(m.bucket, step)).rejects.toThrow(/s\/gone is missing/);
284     expect(m.objects.has("s/out")).toBe(false);
285   });
286
287   it("merges many small blocks in rounds so that no step exceeds its calls", async () => {
288     // 10 calls per step: 300 blocks need two rounds of merging before the parts
289     const r = await run(Array.from({ length: 300 }, (_, i) => 40 * 1024 + i), { stepOps: 10, maxMergeBytes: 1 * MiB });
290     expect(r.maxStepOps).toBeLessThanOrEqual(10);
291     expect(r.multipart).toBe(1);
292   });
293
294   it("keeps a 50,000-block list (Azure's maximum) within the Workers subrequest limit", async () => {
295     // Real data: 50k tiny blocks, as buildkit's gha cache can send.
296     const r = await run(Array.from({ length: 50_000 }, (_, i) => 16 + (i % 7)));
297     expect(r.maxStepOps).toBeLessThanOrEqual(STEP_OPS);
298     expect(r.steps + r.workerOps).toBeLessThan(1000);
299     // Planned only (no bytes): 50k blocks of 200 KiB (10 GB), and of 1 MiB with 4 MiB blocks mixed in.
300     for (const sizes of [Array.from({ length: 50_000 }, () => 200 * 1024), Array.from({ length: 50_000 }, (_, i) => (i % 10 ? MiB : 4 * MiB))]) {
301       const plan = planCompose(sizes.map((size, i) => ({ key: `k${i}`, size })), "s/");
302       const steps = [...plan.rounds.flat(), ...batchParts(plan.parts)];
303       for (const s of steps) {
304         const ops = "kind" in s ? s.slices.length + 1 : s.reduce((n, p) => n + p.slices.length + 1, 0);
305         expect(ops).toBeLessThanOrEqual(STEP_OPS);
306       }
307       // list pages and deletes of 50k staged blocks (50 each), the steps, create and complete
308       expect(50 + 50 + steps.length + 2).toBeLessThan(1000);
309       expect(plan.parts.length).toBeLessThanOrEqual(10_000);
310       expect(plan.parts.slice(0, -1).every((p) => p.reduce((n, s) => n + s.length, 0) === plan.partSize)).toBe(true);
311     }
312     // Writes and composes 50k objects: about 1.5s alone, and several times that on a busy CI runner.
313   }, 20_000);
314 });
315
316 describe("deleting a repository's transient objects", () => {
317   it("removes every object under its prefixes, a page at a time, and nothing else", async () => {
318     const m = memoryBucket();
319     const keys = [...Array.from({ length: 2500 }, (_, i) => `actions-cache/r1/${i}`), "artifacts/r1/1", "logs/r1/j.json", "actions-staging/r1/cache-1/b-x", archiveKey("r1", "abc", "zip")];
320     for (const k of keys) await m.bucket.put(k, new Uint8Array(1));
321     for (const k of ["actions-cache/r10/1", "repos/r1/objects/ab/cd", "artifacts/r2/1"]) await m.bucket.put(k, new Uint8Array(1));
322     m.stats.ops = 0;
323     expect(transientPrefixes("r1").some((p) => p.startsWith("repos/"))).toBe(false);
324     expect(await deletePrefixes(m.bucket, transientPrefixes("r1"))).toBe(true);
325     expect([...m.objects.keys()].sort()).toEqual(["actions-cache/r10/1", "artifacts/r2/1", "repos/r1/objects/ab/cd"]);
326     // 3 pages of caches (list + delete each), then a page for each other prefix
327     expect(m.stats.ops).toBe(6 + 4 * 2);
328     // stops within its budget and says so
329     for (let i = 0; i < 3000; i++) await m.bucket.put(`logs/r3/${i}`, new Uint8Array(1));
330     expect(await deletePrefixes(m.bucket, transientPrefixes("r3"), 4)).toBe(false);
331   });
332 });
333
334 describe("streaming bodies of unknown length", () => {
335   const stream = (chunks: number[]) =>
336     new ReadableStream<Uint8Array>({
337       start(ctl) {
338         chunks.forEach((n, i) => ctl.enqueue(filled(n, i)));
339         ctl.close();
340       },
341     });
342
343   it("stores small bodies in one put and large ones as parts, a part in memory at a time", async () => {
344     const m = memoryBucket();
345     expect(await putStream(m.bucket, "a", stream([10, 20]), 100)).toBe(30);
346     expect(m.stats.multipart).toBe(0);
347     const MiB = 1024 * 1024;
348     expect(await putStream(m.bucket, "b", stream([3 * MiB, 3 * MiB, 3 * MiB, 3 * MiB, 1]), 64 * MiB, undefined, 5 * MiB)).toBe(12 * MiB + 1);
349     expect(m.stats.parts).toEqual([5 * MiB, 5 * MiB, 2 * MiB + 1]);
350     expect(m.objects.get("b")!.data.length).toBe(12 * MiB + 1);
351   });
352
353   it("refuses more than the limit and stores nothing", async () => {
354     const m = memoryBucket();
355     await expect(putStream(m.bucket, "c", stream([60, 60]), 100)).rejects.toBeInstanceOf(BodyTooLarge);
356     await expect(putStream(m.bucket, "d", stream([6 * 1024 * 1024, 6 * 1024 * 1024]), 10 * 1024 * 1024, undefined, 5 * 1024 * 1024)).rejects.toBeInstanceOf(BodyTooLarge);
357     expect(m.objects.has("c")).toBe(false);
358     expect(m.objects.has("d")).toBe(false);
359   });
360 });
361
362 describe("event payloads", () => {
363   const meta: RepoMeta = { repoId: "r1", ownerId: "o1", owner: "acme", name: "app", defaultBranch: "main", ownerKind: "org" };
364   const site = { publicUrl: "https://d.example", meta };
365
366   it("adds repository, sender and organization like GitHub", () => {
367     const p = completePayload(site, { ref: "refs/heads/main", sender: { login: "ada" } }, { id: "u1", login: "ada", kind: "user" });
368     const repo = p.repository as Record<string, unknown>;
369     expect(repo).toMatchObject({ name: "app", full_name: "acme/app", default_branch: "main", clone_url: "https://d.example/acme/app.git", url: "https://d.example/api/v3/repos/acme/app" });
370     expect(repo.owner).toMatchObject({ login: "acme", type: "Organization" });
371     expect(p.sender).toMatchObject({ login: "ada", id: numericId("u1"), type: "User" });
372     expect(p.organization).toMatchObject({ login: "acme" });
373     expect(p.ref).toBe("refs/heads/main");
374   });
375
376   it("completes pull_request with head.repo and keeps trigger fields", () => {
377     const p = completePayload(site, { action: "opened", number: 3, pull_request: { number: 3, title: "T", head: { ref: "feat", sha: "abc" }, base: { ref: "main", sha: "def" }, user: { login: "bob" } } }, { login: "bob", kind: "agent", id: "a1" });
378     const pr = p.pull_request as Record<string, any>;
379     expect(pr.head).toMatchObject({ ref: "feat", sha: "abc", label: "acme:feat" });
380     expect(pr.head.repo.full_name).toBe("acme/app");
381     expect(pr.base.repo.full_name).toBe("acme/app");
382     // Numeric ids, as on GitHub and in the REST API.
383     expect(pr.head.repo.id).toBe(numericId("r1"));
384     expect(pr.base.repo.id).toBe(numericId("r1"));
385     expect(typeof pr.user.id).toBe("number");
386     expect((p.organization as Record<string, unknown>).id).toBe(numericId("o1"));
387     expect(pr).toMatchObject({ number: 3, title: "T", state: "open", html_url: "https://d.example/acme/app/pull/3" });
388     expect(pr.user.login).toBe("bob");
389     expect((p.sender as Record<string, unknown>).type).toBe("Bot");
390   });
391
392   it("builds issue and release objects", () => {
393     const issue = issuePayload(site, { number: 5, kind: "issue", title: "Bug", body: "b", state: "open", stateReason: null, author: { id: "u", login: "ada", kind: "user" }, assignees: ["bob"], labels: ["bug"], createdAt: "t", updatedAt: "t", closedAt: null, comments: 2 });
394     expect(issue).toMatchObject({ number: 5, title: "Bug", labels: [{ name: "bug" }], comments: 2, html_url: "https://d.example/acme/app/issues/5" });
395     expect(issue.pull_request).toBeUndefined();
396     const rel = releasePayload(site, { id: 1, tagName: "v1.0.0", targetCommitish: "main", name: "v1", body: null, draft: false, prerelease: false, createdAt: "t", publishedAt: "t", author: { login: "ada" } });
397     expect(rel).toMatchObject({ tag_name: "v1.0.0", html_url: "https://d.example/acme/app/releases/tag/v1.0.0", upload_url: "https://d.example/api/v3/repos/acme/app/releases/1/assets{?name,label}" });
398   });
399 });
400
401 /** The Durable Object's SqlStorage over node:sqlite: exec(query, ...bindings) with toArray() and one(). */
402 function sqlStorage() {
403   const db = new DatabaseSync(":memory:");
404   return {
405     exec(query: string, ...args: unknown[]) {
406       const params = args.map((a) => (a === undefined ? null : a)) as Array<string | number | null>;
407       let rows: unknown[] = [];
408       if (!params.length && query.trim().replace(/;\s*$/, "").includes(";")) db.exec(query);
409       else rows = db.prepare(query).all(...params);
410       return {
411         toArray: () => rows,
412         one: () => {
413           if (rows.length !== 1) throw new Error(`expected one row, got ${rows.length}`);
414           return rows[0];
415         },
416       };
417     },
418   };
419 }
420
421 /** A Runtime over a fresh repository schema, an in-memory bucket, and jobs running on given refs. */
422 function runtimeFixture() {
423   const sql = sqlStorage();
424   sql.exec(REPO_SCHEMA);
425   migrateRepoSchema(sql as unknown as SqlStorage);
426   const m = memoryBucket();
427   const repo = { sql, meta: { repoId: "r", defaultBranch: "main" } } as unknown as RepoDO & { runtime: Runtime };
428   const env = { BUCKET: m.bucket } as unknown as Env;
429   const rt = new Runtime(repo, env);
430   repo.runtime = rt;
431   // The runner API's side of older runners' caches and artifacts, with every job token taken as valid.
432   const actions = new Actions(repo, env);
433   Object.defineProperty(actions, "authJob", { value: async (id: string) => sql.exec("SELECT * FROM jobs WHERE id = ?", id).one() });
434   let n = 0;
435   /** A running job of a run on `ref`; `payload` is the run's event payload. */
436   const job = (ref: string, payload: Record<string, unknown> = {}) => {
437     n++;
438     const t = new Date().toISOString();
439     sql.exec(
440       "INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, event_payload) VALUES (?, ?, 'w.yml', 'w', 'push', ?, 'abc', '{}', 'in_progress', ?, ?, ?)",
441       `run${n}`,
442       n,
443       ref,
444       t,
445       t,
446       JSON.stringify(payload),
447     );
448     sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES (?, ?, 'j', 'j', 'in_progress')", `job${n}`, `run${n}`);
449     return `job${n}`;
450   };
451   const version = "ab".repeat(32);
452   /** Saves `key` from `jobId` (reserve, then commit). */
453   const save = async (jobId: string, key: string) => {
454     const r = await rt.cacheReserve(jobId, key, version);
455     expect(r).not.toBeNull();
456     await rt.cacheCommit(jobId, r!.id, 10, "etag");
457   };
458   const restore = (jobId: string, ...keys: string[]) => rt.cacheLookup(jobId, keys, version);
459   return { sql, rt, actions, m, job, save, restore };
460 }
461
462 describe("cache scopes (as on GitHub)", () => {
463   it("restores from the job's own ref, then the default branch; never another branch's", async () => {
464     const f = runtimeFixture();
465     const main = f.job("refs/heads/main");
466     await f.save(main, "deps-main");
467     const a = f.job("refs/heads/a");
468     await f.save(a, "deps-a");
469     // a restores its own entry first, else main's
470     expect(f.restore(a, "deps-a")?.scope).toBe("refs/heads/a");
471     expect(f.restore(a, "nothing", "deps-")?.key).toBe("deps-a");
472     expect(f.restore(a, "deps-main")?.scope).toBe("refs/heads/main");
473     // b sees main's, not a's newer one, even by prefix
474     const b = f.job("refs/heads/b");
475     expect(f.restore(b, "deps-a")).toBeNull();
476     expect(f.restore(b, "deps-")?.key).toBe("deps-main");
477     // the default branch never sees a branch's caches
478     const main2 = f.job("refs/heads/main");
479     expect(f.restore(main2, "deps-a")).toBeNull();
480     expect(f.restore(main2, "deps-")?.key).toBe("deps-main");
481     // saves go to the job's own ref: b may save a key that a already has
482     await f.save(b, "deps-a");
483     expect(f.restore(b, "deps-a")?.scope).toBe("refs/heads/b");
484   });
485
486   it("gives pull requests their base branch's caches but not other pull requests'", async () => {
487     const f = runtimeFixture();
488     await f.save(f.job("refs/heads/main"), "k-main");
489     await f.save(f.job("refs/heads/release"), "k-release");
490     const pr1 = f.job("refs/pull/1/merge", { pull_request: { number: 1, base: { ref: "release" } } });
491     expect(f.restore(pr1, "k-")?.key).toBe("k-release");
492     await f.save(pr1, "k-pr1");
493     expect(f.restore(pr1, "k-")?.key).toBe("k-pr1");
494     const pr2 = f.job("refs/pull/2/merge", { pull_request: { number: 2, base: { ref: "main" } } });
495     expect(f.restore(pr2, "k-pr1")).toBeNull();
496     expect(f.restore(pr2, "k-release")).toBeNull();
497     expect(f.restore(pr2, "k-")?.key).toBe("k-main");
498     // nor the base branch or the default branch theirs
499     expect(f.restore(f.job("refs/heads/release"), "k-pr1")).toBeNull();
500     expect(f.restore(f.job("refs/heads/main"), "k-pr")).toBeNull();
501     // a merge group reads its target branch's caches
502     const group = f.job("refs/heads/gh-readonly-queue/release/pr-1-abc", { merge_group: { base_ref: "refs/heads/release" } });
503     expect(f.restore(group, "k-")?.key).toBe("k-release");
504   });
505 });
506
507 describe("artifact retention", () => {
508   const DAY = 86400_000;
509   async function upload(f: ReturnType<typeof runtimeFixture>, jobId: string, name: string, expiresAt?: string) {
510     const { id } = await f.rt.artifactCreate(jobId, name, undefined, expiresAt);
511     await f.m.bucket.put(finalKey({ repo: "r", kind: "artifact", id }), new Uint8Array(100));
512     await f.rt.stage("artifact", id, 100);
513     return f.rt.artifactFinalize(jobId, name, 100, undefined, "etag");
514   }
515   const days = (iso: string) => Math.round((Date.parse(iso) - Date.now()) / DAY);
516
517   it("honours retention-days, clamped to 1..90 days", async () => {
518     const f = runtimeFixture();
519     const j = f.job("refs/heads/main");
520     expect(days((await upload(f, j, "five", new Date(Date.now() + 5 * DAY).toISOString())).expiresAt)).toBe(5);
521     expect(days((await upload(f, j, "default")).expiresAt)).toBe(90);
522     expect(days((await upload(f, j, "too-long", new Date(Date.now() + 400 * DAY).toISOString())).expiresAt)).toBe(90);
523     expect(days((await upload(f, j, "too-short", new Date(Date.now() - DAY).toISOString())).expiresAt)).toBe(1);
524     expect(days((await upload(f, j, "garbage", "next tuesday")).expiresAt)).toBe(90);
525   });
526
527   it("deletes an expired artifact's content and lists it as expired", async () => {
528     const f = runtimeFixture();
529     const j = f.job("refs/heads/main");
530     const a = await upload(f, j, "old", new Date(Date.now() + 2 * DAY).toISOString());
531     const keep = await upload(f, j, "new", new Date(Date.now() + 3 * DAY).toISOString());
532     expect(a.expired).toBe(false);
533     f.sql.exec("UPDATE rt_artifacts SET expires_at = ? WHERE id = ?", new Date(Date.now() - 1000).toISOString(), a.id);
534     // Already shown as expired (and not handed to the toolkit) before the sweep runs.
535     expect(f.rt.artifactGet(a.id)).toBeNull();
536     expect(f.rt.artifactGet(a.id, { includeExpired: true })?.expired).toBe(true);
537     // A job ending sweeps: the content goes, the listing stays.
538     await f.rt.sweepJob(j);
539     expect(f.m.objects.has(finalKey({ repo: "r", kind: "artifact", id: a.id }))).toBe(false);
540     expect(f.m.objects.has(finalKey({ repo: "r", kind: "artifact", id: keep.id }))).toBe(true);
541     expect(f.rt.runArtifacts("run1").map((x) => x.name)).toEqual(["new"]);
542     expect(f.rt.runArtifacts("run1", { includeExpired: true }).map((x) => [x.name, x.expired])).toEqual([
543       ["new", false],
544       ["old", true],
545     ]);
546     const listed = f.rt.repoArtifacts({});
547     expect(listed.artifacts.find((x) => x.id === a.id)?.expired).toBe(true);
548     // Expired artifacts no longer count against the quota.
549     expect(f.sql.exec("SELECT status, staged FROM rt_artifacts WHERE id = ?", a.id).one()).toEqual({ status: "expired", staged: 0 });
550     // The name can be used again, and 90 days on the listing goes too.
551     const again = f.job("refs/heads/main");
552     f.sql.exec("UPDATE jobs SET run_id = 'run1' WHERE id = ?", again);
553     expect((await upload(f, again, "old")).expired).toBe(false);
554     f.sql.exec("UPDATE rt_artifacts SET expires_at = ? WHERE id = ?", new Date(Date.now() - 91 * DAY).toISOString(), keep.id);
555     await f.rt.expireArtifacts();
556     await f.rt.expireArtifacts();
557     expect(f.rt.artifactGet(keep.id, { includeExpired: true })).toBeNull();
558   });
559
560   it("a job starting again from the beginning drops what its stopped attempt uploaded, and nothing else", async () => {
561     const f = runtimeFixture();
562     const j = f.job("refs/heads/main");
563     const other = f.job("refs/heads/main");
564     const mine = await upload(f, j, "report");
565     const theirs = await upload(f, other, "report");
566     await f.save(j, "deps-1");
567     await f.save(other, "deps-2");
568     // A job's end keeps what it finished uploading.
569     await f.rt.sweepJob(j);
570     expect(f.rt.artifactGet(mine.id)).not.toBeNull();
571     // Its restart doesn't: the new attempt uploads its own, under the same names.
572     await f.rt.sweepJob(j, { restart: true });
573     expect(f.rt.artifactGet(mine.id, { includeExpired: true })).toBeNull();
574     expect(f.m.objects.has(finalKey({ repo: "r", kind: "artifact", id: mine.id }))).toBe(false);
575     expect(f.restore(j, "deps-1")).toBeNull();
576     expect(f.rt.artifactGet(theirs.id)).not.toBeNull();
577     expect(f.restore(other, "deps-2")).not.toBeNull();
578     expect((await upload(f, j, "report")).expired).toBe(false);
579   });
580 });
581
582 describe("artifacts from older runners", () => {
583   const DAY = 86400_000;
584   /** What the runner API does with an older runner's upload: check the name, store the tarball, record it. */
585   async function legacyUpload(f: ReturnType<typeof runtimeFixture>, jobId: string, name: string, size: number) {
586     const { r2Keys } = await f.actions.artifactUploadKeys(jobId, "djt_tok", name);
587     await f.m.bucket.put(r2Keys[0]!, new Uint8Array(size));
588     await f.actions.artifactSaved(jobId, "djt_tok", name, size);
589     return r2Keys;
590   }
591
592   it("are listed with numeric ids, and can be read and deleted like the others", async () => {
593     const f = runtimeFixture();
594     const j = f.job("refs/heads/main");
595     const keys = await legacyUpload(f, j, "logs", 40);
596     expect(keys).toEqual(legacyArtifactKeys("r", "run1", "logs"));
597     const listed = f.rt.repoArtifacts({});
598     expect(listed.total).toBe(1);
599     const a = listed.artifacts[0]!;
600     expect(a).toMatchObject({ name: "logs", size: 40, expired: false, runId: "run1", runNumber: 1, mimeType: "application/gzip" });
601     expect(Number.isSafeInteger(a.id) && a.id > 0).toBe(true);
602     expect(f.rt.runArtifacts("run1", { includeExpired: true }).map((x) => x.id)).toEqual([a.id]);
603     expect(f.rt.runArtifacts("1", { includeExpired: true }).map((x) => x.id)).toEqual([a.id]);
604     expect(f.rt.artifactGet(a.id, { includeExpired: true })?.name).toBe("logs");
605     // Its content stays where the runner put it, and downloads as a tarball.
606     expect(artifactObjectKeys("r", a)).toEqual(keys);
607     expect(artifactFileName(a)).toBe("logs.tar.gz");
608     expect(f.actions.artifacts("run1")).toMatchObject([{ id: a.id, name: "logs", r2Keys: keys, fileName: "logs.tar.gz", contentType: "application/gzip" }]);
609     // It counts against the quota, and is kept 90 days (upload-artifact's default).
610     expect(await f.rt.artifactRoom()).toBe(DEFAULT_ARTIFACT_QUOTA - 40);
611     expect(Math.round((Date.parse(a.expiresAt) - Date.now()) / DAY)).toBe(90);
612     // Uploading the name again replaces it.
613     await legacyUpload(f, j, "logs", 50);
614     const again = f.rt.repoArtifacts({}).artifacts;
615     expect(again.map((x) => [x.name, x.size])).toEqual([["logs", 50]]);
616     // Deleting it deletes its content.
617     expect((await f.rt.artifactDelete(again[0]!.id))?.name).toBe("logs");
618     expect(f.m.objects.has(keys[0]!)).toBe(false);
619     expect(f.rt.repoArtifacts({}).total).toBe(0);
620     expect(await f.rt.artifactRoom()).toBe(DEFAULT_ARTIFACT_QUOTA);
621   });
622
623   it("expire like the others", async () => {
624     const f = runtimeFixture();
625     const j = f.job("refs/heads/main");
626     const [key] = await legacyUpload(f, j, "old", 10);
627     const a = f.rt.repoArtifacts({}).artifacts[0]!;
628     f.sql.exec("UPDATE rt_artifacts SET expires_at = ? WHERE id = ?", new Date(Date.now() - 1000).toISOString(), a.id);
629     await f.rt.expireArtifacts();
630     expect(f.m.objects.has(key!)).toBe(false);
631     expect(f.rt.artifactGet(a.id, { includeExpired: true })?.expired).toBe(true);
632     expect(await f.rt.artifactRoom()).toBe(DEFAULT_ARTIFACT_QUOTA);
633   });
634
635   it("share names with the artifact service's uploads, which don't list them", async () => {
636     const f = runtimeFixture();
637     const j = f.job("refs/heads/main");
638     await legacyUpload(f, j, "tarball", 10);
639     // The toolkit talks to GitHub's v4 service, which doesn't list v3 artifacts either.
640     expect(f.rt.jobArtifacts(j)).toEqual([]);
641     await expect(f.rt.artifactCreate(j, "tarball", undefined, undefined)).rejects.toThrow(/already exists/);
642     const { id } = await f.rt.artifactCreate(j, "zip", undefined, undefined);
643     await f.m.bucket.put(finalKey({ repo: "r", kind: "artifact", id }), new Uint8Array(5));
644     await f.rt.artifactFinalize(j, "zip", 5, undefined, "etag");
645     await expect(f.actions.artifactUploadKeys(j, "djt_tok", "zip")).rejects.toThrow(/already exists/);
646   });
647
648   it("recorded before they were listed are moved over, keeping their age", () => {
649     const f = runtimeFixture();
650     f.job("refs/heads/main");
651     const created = new Date(Date.now() - 10 * DAY).toISOString();
652     f.sql.exec("INSERT INTO artifacts (id, run_id, name, size, created_at) VALUES ('x', 'run1', 'old', 7, ?)", created);
653     migrateRepoSchema(f.sql as unknown as SqlStorage);
654     const [a] = f.rt.repoArtifacts({}).artifacts;
655     expect(a).toMatchObject({ name: "old", size: 7, createdAt: created, legacy: true, expired: false });
656     expect(Math.round((Date.parse(a!.expiresAt) - Date.now()) / DAY)).toBe(80);
657     expect(f.sql.exec("SELECT count(*) AS n FROM artifacts").one()).toEqual({ n: 0 });
658   });
659 });
660
661 describe("workflow permission settings", () => {
662   it("audit each change, marking those made through the API", () => {
663     const sql = sqlStorage();
664     sql.exec(REPO_SCHEMA);
665     const events: Array<{ action: string; detail: Record<string, unknown> }> = [];
666     const repo = { sql, meta: { owner: "acme", name: "app" }, emit: (e: { action: string; detail: Record<string, unknown> }) => events.push(e) };
667     const policy = new TokenPolicy(repo as unknown as RepoDO);
668     const admin = { id: "u1", login: "ada", kind: "user", role: "admin" } as Caller;
669     policy.setSettings({ defaultPermissions: "read" }, admin, { via: "api" });
670     // The web settings page calls it without `via`.
671     policy.setSettings({ canApprovePullRequests: true }, admin);
672     expect(events.map((e) => [e.action, e.detail])).toEqual([
673       ["repo.actions_workflow_settings", { defaultPermissions: "read", canApprovePullRequests: false, agentRunSecrets: false, via: "api" }],
674       ["repo.actions_workflow_settings", { defaultPermissions: "read", canApprovePullRequests: true, agentRunSecrets: false }],
675     ]);
676   });
677 });
678
679 describe("an older runner's cache", () => {
680   const DAY = 86400_000;
681   it("drops entries unused for a week, with their objects", async () => {
682     const f = runtimeFixture();
683     const j = f.job("refs/heads/main");
684     expect(await f.actions.cacheSaved(j, "djt_tok", "old", 10)).toEqual([]);
685     expect(await f.actions.cacheSaved(j, "djt_tok", "recent", 10)).toEqual([]);
686     f.sql.exec("UPDATE caches SET last_used = ? WHERE key = ?", new Date(Date.now() - 8 * DAY).toISOString(), "refs/heads/main\nold");
687     f.sql.exec("UPDATE caches SET last_used = ? WHERE key = ?", new Date(Date.now() - 6 * DAY).toISOString(), "refs/heads/main\nrecent");
688     expect(await f.actions.cacheSaved(j, "djt_tok", "new", 10)).toEqual(legacyCacheKeys("r", "refs/heads/main\nold"));
689     expect(f.sql.exec("SELECT key FROM caches ORDER BY key").toArray()).toEqual([{ key: "refs/heads/main\nnew" }, { key: "refs/heads/main\nrecent" }]);
690   });
691 });
692
693 describe("hosted runner slots", () => {
694   it("a repository runs at most SHARE hosted jobs at once; the rest wait their turn", async () => {
695     const sql = sqlStorage();
696     sql.exec(REPO_SCHEMA);
697     migrateRepoSchema(sql as unknown as SqlStorage);
698     const started: string[] = [];
699     const repo = {
700       sql,
701       meta: { repoId: "r", owner: "acme", name: "app", defaultBranch: "main" },
702       upsertCheck: () => {},
703       broadcast: () => {},
704       deploys: { jobCompleted: () => {} },
705       runtime: { sweepJob: async () => {} },
706       alarmBy: () => {},
707     } as unknown as RepoDO;
708     const env = {
709       PUBLIC_URL: "https://d.test",
710       BUCKET: memoryBucket().bucket,
711       RUNNER: { idFromName: (n: string) => n, get: (n: string) => ({ startJob: async () => void started.push(n), stopJob: async () => {} }) },
712       RUN_WORKFLOW: { get: async () => ({ sendEvent: async () => {} }) },
713     } as unknown as Env;
714     const actions = new Actions(repo, env);
715     const t = new Date().toISOString();
716     sql.exec("INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, event_payload) VALUES ('run1', 1, 'w.yml', 'w', 'push', 'refs/heads/main', 'abc', '{}', 'in_progress', ?, ?, '{}')", t, t);
717     for (let i = 0; i < SHARE + 2; i++) sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES (?, 'run1', ?, ?, 'queued')", `j${i}`, `k${i}`, `k${i}`);
718     const launch = (id: string) => (actions as unknown as { launch(id: string): Promise<void> }).launch(id);
719     for (let i = 0; i < SHARE + 2; i++) await launch(`j${i}`);
720     expect(started).toHaveLength(SHARE);
721     const held = () => sql.exec("SELECT id FROM jobs WHERE fleet_held = 1 AND status = 'queued' ORDER BY rowid").toArray().map((r) => r.id);
722     expect(held()).toEqual([`j${SHARE}`, `j${SHARE + 1}`]);
723     // A finished job hands its slot to the oldest one waiting.
724     await (actions as unknown as { completeJob(id: string, c: unknown): Promise<void> }).completeJob("j0", { conclusion: "success", outputs: {} });
725     expect(started).toHaveLength(SHARE + 1);
726     expect(started.at(-1)).toBe(`job-j${SHARE}`);
727     expect(held()).toEqual([`j${SHARE + 1}`]);
728   });
729
730   it("a job whose container gets no machine waits and starts again, and fails only after MAX_PLACEMENT_TRIES", async () => {
731     const sql = sqlStorage();
732     sql.exec(REPO_SCHEMA);
733     migrateRepoSchema(sql as unknown as SqlStorage);
734     let full = true;
735     let startError = "there is no container instance that can be provided to this durable object";
736     const events: { type: string; payload: Record<string, unknown> }[] = [];
737     const repo = {
738       sql,
739       meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" },
740       upsertCheck: () => {},
741       broadcast: () => {},
742       deploys: { jobCompleted: () => {} },
743       runtime: { sweepJob: async () => {} },
744       alarmBy: () => {},
745     } as unknown as RepoDO;
746     const DB = testD1();
747     DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString());
748     const running = () => DB.sqlite.prepare("SELECT id FROM compute_usage WHERE ended_at IS NULL").all();
749     const env = {
750       DB,
751       PUBLIC_URL: "https://d.test",
752       BUCKET: memoryBucket().bucket,
753       RUNNER: {
754         idFromName: (n: string) => n,
755         get: () => ({
756           startJob: async () => {
757             if (full) throw new Error(startError);
758           },
759           stopJob: async () => {},
760         }),
761       },
762       RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) },
763     } as unknown as Env;
764     const actions = new Actions(repo, env);
765     const t = new Date().toISOString();
766     sql.exec("INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, event_payload) VALUES ('run1', 1, 'w.yml', 'w', 'push', 'refs/heads/main', 'abc', '{}', 'in_progress', ?, ?, '{}')", t, t);
767     sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j1', 'run1', 'k', 'k', 'queued'), ('j2', 'run1', 'k2', 'k2', 'queued')");
768     const launch = (id: string) => (actions as unknown as { launch(id: string): Promise<void> }).launch(id);
769     const job = (id: string) => sql.exec("SELECT status, conclusion, container, placement_tries FROM jobs WHERE id = ?", id).toArray()[0];
770
771     // No machine: the job goes back in line and the run's workflow is asked to start it again after a pause.
772     await launch("j1");
773     expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, container: null, placement_tries: 1 });
774     expect(events).toEqual([{ type: "job-j1", payload: { retryInSeconds: placementDelay(1) } }]);
775     // Its log says why it waits.
776     const log = (id: string) => sql.exec<{ text: string }>("SELECT text FROM logs WHERE job_id = ? ORDER BY seq", id).toArray().map((l) => l.text);
777     expect(log("j1")).toEqual([`##[warning]No hosted runner could start the job yet: ${startError}. Diffinery tries again in ${placementDelay(1)} s (wait 1 of ${MAX_PLACEMENT_TRIES}).`]);
778     // While it waits it isn't running, so it holds none of the account's hosted jobs (lib/quotas.ts).
779     expect(running()).toEqual([]);
780
781     // Its container stopping before the runner reported in counts the same way; a started job's loss doesn't.
782     full = false;
783     await actions.relaunchJob("j1");
784     expect(job("j1")).toMatchObject({ status: "waiting", container: "job-j1" });
785     expect(running()).toEqual([{ id: "j1" }]);
786     // The first such stop starts it again after QUICK_RETRY_SECONDS, uncounted; another one waits as usual.
787     await actions.containerStopped("j1", null);
788     expect(job("j1")).toMatchObject({ status: "queued", placement_tries: 1 });
789     expect(events.at(-1)).toEqual({ type: "job-j1", payload: { retryInSeconds: QUICK_RETRY_SECONDS } });
790     expect(log("j1").at(-1)).toBe(`##[warning]No hosted runner could start the job yet: the container stopped before the runner started. Diffinery tries again in ${QUICK_RETRY_SECONDS} s (not counted: the first stop like this is retried at once).`);
791     expect(running()).toEqual([]);
792     await actions.relaunchJob("j1");
793     expect(job("j1")).toMatchObject({ status: "waiting", container: "job-j1" });
794     await actions.containerStopped("j1", 0);
795     expect(job("j1")).toMatchObject({ status: "queued", placement_tries: 2 });
796     expect(events.at(-1)).toEqual({ type: "job-j1", payload: { retryInSeconds: placementDelay(2) } });
797     expect(log("j1").at(-1)).toMatch(/: the container stopped before the runner started\. Diffinery tries again in 60 s \(wait 2 of \d+\)\.$/);
798     expect(running()).toEqual([]);
799     // The same when the runner exits with a code before it fetched the job; and a container that
800     // exits after the job went back in line (its token revoked) leaves the job in line.
801     await actions.relaunchJob("j1");
802     await actions.containerStopped("j1", 2);
803     expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 3 });
804     await actions.containerStopped("j1", 2);
805     expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 3 });
806     await launch("j2");
807     sql.exec("UPDATE jobs SET status = 'in_progress' WHERE id = 'j2'");
808     await actions.containerStopped("j2", 1);
809     expect(job("j2")).toMatchObject({ status: "completed", conclusion: "failure", placement_tries: 0 });
810
811     // A container sidecar restarting under the start counts the same way; an error that won't go away fails the job.
812     full = true;
813     startError = "Container sidecar is shutting down";
814     await actions.relaunchJob("j1");
815     expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 4 });
816     // So does the platform's own internal error starting a container (seen on 2026-10-24).
817     startError = "internal error; reference = 807a2qccbmm6pf5macj9vg76";
818     await actions.relaunchJob("j1");
819     expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 5 });
820     sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j3', 'run1', 'k3', 'k3', 'queued')");
821     startError = "image not found";
822     await launch("j3");
823     expect(job("j3")).toMatchObject({ status: "completed", conclusion: "failure", placement_tries: 0 });
824     expect(sql.exec("SELECT failure FROM jobs WHERE id = 'j3'").one().failure).toMatch(/Could not start a runner: image not found/);
825
826     // Past the limit it fails, and says why.
827     startError = "there is no container instance that can be provided to this durable object";
828     for (let i = 5; i < MAX_PLACEMENT_TRIES; i++) await actions.relaunchJob("j1");
829     expect(job("j1")).toMatchObject({ status: "queued", placement_tries: MAX_PLACEMENT_TRIES });
830     await actions.relaunchJob("j1");
831     expect(job("j1")).toMatchObject({ status: "completed", conclusion: "failure" });
832     expect(sql.exec("SELECT failure FROM jobs WHERE id = 'j1'").one().failure).toMatch(/No hosted runner could be started/);
833     expect([1, 2, 3, 4, 5, 6].map(placementDelay)).toEqual([30, 60, 120, 240, 300, 300]);
834   });
835
836   it("a matrix leg back in line after getting no machine waits out its pause when a sibling finishes", async () => {
837     const sql = sqlStorage();
838     sql.exec(REPO_SCHEMA);
839     migrateRepoSchema(sql as unknown as SqlStorage);
840     let full = true;
841     const started: string[] = [];
842     const events: { type: string; payload: Record<string, unknown> }[] = [];
843     const repo = {
844       sql,
845       meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" },
846       upsertCheck: () => {},
847       broadcast: () => {},
848       deploys: { jobCompleted: () => {} },
849       runtime: { sweepJob: async () => {} },
850       alarmBy: () => {},
851     } as unknown as RepoDO;
852     const DB = testD1();
853     DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString());
854     const env = {
855       DB,
856       PUBLIC_URL: "https://d.test",
857       BUCKET: memoryBucket().bucket,
858       RUNNER: {
859         idFromName: (n: string) => n,
860         get: (n: string) => ({
861           startJob: async () => {
862             started.push(n);
863             if (full) throw new Error("there is no container instance that can be provided to this durable object");
864           },
865           stopJob: async () => {},
866         }),
867       },
868       RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) },
869     } as unknown as Env;
870     const actions = new Actions(repo, env);
871     const a = actions as unknown as { launch(id: string): Promise<void>; completeJob(id: string, c: unknown): Promise<void> };
872     const t = new Date().toISOString();
873     sql.exec("INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, event_payload) VALUES ('run1', 1, 'w.yml', 'w', 'push', 'refs/heads/main', 'abc', '{}', 'in_progress', ?, ?, '{}')", t, t);
874     // Legs of one matrix: m1 and m2 started, m3 not yet (max-parallel).
875     sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('m1', 'run1', 'm', 'm (1)', 'queued'), ('m2', 'run1', 'm', 'm (2)', 'queued'), ('m4', 'run1', 'm', 'm (4)', 'queued'), ('m3', 'run1', 'm', 'm (3)', 'queued')");
876     // m4 is back in line after a safe restart (its runner was replaced before its steps ran).
877     sql.exec("UPDATE jobs SET restarts = 1 WHERE id = 'm4'");
878     const job = (id: string) => sql.exec("SELECT status, placement_tries FROM jobs WHERE id = ?", id).toArray()[0];
879     await a.launch("m1");
880     expect(job("m1")).toMatchObject({ status: "queued", placement_tries: 1 });
881     full = false;
882     await a.launch("m2");
883     expect(job("m2")).toMatchObject({ status: "waiting" });
884     started.length = 0;
885
886     // m2 finishing starts the leg that never started, not the ones back in line: those are started
887     // again by the run's workflow, so each try still waits its turn (placementDelay).
888     await a.completeJob("m2", { conclusion: "success", outputs: {} });
889     expect(started).toEqual(["job-m3"]);
890     expect(job("m1")).toMatchObject({ status: "queued", placement_tries: 1 });
891     expect(job("m4")).toMatchObject({ status: "queued" });
892     expect(events.filter((e) => e.type === "job-m1")).toHaveLength(1);
893   });
894
895   it("a hosted runner on an image other than the deployed one is turned away and the job starts again every STALE_IMAGE_RETRY_SECONDS, up to STALE_IMAGE_TRIES", async () => {
896     const sql = sqlStorage();
897     sql.exec(REPO_SCHEMA);
898     migrateRepoSchema(sql as unknown as SqlStorage);
899     const events: { type: string; payload: Record<string, unknown> }[] = [];
900     const repo = {
901       sql,
902       meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" },
903       upsertCheck: () => {},
904       broadcast: () => {},
905       deploys: { jobCompleted: () => {} },
906       runtime: { sweepJob: async () => {} },
907       alarmBy: (at: number) => void alarms.push(at),
908     } as unknown as RepoDO;
909     const alarms: number[] = [];
910     const DB = testD1();
911     DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString());
912     const env = {
913       DB,
914       PUBLIC_URL: "https://d.test",
915       RUNNER_IMAGE: "diffinery-runner:new",
916       RUNNER: { idFromName: (n: string) => n, get: () => ({ startJob: async () => {}, stopJob: async () => {} }) },
917       RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) },
918     } as unknown as Env;
919     const actions = new Actions(repo, env);
920     const t = new Date().toISOString();
921     sql.exec("INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, event_payload) VALUES ('run1', 1, 'w.yml', 'w', 'push', 'refs/heads/main', 'abc', '{}', 'in_progress', ?, ?, '{}')", t, t);
922     sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j1', 'run1', 'k', 'k', 'queued'), ('s1', 'run1', 's', 's', 'waiting'), ('j2', 'run1', 'k2', 'k2', 'queued')");
923     sql.exec("UPDATE jobs SET labels = '[\"self-hosted\"]' WHERE id = 's1'");
924     const a = actions as unknown as { launch(id: string): Promise<void>; refuseStaleRunner(job: unknown, runner: unknown, deployed?: string): Promise<void>; jobRow(id: string): unknown };
925     const refuse = (id: string, image?: string, deployed?: string) => a.refuseStaleRunner(a.jobRow(id), image ? { image } : null, deployed);
926     const job = (id = "j1") => sql.exec("SELECT status, placement_tries, stale_tries FROM jobs WHERE id = ?", id).toArray()[0];
927     const logs = () => sql.exec<{ text: string }>("SELECT text FROM logs WHERE job_id = 'j1' ORDER BY seq").toArray().map((l) => l.text);
928
929     await a.launch("j1");
930     // The deployed image runs the job.
931     await expect(refuse("j1", "diffinery-runner:new")).resolves.toBeUndefined();
932     expect(job()).toMatchObject({ status: "waiting", placement_tries: 0 });
933     // The previous image (or one that doesn't say) is turned away. The job starts again after a short fixed
934     // wait, from this object's alarm rather than the run's workflow, and the waits don't count as placement tries.
935     for (let i = 0; i < STALE_IMAGE_TRIES; i++) {
936       const before = Date.now();
937       await expect(refuse("j1", i === 1 ? undefined : "diffinery-runner:old")).rejects.toMatchObject({ status: 409 });
938       expect(job()).toMatchObject({ status: "queued", placement_tries: 0, stale_tries: i + 1 });
939       const at = actions.nextImageRetry()!;
940       expect(at).toBeGreaterThanOrEqual(before + STALE_IMAGE_RETRY_SECONDS * 1000);
941       expect(alarms.at(-1)).toBe(at);
942       // Not before its time.
943       await actions.relaunchImageHeld(at - 1);
944       expect(job()).toMatchObject({ status: "queued" });
945       await actions.relaunchImageHeld(at);
946       expect(job()).toMatchObject({ status: "waiting" });
947       expect(actions.nextImageRetry()).toBeNull();
948     }
949     expect(events).toEqual([]);
950     expect(logs()[0]).toBe(`##[warning]No hosted runner could start the job yet: a runner started on diffinery-runner:old while the deploy of diffinery-runner:new was rolling out. Diffinery tries again in ${STALE_IMAGE_RETRY_SECONDS} s (rollout wait 1 of ${STALE_IMAGE_TRIES}, not counted).`);
951     // A rollout that doesn't finish doesn't hold the job up past STALE_IMAGE_TRIES.
952     await expect(refuse("j1", "diffinery-runner:old")).resolves.toBeUndefined();
953     // A job cancelled while it waits isn't started again.
954     sql.exec("UPDATE jobs SET status = 'queued', image_retry_at = 1 WHERE id = 'j1'");
955     sql.exec("UPDATE jobs SET status = 'completed' WHERE id = 'j1'");
956     await actions.relaunchImageHeld();
957     expect(job()).toMatchObject({ status: "completed" });
958     // Self-hosted runners run any image.
959     await expect(refuse("s1", "my-image")).resolves.toBeUndefined();
960
961     // Right after a deploy this object can still run the previous version (its RUNNER_IMAGE is the
962     // old one) while the Worker that took the runner's request has the new one: the Worker's wins.
963     await a.launch("j2");
964     await expect(refuse("j2", "diffinery-runner:next", "diffinery-runner:next")).resolves.toBeUndefined();
965     expect(job("j2")).toMatchObject({ status: "waiting", placement_tries: 0 });
966     await expect(refuse("j2", "diffinery-runner:new", "diffinery-runner:next")).rejects.toMatchObject({ status: 409 });
967     expect(job("j2")).toMatchObject({ status: "queued", stale_tries: 1 });
968   });
969
970   it("a running job whose runner froze is ended at its timeout plus JOB_DEADLINE_GRACE_MS, and a healthy one is left alone", async () => {
971     const sql = sqlStorage();
972     sql.exec(REPO_SCHEMA);
973     migrateRepoSchema(sql as unknown as SqlStorage);
974     const events: { type: string; payload: Record<string, unknown> }[] = [];
975     const stopped: string[] = [];
976     const repo = {
977       sql,
978       meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" },
979       upsertCheck: () => {},
980       broadcast: () => {},
981       deploys: { jobCompleted: () => {} },
982       runtime: { sweepJob: async () => {} },
983       alarmBy: () => {},
984     } as unknown as RepoDO;
985     const env = {
986       PUBLIC_URL: "https://d.test",
987       BUCKET: memoryBucket().bucket,
988       RUNNER: { idFromName: (n: string) => n, get: (n: string) => ({ startJob: async () => {}, stopJob: async () => void stopped.push(n) }) },
989       RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) },
990     } as unknown as Env;
991     const actions = new Actions(repo, env);
992     const t0 = Date.parse("2026-10-16T12:00:00Z");
993     const at = (ms: number) => new Date(t0 + ms).toISOString();
994     sql.exec("INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, event_payload) VALUES ('run1', 1, 'w.yml', 'w', 'push', 'refs/heads/main', 'abc', '{}', 'in_progress', ?, ?, '{}')", at(0), at(0));
995     // Each runner fetched its job at t0 (started_at). f: frozen, timeout 10 min; h: healthy, timeout 30;
996     // d: no timeout of its own (6 hours); s: on a self-hosted runner, frozen too; q: still in line; c: finished.
997     const add = (id: string, minutes: number | null, o: { status?: string; container?: string | null; started?: string | null } = {}) =>
998       sql.exec(
999         "INSERT INTO jobs (id, run_id, job_key, name, status, timeout_minutes, container, started_at, strategy) VALUES (?, 'run1', ?, ?, ?, ?, ?, ?, '{\"fail-fast\":false}')",
1000         id, id, id, o.status ?? "in_progress", minutes, o.container === undefined ? `job-${id}` : o.container, o.started === undefined ? at(0) : o.started,
1001       );
1002     add("f", 10);
1003     add("h", 30);
1004     add("d", 0);
1005     add("s", 10, { container: "self-hosted:box" });
1006     add("q", 10, { status: "queued", container: null, started: null });
1007     add("c", 10, { status: "completed" });
1008     const job = (id: string) => sql.exec<{ status: string; conclusion: string | null; failure: string | null }>("SELECT status, conclusion, failure FROM jobs WHERE id = ?", id).one();
1009     const min = 60_000;
1010     expect(actions.nextJobDeadline()).toBe(t0 + 10 * min + JOB_DEADLINE_GRACE_MS);
1011
1012     // Past f's timeout but within the grace its runner has to stop the steps and report: nothing happens.
1013     await actions.enforceJobDeadlines(t0 + 10 * min + JOB_DEADLINE_GRACE_MS - 1);
1014     expect(["f", "h", "d", "s"].map((id) => job(id).status)).toEqual(["in_progress", "in_progress", "in_progress", "in_progress"]);
1015     expect(stopped).toEqual([]);
1016
1017     // At the deadline the frozen ones end: the hosted container is stopped, both fail as timed out.
1018     await actions.enforceJobDeadlines(t0 + 10 * min + JOB_DEADLINE_GRACE_MS);
1019     for (const id of ["f", "s"]) {
1020       expect(job(id)).toMatchObject({ status: "completed", conclusion: "timed_out" });
1021       expect(JSON.parse(job(id).failure!).message).toBe(`The job exceeded its timeout of 10 minutes, and its runner stopped responding: Diffinery ended it ${JOB_DEADLINE_GRACE_MS / min} minutes later.`);
1022     }
1023     expect(stopped).toEqual(["job-f"]);
1024     expect(events.filter((e) => e.type === "job-f")).toEqual([{ type: "job-f", payload: { conclusion: "timed_out" } }]);
1025     // The others are untouched, and the next wake-up is h's deadline.
1026     expect([job("h").status, job("d").status, job("q").status, job("c").status]).toEqual(["in_progress", "in_progress", "queued", "completed"]);
1027     expect(actions.nextJobDeadline()).toBe(t0 + 30 * min + JOB_DEADLINE_GRACE_MS);
1028     // A job without timeout-minutes gets GitHub's 6 hours.
1029     await actions.enforceJobDeadlines(t0 + 360 * min + JOB_DEADLINE_GRACE_MS - 1);
1030     expect(job("d").status).toBe("in_progress");
1031     await actions.enforceJobDeadlines(t0 + 360 * min + JOB_DEADLINE_GRACE_MS);
1032     expect(job("d")).toMatchObject({ status: "completed", conclusion: "timed_out" });
1033     expect(actions.nextJobDeadline()).toBeNull();
1034   });
1035
1036   it("cancelling a run lets running hosted jobs run their steps that run after a cancel, and ends them after CANCEL_STOP_MS", async () => {
1037     const sql = sqlStorage();
1038     sql.exec(REPO_SCHEMA);
1039     migrateRepoSchema(sql as unknown as SqlStorage);
1040     const events: { type: string; payload: Record<string, unknown> }[] = [];
1041     const stopped: [string, boolean][] = [];
1042     const alarms: number[] = [];
1043     const repo = {
1044       sql,
1045       meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" },
1046       upsertCheck: () => {},
1047       broadcast: () => {},
1048       deploys: { jobCompleted: () => {} },
1049       runtime: { sweepJob: async () => {}, grantFor: () => ({ role: "read", permissions: {} }) },
1050       alarmBy: (at: number) => void alarms.push(at),
1051     } as unknown as RepoDO;
1052     const env = {
1053       PUBLIC_URL: "https://d.test",
1054       BUCKET: memoryBucket().bucket,
1055       RUNNER: { idFromName: (n: string) => n, get: (n: string) => ({ startJob: async () => {}, stopJob: async (graceful = false) => void stopped.push([n, graceful]) }) },
1056       RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) },
1057     } as unknown as Env;
1058     const actions = new Actions(repo, env);
1059     const t = new Date().toISOString();
1060     sql.exec("INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, event_payload) VALUES ('run1', 1, 'w.yml', 'w', 'push', 'refs/heads/main', 'abc', '{}', 'in_progress', ?, ?, '{}')", t, t);
1061     // a and b run on hosted runners (a reports after its cleanup steps, b's runner never does); s runs
1062     // on a self-hosted runner; w's container hasn't fetched the job yet; q waits in line.
1063     const token = (id: string) => `djt_r_${id}_secret`;
1064     const add = async (id: string, status: string, container: string | null) =>
1065       sql.exec(
1066         "INSERT INTO jobs (id, run_id, job_key, name, status, timeout_minutes, container, started_at, token_hash, token_at, strategy) VALUES (?, 'run1', ?, ?, ?, 60, ?, ?, ?, ?, '{\"fail-fast\":false}')",
1067         id, id, id, status, container, status === "in_progress" ? t : null, await sha256(token(id)), t,
1068       );
1069     await add("a", "in_progress", "job-a");
1070     await add("b", "in_progress", "job-b");
1071     await add("s", "in_progress", "self-hosted:box");
1072     await add("w", "waiting", "job-w");
1073     await add("q", "queued", null);
1074     const job = (id: string) => sql.exec<{ status: string; conclusion: string | null; failure: string | null }>("SELECT status, conclusion, failure FROM jobs WHERE id = ?", id).one();
1075     const before = Date.now();
1076     await actions.cancel("run1", { kind: "user", login: "octo", role: "write" } as unknown as Caller);
1077     // a and b get SIGINT and keep running; the rest end at once, w's container stopped outright.
1078     expect(stopped).toEqual([["job-a", true], ["job-b", true], ["job-w", false]]);
1079     expect([job("a").status, job("b").status]).toEqual(["in_progress", "in_progress"]);
1080     for (const id of ["s", "w", "q"]) expect(job(id)).toMatchObject({ status: "completed", conclusion: "cancelled", failure: JSON.stringify({ message: "Cancelled by octo" }) });
1081     expect(alarms.at(-1)).toBeGreaterThanOrEqual(before + CANCEL_STOP_MS);
1082     const deadline = actions.nextJobDeadline()!;
1083     expect(deadline - before).toBeGreaterThanOrEqual(CANCEL_STOP_MS);
1084     expect(deadline - Date.now()).toBeLessThanOrEqual(CANCEL_STOP_MS);
1085     // a's runner reports once its cleanup steps ran: cancelled, saying who cancelled it, never started again.
1086     await actions.runnerComplete("a", token("a"), { conclusion: "cancelled", outputs: {}, failure: { message: "The job was cancelled." }, replaced: true });
1087     expect(job("a")).toMatchObject({ status: "completed", conclusion: "cancelled", failure: JSON.stringify({ message: "Cancelled by octo" }) });
1088
1089     // b's runner never reports: at CANCEL_STOP_MS its container is stopped and the job ends.
1090     await actions.enforceJobDeadlines(deadline - 1);
1091     expect(job("b").status).toBe("in_progress");
1092     await actions.enforceJobDeadlines(deadline);
1093     expect(stopped.at(-1)).toEqual(["job-b", false]);
1094     expect(job("b").conclusion).toBe("cancelled");
1095     expect(JSON.parse(job("b").failure!).message).toBe(`Cancelled by octo. Its runner didn't report within ${CANCEL_STOP_MS / 60_000} minutes of the cancel, so Diffinery stopped it.`);
1096     expect(actions.nextJobDeadline()).toBeNull();
1097   });
1098
1099   it("cancelling a cancelled run again stops its jobs at once, and a cancelled job whose container stops before its runner reports ends as cancelled", async () => {
1100     const sql = sqlStorage();
1101     sql.exec(REPO_SCHEMA);
1102     migrateRepoSchema(sql as unknown as SqlStorage);
1103     const repo = {
1104       sql,
1105       meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" },
1106       upsertCheck: () => {},
1107       broadcast: () => {},
1108       deploys: { jobCompleted: () => {} },
1109       runtime: { sweepJob: async () => {} },
1110       alarmBy: () => {},
1111     } as unknown as RepoDO;
1112     const env = {
1113       PUBLIC_URL: "https://d.test",
1114       BUCKET: memoryBucket().bucket,
1115       RUNNER_IMAGE: "diffinery-runner:new",
1116       RUNNER: { idFromName: (n: string) => n, get: (n: string) => ({ startJob: async () => {}, stopJob: async (graceful = false) => void stopped.push([n, graceful]) }) },
1117       RUN_WORKFLOW: { get: async () => ({ sendEvent: async () => {} }) },
1118     } as unknown as Env;
1119     const stopped: [string, boolean][] = [];
1120     const actions = new Actions(repo, env);
1121     const t = new Date().toISOString();
1122     const job = (id: string) => sql.exec("SELECT status, conclusion, failure FROM jobs WHERE id = ?", id).one();
1123     for (const run of ["run1", "run2"]) sql.exec("INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, event_payload) VALUES (?, 1, 'w.yml', 'w', 'push', 'refs/heads/main', 'abc', '{}', 'in_progress', ?, ?, '{}')", run, t, t);
1124     // On an older runner image than the deployed one, which would otherwise read as a rollout and start the job again.
1125     for (const [id, run] of [["j1", "run1"], ["j2", "run2"]]) sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status, container, started_at, runner_info, strategy) VALUES (?, ?, ?, ?, 'in_progress', ?, ?, '{\"image\":\"diffinery-runner:old\"}', '{}')", id, run, id, id, `job-${id}`, t);
1126     const octo = { kind: "user", login: "octo", role: "write" } as unknown as Caller;
1127     await actions.cancel("run1", octo);
1128     await actions.cancel("run1", { kind: "user", login: "hubot", role: "write" } as unknown as Caller);
1129     expect(stopped).toEqual([["job-j1", true], ["job-j1", false]]);
1130     expect(job("j1")).toMatchObject({ status: "completed", conclusion: "cancelled", failure: JSON.stringify({ message: "Cancelled by octo" }) });
1131     await actions.cancel("run2", octo);
1132     await actions.containerStopped("j2", null, "job-j2", true);
1133     expect(job("j2")).toMatchObject({ status: "completed", conclusion: "cancelled", failure: JSON.stringify({ message: "Cancelled by octo" }) });
1134   });
1135
1136   it("a job whose runner a rollout replaced starts again on a new runner before its steps ran or when it can only read, and fails otherwise", async () => {
1137     const sql = sqlStorage();
1138     sql.exec(REPO_SCHEMA);
1139     migrateRepoSchema(sql as unknown as SqlStorage);
1140     const events: { type: string; payload: Record<string, unknown> }[] = [];
1141     const swept: [string, boolean][] = [];
1142     const repo = {
1143       sql,
1144       meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" },
1145       upsertCheck: () => {},
1146       broadcast: () => {},
1147       deploys: { jobCompleted: () => {} },
1148       runtime: { sweepJob: async (id: string, o?: { restart?: boolean }) => void swept.push([id, !!o?.restart]), grantFor: () => ({ role: "read", permissions: {} }) },
1149       policy: { settings: () => ({ defaultPermissions: "write" }), isForkPull: () => false },
1150       alarmBy: () => {},
1151     } as unknown as RepoDO;
1152     const DB = testD1();
1153     DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString());
1154     const started: { container: string; token: string }[] = [];
1155     const env = {
1156       DB,
1157       PUBLIC_URL: "https://d.test",
1158       BUCKET: memoryBucket().bucket,
1159       RUNNER_IMAGE: "diffinery-runner:new",
1160       RUNNER: { idFromName: (n: string) => n, get: () => ({ startJob: async (i: { container: string; token: string }) => void started.push(i), stopJob: async () => {} }) },
1161       RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) },
1162     } as unknown as Env;
1163     const actions = new Actions(repo, env);
1164     // The job's steps: checkout, then one whose effects running it twice would repeat.
1165     // k to k4 get the repository's default token (read and write); the others only read (r1 hands its read-only token to `gh`), unless they name a secret, deploy to an environment or ask for an OIDC token.
1166     (actions as unknown as { expanded: () => Promise<unknown> }).expanded = async () => {
1167       const steps = [{ uses: "actions/checkout@v4" }, { run: "make deploy" }];
1168       const read = { contents: "read" };
1169       const install = [{ uses: "actions/checkout@v4" }, { run: "pnpm install" }, { run: "pnpm test" }];
1170       const jobs: Record<string, unknown> = {
1171         k: { steps },
1172         k2: { steps },
1173         k3: { steps },
1174         k4: { steps },
1175         r1: { steps: install, permissions: read, env: { GH_TOKEN: "${{ secrets.GITHUB_TOKEN }}" } },
1176         r2: { steps: [...install, { run: "npm publish", env: { NODE_AUTH_TOKEN: "${{ secrets.NPM_TOKEN }}" } }], permissions: read },
1177         r3: { steps: install, permissions: read, environment: { name: "production" } },
1178         r4: { steps: install, permissions: { contents: "read", "id-token": "write" } },
1179         r5: { steps: [...install, { run: "echo ${{ toJSON(secrets) }}" }], permissions: read },
1180       };
1181       const workflow = { env: {}, jobs };
1182       return { workflow, calls: new Map(), parentOf: new Map(), needs: new Map(), workflowOf: new Map(Object.keys(jobs).map((key) => [key, workflow])) };
1183     };
1184     const t = new Date().toISOString();
1185     sql.exec("INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, event_payload) VALUES ('run1', 1, 'w.yml', 'w', 'push', 'refs/heads/main', 'abc', '{}', 'in_progress', ?, ?, '{}')", t, t);
1186     sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j1', 'run1', 'k', 'k', 'queued'), ('j2', 'run1', 'k2', 'k2', 'queued')");
1187     const launch = (id: string) => (actions as unknown as { launch(id: string): Promise<void> }).launch(id);
1188     const job = (id: string) => sql.exec("SELECT status, conclusion, container, token_hash, steps, restarts, failure FROM jobs WHERE id = ?", id).toArray()[0];
1189     const logs = (id: string) => sql.exec("SELECT seq, step, text FROM logs WHERE job_id = ? ORDER BY seq", id).toArray();
1190     // The runner fetched the job on `image` and reported these steps as started.
1191     const running = (id: string, image: string, steps: number[], names = ["Run actions/checkout@v4", "Deploy"]) => {
1192       const reports = steps.map((index) => ({ index, name: names[index] ?? `Step ${index}`, status: "in_progress", startedAt: t }));
1193       sql.exec("UPDATE jobs SET status = 'in_progress', started_at = ?, runner_info = ?, steps = ? WHERE id = ?", t, JSON.stringify({ image }), JSON.stringify(reports), id);
1194       sql.exec("INSERT INTO logs (job_id, seq, t, step, text) VALUES (?, 0, ?, -1, 'set up'), (?, 1, ?, 0, 'checking out')", id, t, id, t);
1195     };
1196     const replaced = { conclusion: "cancelled" as const, outputs: {}, replaced: true };
1197
1198     // The runner reports the replacement during checkout: the job goes back in line, its log starts over with a line that says why.
1199     await launch("j1");
1200     const first = started.at(-1)!;
1201     expect(first.container).toBe("job-j1");
1202     running("j1", "diffinery-runner:old", [0]);
1203     await actions.runnerComplete("j1", first.token, replaced);
1204     expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, container: null, token_hash: null, steps: "[]", restarts: 1 });
1205     expect(events.at(-1)).toEqual({ type: "job-j1", payload: { retryInSeconds: 10 } });
1206     expect(logs("j1")).toEqual([{ seq: 2, step: -1, text: expect.stringMatching(/replaced this job's runner.*started the job again from the beginning on a new runner \(restart 1 of 2\)/) }]);
1207     // The replaced runner can't report any more, and its container stopping changes nothing.
1208     await expect(actions.runnerComplete("j1", first.token, replaced)).rejects.toMatchObject({ status: 401 });
1209     await actions.containerStopped("j1", 143, "job-j1");
1210     expect(job("j1")).toMatchObject({ status: "queued", restarts: 1 });
1211
1212     // It starts in a new container; the old one stopping late doesn't touch it.
1213     await actions.relaunchJob("j1");
1214     expect(started.at(-1)!.container).toBe("job-j1-r1");
1215     await actions.containerStopped("j1", 143, "job-j1");
1216     expect(job("j1")).toMatchObject({ status: "waiting", container: "job-j1-r1" });
1217
1218     // A container on a replaced image that stops without reporting counts the same.
1219     running("j1", "diffinery-runner:old", []);
1220     await actions.containerStopped("j1", 143, "job-j1-r1");
1221     expect(job("j1")).toMatchObject({ status: "queued", restarts: 2 });
1222
1223     // Past MAX_RUNNER_RESTARTS the job fails, and says why.
1224     await actions.relaunchJob("j1");
1225     expect(started.at(-1)!.container).toBe("job-j1-r2");
1226     running("j1", "diffinery-runner:old", []);
1227     await actions.runnerComplete("j1", started.at(-1)!.token, replaced);
1228     expect(job("j1")).toMatchObject({ status: "completed", conclusion: "failure", restarts: MAX_RUNNER_RESTARTS });
1229     expect(JSON.parse(job("j1")!.failure as string).message).toMatch(/replaced this job's runner with a new version before the job's steps ran, 3 times\. Re-run the job\./);
1230
1231     // Once a step other than checkout ran, the job isn't started again: it fails and names the step.
1232     await launch("j2");
1233     running("j2", "diffinery-runner:old", [0, 1]);
1234     await actions.runnerComplete("j2", started.at(-1)!.token, replaced);
1235     expect(job("j2")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 });
1236     expect(JSON.parse(job("j2")!.failure as string).message).toMatch(/while step "Deploy" ran\. Diffinery doesn't start a job again by itself once its steps have run if it could change something outside the run/);
1237
1238     // A job that can only read starts again even after its steps ran (here the platform stopped
1239     // `pnpm install`), and what its stopped attempt uploaded goes. Each restart is counted.
1240     sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('r1', 'run1', 'r1', 'r1', 'queued'), ('r2', 'run1', 'r2', 'r2', 'queued'), ('r3', 'run1', 'r3', 'r3', 'queued'), ('r4', 'run1', 'r4', 'r4', 'queued'), ('r5', 'run1', 'r5', 'r5', 'queued')");
1241     const install = ["Run actions/checkout@v4", "Run pnpm install", "Run pnpm test"];
1242     await launch("r1");
1243     running("r1", "diffinery-runner:new", [0, 1], install);
1244     await actions.runnerComplete("r1", started.at(-1)!.token, replaced);
1245     expect(job("r1")).toMatchObject({ status: "queued", conclusion: null, steps: "[]", restarts: 1 });
1246     expect(swept.at(-1)).toEqual(["r1", true]);
1247     expect(logs("r1")).toEqual([{ seq: 2, step: -1, text: expect.stringMatching(/while step "Run pnpm install" ran\. The job can only read .* Diffinery started the job again from the beginning on a new runner \(restart 1 of 2\)/) }]);
1248     // It still stops after MAX_RUNNER_RESTARTS, wherever the platform stopped it.
1249     for (let i = 1; i <= MAX_RUNNER_RESTARTS; i++) {
1250       await actions.relaunchJob("r1");
1251       running("r1", "diffinery-runner:new", [0, 1, 2], install);
1252       await actions.runnerComplete("r1", started.at(-1)!.token, replaced);
1253     }
1254     expect(job("r1")).toMatchObject({ status: "completed", conclusion: "failure", restarts: MAX_RUNNER_RESTARTS });
1255     // (On the deployed image no deploy replaced it, so the message says the platform stopped it.)
1256     expect(JSON.parse(job("r1")!.failure as string).message).toBe("The platform stopped this job's container after its steps had started, 3 times. Re-run the job.");
1257     // One that names a secret, deploys to an environment, can get an OIDC token, or reads every secret fails as before, naming the step that was stopped.
1258     for (const id of ["r2", "r3", "r4", "r5"]) {
1259       await launch(id);
1260       running(id, "diffinery-runner:new", [0, 1, 2], install);
1261       await actions.runnerComplete(id, started.at(-1)!.token, replaced);
1262       expect(job(id)).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 });
1263       expect(JSON.parse(job(id)!.failure as string).message).toMatch(/while step "Run pnpm test" ran\. Diffinery doesn't start a job again by itself/);
1264     }
1265     // Right after a deploy this object can still have the previous RUNNER_IMAGE: the image of the
1266     // Worker that took the report decides whether a deploy replaced the runner.
1267     sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('r6', 'run1', 'r1', 'r6', 'queued'), ('r7', 'run1', 'r1', 'r7', 'queued')");
1268     for (const [id, deployed, says] of [["r6", "diffinery-runner:next", "The platform stopped this job's container"], ["r7", undefined, "The platform replaced this job's runner with a new version"]] as const) {
1269       await launch(id);
1270       running(id, "diffinery-runner:next", [0, 1], install);
1271       await actions.runnerComplete(id, started.at(-1)!.token, replaced, deployed);
1272       expect(job(id)).toMatchObject({ status: "queued", restarts: 1 });
1273       expect(logs(id)[0]!.text).toMatch(new RegExp(`^##\\[warning\\]${says} while step "Run pnpm install" ran`));
1274     }
1275
1276     // A job that finished despite the replacement, or a runner on the deployed image that stops, completes as before.
1277     sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j3', 'run1', 'k3', 'k3', 'queued'), ('j4', 'run1', 'k4', 'k4', 'queued')");
1278     await launch("j3");
1279     running("j3", "diffinery-runner:old", [0, 1]);
1280     await actions.runnerComplete("j3", started.at(-1)!.token, { conclusion: "success", outputs: {}, replaced: true });
1281     expect(job("j3")).toMatchObject({ status: "completed", conclusion: "success" });
1282     await launch("j4");
1283     running("j4", "diffinery-runner:new", []);
1284     await actions.containerStopped("j4", 137, "job-j4");
1285     expect(job("j4")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 });
1286     expect(JSON.parse(job("j4")!.failure as string).message).toBe("The runner stopped unexpectedly (exit code 137).");
1287
1288     // A container on the deployed image that the platform stops or loses mid-job (no exit code, not
1289     // stopped by Diffinery) counts as a replacement too: a job that can only read starts again...
1290     sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('s1', 'run1', 'r1', 's1', 'queued'), ('s2', 'run1', 'k3', 's2', 'queued')");
1291     await launch("s1");
1292     running("s1", "diffinery-runner:new", [0, 1], install);
1293     await actions.containerStopped("s1", 0, "job-s1");
1294     expect(job("s1")).toMatchObject({ status: "queued", conclusion: null, restarts: 1 });
1295     expect(swept.at(-1)).toEqual(["s1", true]);
1296     expect(logs("s1")).toEqual([{ seq: 2, step: -1, text: expect.stringMatching(/^##\[warning\]The platform stopped this job's container while step "Run pnpm install" ran\. The job can only read .* \(restart 1 of 2\)\.$/) }]);
1297     // ...a late stop of that container changes nothing, and the restarts stop at MAX_RUNNER_RESTARTS.
1298     await actions.containerStopped("s1", null, "job-s1");
1299     expect(job("s1")).toMatchObject({ status: "queued", restarts: 1 });
1300     for (let i = 1; i <= MAX_RUNNER_RESTARTS; i++) {
1301       await actions.relaunchJob("s1");
1302       running("s1", "diffinery-runner:new", [0, 1], install);
1303       await actions.containerStopped("s1", null, started.at(-1)!.container);
1304     }
1305     expect(job("s1")).toMatchObject({ status: "completed", conclusion: "failure", restarts: MAX_RUNNER_RESTARTS });
1306     expect(JSON.parse(job("s1")!.failure as string).message).toBe("The platform stopped this job's container after its steps had started, 3 times. Re-run the job.");
1307     // One that could change something outside the run fails, naming the step.
1308     await launch("s2");
1309     running("s2", "diffinery-runner:new", [0, 1]);
1310     await actions.containerStopped("s2", 0, "job-s2");
1311     expect(job("s2")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 });
1312     expect(JSON.parse(job("s2")!.failure as string).message).toMatch(/^The platform stopped this job's container while step "Deploy" ran\. Diffinery doesn't start a job again by itself/);
1313     // Diffinery's own stop (a cancel, a timeout, fail-fast) never starts a job again, and a runner that crashes isn't retried.
1314     // (Each added after the one before failed, so fail-fast doesn't cancel it.)
1315     sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('s3', 'run1', 'r1', 's3', 'queued')");
1316     await launch("s3");
1317     running("s3", "diffinery-runner:new", [0, 1], install);
1318     await actions.containerStopped("s3", 0, "job-s3", true);
1319     expect(job("s3")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 });
1320     sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('s4', 'run1', 'r1', 's4', 'queued')");
1321     await launch("s4");
1322     running("s4", "diffinery-runner:new", [0, 1], install);
1323     await actions.containerStopped("s4", 1, "job-s4");
1324     expect(job("s4")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 });
1325   });
1326 });
1327
1328 describe("runner info", () => {
1329   it("keeps only the runner's short image, location and instance strings", () => {
1330     expect(parseRunnerInfo(JSON.stringify({ image: "diffinery-runner:abc", location: "fra06, EEUR", instance: "d0", cpu: "AMD EPYC (family 25, model 17), 4 vCPUs, 2550 MHz", token: "x", n: 1 }))).toEqual({ image: "diffinery-runner:abc", location: "fra06, EEUR", instance: "d0", cpu: "AMD EPYC (family 25, model 17), 4 vCPUs, 2550 MHz" });
1331     expect(parseRunnerInfo(JSON.stringify({ image: "x".repeat(300) }))?.image).toHaveLength(200);
1332     for (const bad of [null, "", "not json", "{}", JSON.stringify({ image: 5 }), "x".repeat(2000)]) expect(parseRunnerInfo(bad)).toBeNull();
1333   });
1334 });
1335
1336 describe("unfinished runs", () => {
1337   it("past MAX_ACTIVE_RUNS a dispatch is refused and other events start nothing", async () => {
1338     const sql = sqlStorage();
1339     sql.exec(REPO_SCHEMA);
1340     migrateRepoSchema(sql as unknown as SqlStorage);
1341     const actions = new Actions({ sql, meta: { repoId: "r", owner: "acme", name: "app", defaultBranch: "main" } } as unknown as RepoDO, { PUBLIC_URL: "https://d.test" } as unknown as Env);
1342     const t = new Date().toISOString();
1343     for (let i = 0; i < MAX_ACTIVE_RUNS; i++) sql.exec("INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, event_payload) VALUES (?, ?, 'w.yml', 'w', 'push', 'refs/heads/main', 'abc', '{}', 'queued', ?, ?, '{}')", `run${i}`, i + 1, t, t);
1344     const wf = { path: ".github/workflows/w.yml", on: {}, jobs: {} } as never;
1345     const caller = { id: "u1", login: "ada", kind: "user" as const };
1346     await expect(actions.createRun(wf, { event: "workflow_dispatch", ref: "refs/heads/main", sha: "abc", payload: {}, caller, title: "w" })).rejects.toThrow(/500 unfinished workflow runs/);
1347     expect(await actions.createRun(wf, { event: "push", ref: "refs/heads/main", sha: "abc", payload: {}, caller, title: "w" })).toBeNull();
1348   });
1349 });