| 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&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 | // Each placement after a runner that never started is in a new container object. |
| 785 | expect(job("j1")).toMatchObject({ status: "waiting", container: "job-j1-p1" }); |
| 786 | expect(running()).toEqual([{ id: "j1" }]); |
| 787 | // The first such stop starts it again after QUICK_RETRY_SECONDS, uncounted; another one waits as usual. |
| 788 | await actions.containerStopped("j1", null); |
| 789 | expect(job("j1")).toMatchObject({ status: "queued", placement_tries: 1 }); |
| 790 | expect(events.at(-1)).toEqual({ type: "job-j1", payload: { retryInSeconds: QUICK_RETRY_SECONDS } }); |
| 791 | 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 failure like this is retried at once).`); |
| 792 | expect(running()).toEqual([]); |
| 793 | await actions.relaunchJob("j1"); |
| 794 | expect(job("j1")).toMatchObject({ status: "waiting", container: "job-j1-p2" }); |
| 795 | await actions.containerStopped("j1", 0); |
| 796 | expect(job("j1")).toMatchObject({ status: "queued", placement_tries: 2 }); |
| 797 | expect(events.at(-1)).toEqual({ type: "job-j1", payload: { retryInSeconds: placementDelay(2) } }); |
| 798 | expect(log("j1").at(-1)).toMatch(/: the container stopped before the runner started\. Diffinery tries again in 60 s \(wait 2 of \d+\)\.$/); |
| 799 | expect(running()).toEqual([]); |
| 800 | // The same when the runner exits with a code before it fetched the job; and a container that |
| 801 | // exits after the job went back in line (its token revoked) leaves the job in line. |
| 802 | await actions.relaunchJob("j1"); |
| 803 | await actions.containerStopped("j1", 2); |
| 804 | expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 3 }); |
| 805 | await actions.containerStopped("j1", 2); |
| 806 | expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 3 }); |
| 807 | await launch("j2"); |
| 808 | sql.exec("UPDATE jobs SET status = 'in_progress' WHERE id = 'j2'"); |
| 809 | await actions.containerStopped("j2", 1); |
| 810 | expect(job("j2")).toMatchObject({ status: "completed", conclusion: "failure", placement_tries: 0 }); |
| 811 | |
| 812 | // A container sidecar restarting under the start counts the same way; an error that won't go away fails the job. |
| 813 | full = true; |
| 814 | startError = "Container sidecar is shutting down"; |
| 815 | await actions.relaunchJob("j1"); |
| 816 | expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 4 }); |
| 817 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j3', 'run1', 'k3', 'k3', 'queued')"); |
| 818 | startError = "image not found"; |
| 819 | await launch("j3"); |
| 820 | expect(job("j3")).toMatchObject({ status: "completed", conclusion: "failure", placement_tries: 0 }); |
| 821 | expect(sql.exec("SELECT failure FROM jobs WHERE id = 'j3'").one().failure).toMatch(/Could not start a runner: image not found/); |
| 822 | |
| 823 | // Past the limit it fails, and says why. |
| 824 | startError = "there is no container instance that can be provided to this durable object"; |
| 825 | for (let i = 4; i < MAX_PLACEMENT_TRIES; i++) await actions.relaunchJob("j1"); |
| 826 | expect(job("j1")).toMatchObject({ status: "queued", placement_tries: MAX_PLACEMENT_TRIES }); |
| 827 | await actions.relaunchJob("j1"); |
| 828 | expect(job("j1")).toMatchObject({ status: "completed", conclusion: "failure" }); |
| 829 | expect(sql.exec("SELECT failure FROM jobs WHERE id = 'j1'").one().failure).toMatch(/No hosted runner could be started/); |
| 830 | expect([1, 2, 3, 4, 5, 6].map(placementDelay)).toEqual([30, 60, 120, 240, 300, 300]); |
| 831 | }); |
| 832 | |
| 833 | it("a start the runtime fails for a passing reason is retried, and never once the job's runner fetched it", async () => { |
| 834 | const sql = sqlStorage(); |
| 835 | sql.exec(REPO_SCHEMA); |
| 836 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 837 | let fail: (() => Error) | null = null; |
| 838 | let token = ""; |
| 839 | let workflowDown = false; |
| 840 | const alarms: number[] = []; |
| 841 | const events: { type: string; payload: Record<string, unknown> }[] = []; |
| 842 | const repo = { |
| 843 | sql, |
| 844 | meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" }, |
| 845 | upsertCheck: () => {}, |
| 846 | broadcast: () => {}, |
| 847 | deploys: { jobCompleted: () => {} }, |
| 848 | runtime: { sweepJob: async () => {}, grantFor: () => ({ role: "write", permissions: {} }) }, |
| 849 | alarmBy: (at: number) => void alarms.push(at), |
| 850 | } as unknown as RepoDO; |
| 851 | const DB = testD1(); |
| 852 | DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString()); |
| 853 | const env = { |
| 854 | DB, |
| 855 | PUBLIC_URL: "https://d.test", |
| 856 | BUCKET: memoryBucket().bucket, |
| 857 | RUNNER: { |
| 858 | idFromName: (n: string) => n, |
| 859 | get: () => ({ |
| 860 | startJob: async (input: { token: string }) => { |
| 861 | token = input.token; |
| 862 | if (fail) throw fail(); |
| 863 | }, |
| 864 | stopJob: async () => {}, |
| 865 | }), |
| 866 | }, |
| 867 | RUN_WORKFLOW: { |
| 868 | get: async () => ({ |
| 869 | sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => { |
| 870 | if (workflowDown) throw new Error("internal error; reference = w"); |
| 871 | events.push(e); |
| 872 | }, |
| 873 | }), |
| 874 | }, |
| 875 | } as unknown as Env; |
| 876 | const actions = new Actions(repo, env); |
| 877 | const t = new Date().toISOString(); |
| 878 | 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); |
| 879 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j1', 'run1', 'k', 'k', 'queued'), ('j2', 'run1', 'k2', 'k2', 'queued'), ('j3', 'run1', 'k3', 'k3', 'queued')"); |
| 880 | const launch = (id: string) => (actions as unknown as { launch(id: string): Promise<void> }).launch(id); |
| 881 | const job = (id: string) => sql.exec("SELECT status, conclusion, container, placement_tries, early_stops FROM jobs WHERE id = ?", id).toArray()[0]; |
| 882 | const log = (id: string) => sql.exec<{ text: string }>("SELECT text FROM logs WHERE job_id = ? ORDER BY seq", id).toArray().map((l) => l.text); |
| 883 | |
| 884 | // The runtime's own internal error (seen on a real job, which a re-run then started at once): the first |
| 885 | // one starts the job again after QUICK_RETRY_SECONDS, not counted; the next waits like a full platform. |
| 886 | fail = () => new Error("internal error; reference = p80sjrfoqg66qdlk07p2qlco"); |
| 887 | await launch("j1"); |
| 888 | expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, container: null, placement_tries: 0, early_stops: 1 }); |
| 889 | expect(events.at(-1)).toEqual({ type: "job-j1", payload: { retryInSeconds: QUICK_RETRY_SECONDS } }); |
| 890 | expect(log("j1").at(-1)).toBe(`##[warning]No hosted runner could start the job yet: internal error; reference = p80sjrfoqg66qdlk07p2qlco. Diffinery tries again in ${QUICK_RETRY_SECONDS} s (not counted: the first failure like this is retried at once).`); |
| 891 | await actions.relaunchJob("j1"); |
| 892 | expect(job("j1")).toMatchObject({ status: "queued", placement_tries: 1 }); |
| 893 | expect(events.at(-1)).toEqual({ type: "job-j1", payload: { retryInSeconds: placementDelay(1) } }); |
| 894 | // A Durable Object the runtime replaced under the start, and any error the runtime marks retryable, go back in line too. |
| 895 | fail = () => new Error("Connection closed: this Durable Object instance is no longer active. Reconnect or retry the request."); |
| 896 | await actions.relaunchJob("j1"); |
| 897 | expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 2 }); |
| 898 | fail = () => new Error("out of memory: SQLITE_NOMEM"); |
| 899 | await actions.relaunchJob("j1"); |
| 900 | expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 3 }); |
| 901 | fail = () => Object.assign(new Error("something the runtime hasn't said before"), { retryable: true }); |
| 902 | await actions.relaunchJob("j1"); |
| 903 | expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 4 }); |
| 904 | // The retries stay bounded. |
| 905 | for (let i = 4; i < MAX_PLACEMENT_TRIES; i++) await actions.relaunchJob("j1"); |
| 906 | expect(job("j1")).toMatchObject({ status: "queued", placement_tries: MAX_PLACEMENT_TRIES }); |
| 907 | await actions.relaunchJob("j1"); |
| 908 | expect(job("j1")).toMatchObject({ status: "completed", conclusion: "failure" }); |
| 909 | |
| 910 | // A start that reports a failure after the job's runner fetched it leaves the job running: it isn't |
| 911 | // started again, nor failed under the runner. |
| 912 | fail = () => { |
| 913 | sql.exec("UPDATE jobs SET status = 'in_progress' WHERE id = 'j2'"); |
| 914 | return new Error("internal error; reference = x"); |
| 915 | }; |
| 916 | await launch("j2"); |
| 917 | expect(job("j2")).toMatchObject({ status: "in_progress", conclusion: null, container: "job-j2", placement_tries: 0, early_stops: 0 }); |
| 918 | fail = () => { |
| 919 | sql.exec("UPDATE jobs SET status = 'in_progress' WHERE id = 'j3'"); |
| 920 | return new Error("image not found"); |
| 921 | }; |
| 922 | await launch("j3"); |
| 923 | expect(job("j3")).toMatchObject({ status: "in_progress", conclusion: null }); |
| 924 | |
| 925 | // When the run's workflow can't be told to start the job again, this object does, from its alarm. |
| 926 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j5', 'run1', 'k5', 'k5', 'queued')"); |
| 927 | workflowDown = true; |
| 928 | fail = () => new Error("internal error; reference = z"); |
| 929 | const before = Date.now(); |
| 930 | await launch("j5"); |
| 931 | expect(job("j5")).toMatchObject({ status: "queued", conclusion: null, early_stops: 1 }); |
| 932 | const at = actions.nextImageRetry()!; |
| 933 | expect(at).toBeGreaterThanOrEqual(before + QUICK_RETRY_SECONDS * 1000); |
| 934 | expect(alarms.at(-1)).toBe(at); |
| 935 | workflowDown = false; |
| 936 | fail = null; |
| 937 | await actions.relaunchImageHeld(at); |
| 938 | expect(job("j5")).toMatchObject({ status: "waiting", container: "job-j5-p1" }); |
| 939 | // The workflow's own start, if its event got through after all, finds it started and does nothing. |
| 940 | await actions.relaunchJob("j5"); |
| 941 | expect(job("j5")).toMatchObject({ status: "waiting", container: "job-j5-p1" }); |
| 942 | // The object the failed start was in is a different one: a container that comes up there after all, |
| 943 | // and stops when its revoked token is refused, leaves the new placement alone. |
| 944 | await actions.containerStopped("j5", 1, "job-j5"); |
| 945 | expect(job("j5")).toMatchObject({ status: "waiting", container: "job-j5-p1" }); |
| 946 | |
| 947 | // And the other way round: a runner whose job went back in line while it fetched the job (its start |
| 948 | // was reported failed) is turned away before the job is marked started, so only the next runner runs it. |
| 949 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j4', 'run1', 'k4', 'k4', 'queued')"); |
| 950 | fail = null; |
| 951 | await launch("j4"); |
| 952 | const a = actions as unknown as { expanded: unknown; callContext: unknown; placementFailed(id: string, why: string): Promise<void> }; |
| 953 | a.expanded = async () => ({ workflow: { jobs: { k4: { steps: [] } } }, workflowOf: new Map([["k4", { env: {} }]]), parentOf: new Map(), needs: new Map() }); |
| 954 | a.callContext = async () => { |
| 955 | await a.placementFailed("j4", "internal error; reference = y"); |
| 956 | return { inputs: {}, secrets: {} }; |
| 957 | }; |
| 958 | await expect(actions.runnerSpec("j4", token)).rejects.toMatchObject({ status: 409 }); |
| 959 | expect(job("j4")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 1 }); |
| 960 | }); |
| 961 | |
| 962 | it("a matrix leg back in line after getting no machine waits out its pause when a sibling finishes", async () => { |
| 963 | const sql = sqlStorage(); |
| 964 | sql.exec(REPO_SCHEMA); |
| 965 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 966 | let full = true; |
| 967 | const started: string[] = []; |
| 968 | const events: { type: string; payload: Record<string, unknown> }[] = []; |
| 969 | const repo = { |
| 970 | sql, |
| 971 | meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" }, |
| 972 | upsertCheck: () => {}, |
| 973 | broadcast: () => {}, |
| 974 | deploys: { jobCompleted: () => {} }, |
| 975 | runtime: { sweepJob: async () => {} }, |
| 976 | alarmBy: () => {}, |
| 977 | } as unknown as RepoDO; |
| 978 | const DB = testD1(); |
| 979 | DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString()); |
| 980 | const env = { |
| 981 | DB, |
| 982 | PUBLIC_URL: "https://d.test", |
| 983 | BUCKET: memoryBucket().bucket, |
| 984 | RUNNER: { |
| 985 | idFromName: (n: string) => n, |
| 986 | get: (n: string) => ({ |
| 987 | startJob: async () => { |
| 988 | started.push(n); |
| 989 | if (full) throw new Error("there is no container instance that can be provided to this durable object"); |
| 990 | }, |
| 991 | stopJob: async () => {}, |
| 992 | }), |
| 993 | }, |
| 994 | RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) }, |
| 995 | } as unknown as Env; |
| 996 | const actions = new Actions(repo, env); |
| 997 | const a = actions as unknown as { launch(id: string): Promise<void>; completeJob(id: string, c: unknown): Promise<void> }; |
| 998 | const t = new Date().toISOString(); |
| 999 | 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); |
| 1000 | // Legs of one matrix: m1 and m2 started, m3 not yet (max-parallel). |
| 1001 | 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')"); |
| 1002 | // m4 is back in line after a safe restart (its runner was replaced before its steps ran). |
| 1003 | sql.exec("UPDATE jobs SET restarts = 1 WHERE id = 'm4'"); |
| 1004 | const job = (id: string) => sql.exec("SELECT status, placement_tries FROM jobs WHERE id = ?", id).toArray()[0]; |
| 1005 | await a.launch("m1"); |
| 1006 | expect(job("m1")).toMatchObject({ status: "queued", placement_tries: 1 }); |
| 1007 | full = false; |
| 1008 | await a.launch("m2"); |
| 1009 | expect(job("m2")).toMatchObject({ status: "waiting" }); |
| 1010 | started.length = 0; |
| 1011 | |
| 1012 | // m2 finishing starts the leg that never started, not the ones back in line: those are started |
| 1013 | // again by the run's workflow, so each try still waits its turn (placementDelay). |
| 1014 | await a.completeJob("m2", { conclusion: "success", outputs: {} }); |
| 1015 | expect(started).toEqual(["job-m3"]); |
| 1016 | expect(job("m1")).toMatchObject({ status: "queued", placement_tries: 1 }); |
| 1017 | expect(job("m4")).toMatchObject({ status: "queued" }); |
| 1018 | expect(events.filter((e) => e.type === "job-m1")).toHaveLength(1); |
| 1019 | }); |
| 1020 | |
| 1021 | 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 () => { |
| 1022 | const sql = sqlStorage(); |
| 1023 | sql.exec(REPO_SCHEMA); |
| 1024 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 1025 | const events: { type: string; payload: Record<string, unknown> }[] = []; |
| 1026 | const repo = { |
| 1027 | sql, |
| 1028 | meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" }, |
| 1029 | upsertCheck: () => {}, |
| 1030 | broadcast: () => {}, |
| 1031 | deploys: { jobCompleted: () => {} }, |
| 1032 | runtime: { sweepJob: async () => {} }, |
| 1033 | alarmBy: (at: number) => void alarms.push(at), |
| 1034 | } as unknown as RepoDO; |
| 1035 | const alarms: number[] = []; |
| 1036 | const DB = testD1(); |
| 1037 | DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString()); |
| 1038 | const env = { |
| 1039 | DB, |
| 1040 | PUBLIC_URL: "https://d.test", |
| 1041 | RUNNER_IMAGE: "diffinery-runner:new", |
| 1042 | RUNNER: { idFromName: (n: string) => n, get: () => ({ startJob: async () => {}, stopJob: async () => {} }) }, |
| 1043 | RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) }, |
| 1044 | } as unknown as Env; |
| 1045 | const actions = new Actions(repo, env); |
| 1046 | const t = new Date().toISOString(); |
| 1047 | 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); |
| 1048 | 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')"); |
| 1049 | sql.exec("UPDATE jobs SET labels = '[\"self-hosted\"]' WHERE id = 's1'"); |
| 1050 | const a = actions as unknown as { launch(id: string): Promise<void>; refuseStaleRunner(job: unknown, runner: unknown, deployed?: string): Promise<void>; jobRow(id: string): unknown }; |
| 1051 | const refuse = (id: string, image?: string, deployed?: string) => a.refuseStaleRunner(a.jobRow(id), image ? { image } : null, deployed); |
| 1052 | const job = (id = "j1") => sql.exec("SELECT status, placement_tries, stale_tries FROM jobs WHERE id = ?", id).toArray()[0]; |
| 1053 | const logs = () => sql.exec<{ text: string }>("SELECT text FROM logs WHERE job_id = 'j1' ORDER BY seq").toArray().map((l) => l.text); |
| 1054 | |
| 1055 | await a.launch("j1"); |
| 1056 | // The deployed image runs the job. |
| 1057 | await expect(refuse("j1", "diffinery-runner:new")).resolves.toBeUndefined(); |
| 1058 | expect(job()).toMatchObject({ status: "waiting", placement_tries: 0 }); |
| 1059 | // The previous image (or one that doesn't say) is turned away. The job starts again after a short fixed |
| 1060 | // wait, from this object's alarm rather than the run's workflow, and the waits don't count as placement tries. |
| 1061 | for (let i = 0; i < STALE_IMAGE_TRIES; i++) { |
| 1062 | const before = Date.now(); |
| 1063 | await expect(refuse("j1", i === 1 ? undefined : "diffinery-runner:old")).rejects.toMatchObject({ status: 409 }); |
| 1064 | expect(job()).toMatchObject({ status: "queued", placement_tries: 0, stale_tries: i + 1 }); |
| 1065 | const at = actions.nextImageRetry()!; |
| 1066 | expect(at).toBeGreaterThanOrEqual(before + STALE_IMAGE_RETRY_SECONDS * 1000); |
| 1067 | expect(alarms.at(-1)).toBe(at); |
| 1068 | // Not before its time. |
| 1069 | await actions.relaunchImageHeld(at - 1); |
| 1070 | expect(job()).toMatchObject({ status: "queued" }); |
| 1071 | await actions.relaunchImageHeld(at); |
| 1072 | expect(job()).toMatchObject({ status: "waiting" }); |
| 1073 | expect(actions.nextImageRetry()).toBeNull(); |
| 1074 | } |
| 1075 | expect(events).toEqual([]); |
| 1076 | 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).`); |
| 1077 | // A rollout that doesn't finish doesn't hold the job up past STALE_IMAGE_TRIES. |
| 1078 | await expect(refuse("j1", "diffinery-runner:old")).resolves.toBeUndefined(); |
| 1079 | // A job cancelled while it waits isn't started again. |
| 1080 | sql.exec("UPDATE jobs SET status = 'queued', image_retry_at = 1 WHERE id = 'j1'"); |
| 1081 | sql.exec("UPDATE jobs SET status = 'completed' WHERE id = 'j1'"); |
| 1082 | await actions.relaunchImageHeld(); |
| 1083 | expect(job()).toMatchObject({ status: "completed" }); |
| 1084 | // Self-hosted runners run any image. |
| 1085 | await expect(refuse("s1", "my-image")).resolves.toBeUndefined(); |
| 1086 | |
| 1087 | // Right after a deploy this object can still run the previous version (its RUNNER_IMAGE is the |
| 1088 | // old one) while the Worker that took the runner's request has the new one: the Worker's wins. |
| 1089 | await a.launch("j2"); |
| 1090 | await expect(refuse("j2", "diffinery-runner:next", "diffinery-runner:next")).resolves.toBeUndefined(); |
| 1091 | expect(job("j2")).toMatchObject({ status: "waiting", placement_tries: 0 }); |
| 1092 | await expect(refuse("j2", "diffinery-runner:new", "diffinery-runner:next")).rejects.toMatchObject({ status: 409 }); |
| 1093 | expect(job("j2")).toMatchObject({ status: "queued", stale_tries: 1 }); |
| 1094 | }); |
| 1095 | |
| 1096 | 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 () => { |
| 1097 | const sql = sqlStorage(); |
| 1098 | sql.exec(REPO_SCHEMA); |
| 1099 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 1100 | const events: { type: string; payload: Record<string, unknown> }[] = []; |
| 1101 | const stopped: string[] = []; |
| 1102 | const repo = { |
| 1103 | sql, |
| 1104 | meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" }, |
| 1105 | upsertCheck: () => {}, |
| 1106 | broadcast: () => {}, |
| 1107 | deploys: { jobCompleted: () => {} }, |
| 1108 | runtime: { sweepJob: async () => {} }, |
| 1109 | alarmBy: () => {}, |
| 1110 | } as unknown as RepoDO; |
| 1111 | const env = { |
| 1112 | PUBLIC_URL: "https://d.test", |
| 1113 | BUCKET: memoryBucket().bucket, |
| 1114 | RUNNER: { idFromName: (n: string) => n, get: (n: string) => ({ startJob: async () => {}, stopJob: async () => void stopped.push(n) }) }, |
| 1115 | RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) }, |
| 1116 | } as unknown as Env; |
| 1117 | const actions = new Actions(repo, env); |
| 1118 | const t0 = Date.parse("2026-10-16T12:00:00Z"); |
| 1119 | const at = (ms: number) => new Date(t0 + ms).toISOString(); |
| 1120 | 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)); |
| 1121 | // Each runner fetched its job at t0 (started_at). f: frozen, timeout 10 min; h: healthy, timeout 30; |
| 1122 | // d: no timeout of its own (6 hours); s: on a self-hosted runner, frozen too; q: still in line; c: finished. |
| 1123 | const add = (id: string, minutes: number | null, o: { status?: string; container?: string | null; started?: string | null } = {}) => |
| 1124 | sql.exec( |
| 1125 | "INSERT INTO jobs (id, run_id, job_key, name, status, timeout_minutes, container, started_at, strategy) VALUES (?, 'run1', ?, ?, ?, ?, ?, ?, '{\"fail-fast\":false}')", |
| 1126 | id, id, id, o.status ?? "in_progress", minutes, o.container === undefined ? `job-${id}` : o.container, o.started === undefined ? at(0) : o.started, |
| 1127 | ); |
| 1128 | add("f", 10); |
| 1129 | add("h", 30); |
| 1130 | add("d", 0); |
| 1131 | add("s", 10, { container: "self-hosted:box" }); |
| 1132 | add("q", 10, { status: "queued", container: null, started: null }); |
| 1133 | add("c", 10, { status: "completed" }); |
| 1134 | const job = (id: string) => sql.exec<{ status: string; conclusion: string | null; failure: string | null }>("SELECT status, conclusion, failure FROM jobs WHERE id = ?", id).one(); |
| 1135 | const min = 60_000; |
| 1136 | expect(actions.nextJobDeadline()).toBe(t0 + 10 * min + JOB_DEADLINE_GRACE_MS); |
| 1137 | |
| 1138 | // Past f's timeout but within the grace its runner has to stop the steps and report: nothing happens. |
| 1139 | await actions.enforceJobDeadlines(t0 + 10 * min + JOB_DEADLINE_GRACE_MS - 1); |
| 1140 | expect(["f", "h", "d", "s"].map((id) => job(id).status)).toEqual(["in_progress", "in_progress", "in_progress", "in_progress"]); |
| 1141 | expect(stopped).toEqual([]); |
| 1142 | |
| 1143 | // At the deadline the frozen ones end: the hosted container is stopped, both fail as timed out. |
| 1144 | await actions.enforceJobDeadlines(t0 + 10 * min + JOB_DEADLINE_GRACE_MS); |
| 1145 | for (const id of ["f", "s"]) { |
| 1146 | expect(job(id)).toMatchObject({ status: "completed", conclusion: "timed_out" }); |
| 1147 | 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.`); |
| 1148 | } |
| 1149 | expect(stopped).toEqual(["job-f"]); |
| 1150 | expect(events.filter((e) => e.type === "job-f")).toEqual([{ type: "job-f", payload: { conclusion: "timed_out" } }]); |
| 1151 | // The others are untouched, and the next wake-up is h's deadline. |
| 1152 | expect([job("h").status, job("d").status, job("q").status, job("c").status]).toEqual(["in_progress", "in_progress", "queued", "completed"]); |
| 1153 | expect(actions.nextJobDeadline()).toBe(t0 + 30 * min + JOB_DEADLINE_GRACE_MS); |
| 1154 | // A job without timeout-minutes gets GitHub's 6 hours. |
| 1155 | await actions.enforceJobDeadlines(t0 + 360 * min + JOB_DEADLINE_GRACE_MS - 1); |
| 1156 | expect(job("d").status).toBe("in_progress"); |
| 1157 | await actions.enforceJobDeadlines(t0 + 360 * min + JOB_DEADLINE_GRACE_MS); |
| 1158 | expect(job("d")).toMatchObject({ status: "completed", conclusion: "timed_out" }); |
| 1159 | expect(actions.nextJobDeadline()).toBeNull(); |
| 1160 | }); |
| 1161 | |
| 1162 | it("cancelling a run lets running hosted jobs run their steps that run after a cancel, and ends them after CANCEL_STOP_MS", async () => { |
| 1163 | const sql = sqlStorage(); |
| 1164 | sql.exec(REPO_SCHEMA); |
| 1165 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 1166 | const events: { type: string; payload: Record<string, unknown> }[] = []; |
| 1167 | const stopped: [string, boolean][] = []; |
| 1168 | const alarms: number[] = []; |
| 1169 | const repo = { |
| 1170 | sql, |
| 1171 | meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" }, |
| 1172 | upsertCheck: () => {}, |
| 1173 | broadcast: () => {}, |
| 1174 | deploys: { jobCompleted: () => {} }, |
| 1175 | runtime: { sweepJob: async () => {}, grantFor: () => ({ role: "read", permissions: {} }) }, |
| 1176 | alarmBy: (at: number) => void alarms.push(at), |
| 1177 | } as unknown as RepoDO; |
| 1178 | const env = { |
| 1179 | PUBLIC_URL: "https://d.test", |
| 1180 | BUCKET: memoryBucket().bucket, |
| 1181 | RUNNER: { idFromName: (n: string) => n, get: (n: string) => ({ startJob: async () => {}, stopJob: async (graceful = false) => void stopped.push([n, graceful]) }) }, |
| 1182 | RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) }, |
| 1183 | } as unknown as Env; |
| 1184 | const actions = new Actions(repo, env); |
| 1185 | const t = new Date().toISOString(); |
| 1186 | 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); |
| 1187 | // a and b run on hosted runners (a reports after its cleanup steps, b's runner never does); s runs |
| 1188 | // on a self-hosted runner; w's container hasn't fetched the job yet; q waits in line. |
| 1189 | const token = (id: string) => `djt_r_${id}_secret`; |
| 1190 | const add = async (id: string, status: string, container: string | null) => |
| 1191 | sql.exec( |
| 1192 | "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}')", |
| 1193 | id, id, id, status, container, status === "in_progress" ? t : null, await sha256(token(id)), t, |
| 1194 | ); |
| 1195 | await add("a", "in_progress", "job-a"); |
| 1196 | await add("b", "in_progress", "job-b"); |
| 1197 | await add("s", "in_progress", "self-hosted:box"); |
| 1198 | await add("w", "waiting", "job-w"); |
| 1199 | await add("q", "queued", null); |
| 1200 | const job = (id: string) => sql.exec<{ status: string; conclusion: string | null; failure: string | null }>("SELECT status, conclusion, failure FROM jobs WHERE id = ?", id).one(); |
| 1201 | const before = Date.now(); |
| 1202 | await actions.cancel("run1", { kind: "user", login: "octo", role: "write" } as unknown as Caller); |
| 1203 | // a and b get SIGINT and keep running; the rest end at once, w's container stopped outright. |
| 1204 | expect(stopped).toEqual([["job-a", true], ["job-b", true], ["job-w", false]]); |
| 1205 | expect([job("a").status, job("b").status]).toEqual(["in_progress", "in_progress"]); |
| 1206 | for (const id of ["s", "w", "q"]) expect(job(id)).toMatchObject({ status: "completed", conclusion: "cancelled", failure: JSON.stringify({ message: "Cancelled by octo" }) }); |
| 1207 | expect(alarms.at(-1)).toBeGreaterThanOrEqual(before + CANCEL_STOP_MS); |
| 1208 | const deadline = actions.nextJobDeadline()!; |
| 1209 | expect(deadline - before).toBeGreaterThanOrEqual(CANCEL_STOP_MS); |
| 1210 | expect(deadline - Date.now()).toBeLessThanOrEqual(CANCEL_STOP_MS); |
| 1211 | // a's runner reports once its cleanup steps ran: cancelled, saying who cancelled it, never started again. |
| 1212 | await actions.runnerComplete("a", token("a"), { conclusion: "cancelled", outputs: {}, failure: { message: "The job was cancelled." }, replaced: true }); |
| 1213 | expect(job("a")).toMatchObject({ status: "completed", conclusion: "cancelled", failure: JSON.stringify({ message: "Cancelled by octo" }) }); |
| 1214 | |
| 1215 | // b's runner never reports: at CANCEL_STOP_MS its container is stopped and the job ends. |
| 1216 | await actions.enforceJobDeadlines(deadline - 1); |
| 1217 | expect(job("b").status).toBe("in_progress"); |
| 1218 | await actions.enforceJobDeadlines(deadline); |
| 1219 | expect(stopped.at(-1)).toEqual(["job-b", false]); |
| 1220 | expect(job("b").conclusion).toBe("cancelled"); |
| 1221 | 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.`); |
| 1222 | expect(actions.nextJobDeadline()).toBeNull(); |
| 1223 | }); |
| 1224 | |
| 1225 | 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 () => { |
| 1226 | const sql = sqlStorage(); |
| 1227 | sql.exec(REPO_SCHEMA); |
| 1228 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 1229 | const repo = { |
| 1230 | sql, |
| 1231 | meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" }, |
| 1232 | upsertCheck: () => {}, |
| 1233 | broadcast: () => {}, |
| 1234 | deploys: { jobCompleted: () => {} }, |
| 1235 | runtime: { sweepJob: async () => {} }, |
| 1236 | alarmBy: () => {}, |
| 1237 | } as unknown as RepoDO; |
| 1238 | const env = { |
| 1239 | PUBLIC_URL: "https://d.test", |
| 1240 | BUCKET: memoryBucket().bucket, |
| 1241 | RUNNER_IMAGE: "diffinery-runner:new", |
| 1242 | RUNNER: { idFromName: (n: string) => n, get: (n: string) => ({ startJob: async () => {}, stopJob: async (graceful = false) => void stopped.push([n, graceful]) }) }, |
| 1243 | RUN_WORKFLOW: { get: async () => ({ sendEvent: async () => {} }) }, |
| 1244 | } as unknown as Env; |
| 1245 | const stopped: [string, boolean][] = []; |
| 1246 | const actions = new Actions(repo, env); |
| 1247 | const t = new Date().toISOString(); |
| 1248 | const job = (id: string) => sql.exec("SELECT status, conclusion, failure FROM jobs WHERE id = ?", id).one(); |
| 1249 | 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); |
| 1250 | // On an older runner image than the deployed one, which would otherwise read as a rollout and start the job again. |
| 1251 | 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); |
| 1252 | const octo = { kind: "user", login: "octo", role: "write" } as unknown as Caller; |
| 1253 | await actions.cancel("run1", octo); |
| 1254 | await actions.cancel("run1", { kind: "user", login: "hubot", role: "write" } as unknown as Caller); |
| 1255 | expect(stopped).toEqual([["job-j1", true], ["job-j1", false]]); |
| 1256 | expect(job("j1")).toMatchObject({ status: "completed", conclusion: "cancelled", failure: JSON.stringify({ message: "Cancelled by octo" }) }); |
| 1257 | await actions.cancel("run2", octo); |
| 1258 | await actions.containerStopped("j2", null, "job-j2", true); |
| 1259 | expect(job("j2")).toMatchObject({ status: "completed", conclusion: "cancelled", failure: JSON.stringify({ message: "Cancelled by octo" }) }); |
| 1260 | }); |
| 1261 | |
| 1262 | 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 () => { |
| 1263 | const sql = sqlStorage(); |
| 1264 | sql.exec(REPO_SCHEMA); |
| 1265 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 1266 | const events: { type: string; payload: Record<string, unknown> }[] = []; |
| 1267 | const swept: [string, boolean][] = []; |
| 1268 | const repo = { |
| 1269 | sql, |
| 1270 | meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" }, |
| 1271 | upsertCheck: () => {}, |
| 1272 | broadcast: () => {}, |
| 1273 | deploys: { jobCompleted: () => {} }, |
| 1274 | runtime: { sweepJob: async (id: string, o?: { restart?: boolean }) => void swept.push([id, !!o?.restart]), grantFor: () => ({ role: "read", permissions: {} }) }, |
| 1275 | policy: { settings: () => ({ defaultPermissions: "write" }), isForkPull: () => false }, |
| 1276 | alarmBy: () => {}, |
| 1277 | } as unknown as RepoDO; |
| 1278 | const DB = testD1(); |
| 1279 | DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString()); |
| 1280 | const started: { container: string; token: string }[] = []; |
| 1281 | const env = { |
| 1282 | DB, |
| 1283 | PUBLIC_URL: "https://d.test", |
| 1284 | BUCKET: memoryBucket().bucket, |
| 1285 | RUNNER_IMAGE: "diffinery-runner:new", |
| 1286 | RUNNER: { idFromName: (n: string) => n, get: () => ({ startJob: async (i: { container: string; token: string }) => void started.push(i), stopJob: async () => {} }) }, |
| 1287 | RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) }, |
| 1288 | } as unknown as Env; |
| 1289 | const actions = new Actions(repo, env); |
| 1290 | // The job's steps: checkout, then one whose effects running it twice would repeat. |
| 1291 | // 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. |
| 1292 | (actions as unknown as { expanded: () => Promise<unknown> }).expanded = async () => { |
| 1293 | const steps = [{ uses: "actions/checkout@v4" }, { run: "make deploy" }]; |
| 1294 | const read = { contents: "read" }; |
| 1295 | const install = [{ uses: "actions/checkout@v4" }, { run: "pnpm install" }, { run: "pnpm test" }]; |
| 1296 | const jobs: Record<string, unknown> = { |
| 1297 | k: { steps }, |
| 1298 | k2: { steps }, |
| 1299 | k3: { steps }, |
| 1300 | k4: { steps }, |
| 1301 | r1: { steps: install, permissions: read, env: { GH_TOKEN: "${{ secrets.GITHUB_TOKEN }}" } }, |
| 1302 | r2: { steps: [...install, { run: "npm publish", env: { NODE_AUTH_TOKEN: "${{ secrets.NPM_TOKEN }}" } }], permissions: read }, |
| 1303 | r3: { steps: install, permissions: read, environment: { name: "production" } }, |
| 1304 | r4: { steps: install, permissions: { contents: "read", "id-token": "write" } }, |
| 1305 | r5: { steps: [...install, { run: "echo ${{ toJSON(secrets) }}" }], permissions: read }, |
| 1306 | }; |
| 1307 | const workflow = { env: {}, jobs }; |
| 1308 | return { workflow, calls: new Map(), parentOf: new Map(), needs: new Map(), workflowOf: new Map(Object.keys(jobs).map((key) => [key, workflow])) }; |
| 1309 | }; |
| 1310 | const t = new Date().toISOString(); |
| 1311 | 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); |
| 1312 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j1', 'run1', 'k', 'k', 'queued'), ('j2', 'run1', 'k2', 'k2', 'queued')"); |
| 1313 | const launch = (id: string) => (actions as unknown as { launch(id: string): Promise<void> }).launch(id); |
| 1314 | const job = (id: string) => sql.exec("SELECT status, conclusion, container, token_hash, steps, restarts, failure FROM jobs WHERE id = ?", id).toArray()[0]; |
| 1315 | const logs = (id: string) => sql.exec("SELECT seq, step, text FROM logs WHERE job_id = ? ORDER BY seq", id).toArray(); |
| 1316 | // The runner fetched the job on `image` and reported these steps as started. |
| 1317 | const running = (id: string, image: string, steps: number[], names = ["Run actions/checkout@v4", "Deploy"]) => { |
| 1318 | const reports = steps.map((index) => ({ index, name: names[index] ?? `Step ${index}`, status: "in_progress", startedAt: t })); |
| 1319 | sql.exec("UPDATE jobs SET status = 'in_progress', started_at = ?, runner_info = ?, steps = ? WHERE id = ?", t, JSON.stringify({ image }), JSON.stringify(reports), id); |
| 1320 | sql.exec("INSERT INTO logs (job_id, seq, t, step, text) VALUES (?, 0, ?, -1, 'set up'), (?, 1, ?, 0, 'checking out')", id, t, id, t); |
| 1321 | }; |
| 1322 | const replaced = { conclusion: "cancelled" as const, outputs: {}, replaced: true }; |
| 1323 | |
| 1324 | // The runner reports the replacement during checkout: the job goes back in line, its log starts over with a line that says why. |
| 1325 | await launch("j1"); |
| 1326 | const first = started.at(-1)!; |
| 1327 | expect(first.container).toBe("job-j1"); |
| 1328 | running("j1", "diffinery-runner:old", [0]); |
| 1329 | await actions.runnerComplete("j1", first.token, replaced); |
| 1330 | expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, container: null, token_hash: null, steps: "[]", restarts: 1 }); |
| 1331 | expect(events.at(-1)).toEqual({ type: "job-j1", payload: { retryInSeconds: 10 } }); |
| 1332 | 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\)/) }]); |
| 1333 | // The replaced runner can't report any more, and its container stopping changes nothing. |
| 1334 | await expect(actions.runnerComplete("j1", first.token, replaced)).rejects.toMatchObject({ status: 401 }); |
| 1335 | await actions.containerStopped("j1", 143, "job-j1"); |
| 1336 | expect(job("j1")).toMatchObject({ status: "queued", restarts: 1 }); |
| 1337 | |
| 1338 | // It starts in a new container; the old one stopping late doesn't touch it. |
| 1339 | await actions.relaunchJob("j1"); |
| 1340 | expect(started.at(-1)!.container).toBe("job-j1-r1"); |
| 1341 | await actions.containerStopped("j1", 143, "job-j1"); |
| 1342 | expect(job("j1")).toMatchObject({ status: "waiting", container: "job-j1-r1" }); |
| 1343 | |
| 1344 | // A container on a replaced image that stops without reporting counts the same. |
| 1345 | running("j1", "diffinery-runner:old", []); |
| 1346 | await actions.containerStopped("j1", 143, "job-j1-r1"); |
| 1347 | expect(job("j1")).toMatchObject({ status: "queued", restarts: 2 }); |
| 1348 | |
| 1349 | // Past MAX_RUNNER_RESTARTS the job fails, and says why. |
| 1350 | await actions.relaunchJob("j1"); |
| 1351 | expect(started.at(-1)!.container).toBe("job-j1-r2"); |
| 1352 | running("j1", "diffinery-runner:old", []); |
| 1353 | await actions.runnerComplete("j1", started.at(-1)!.token, replaced); |
| 1354 | expect(job("j1")).toMatchObject({ status: "completed", conclusion: "failure", restarts: MAX_RUNNER_RESTARTS }); |
| 1355 | 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\./); |
| 1356 | |
| 1357 | // Once a step other than checkout ran, the job isn't started again: it fails and names the step. |
| 1358 | await launch("j2"); |
| 1359 | running("j2", "diffinery-runner:old", [0, 1]); |
| 1360 | await actions.runnerComplete("j2", started.at(-1)!.token, replaced); |
| 1361 | expect(job("j2")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 }); |
| 1362 | 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/); |
| 1363 | |
| 1364 | // A job that can only read starts again even after its steps ran (here the platform stopped |
| 1365 | // `pnpm install`), and what its stopped attempt uploaded goes. Each restart is counted. |
| 1366 | 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')"); |
| 1367 | const install = ["Run actions/checkout@v4", "Run pnpm install", "Run pnpm test"]; |
| 1368 | await launch("r1"); |
| 1369 | running("r1", "diffinery-runner:new", [0, 1], install); |
| 1370 | await actions.runnerComplete("r1", started.at(-1)!.token, replaced); |
| 1371 | expect(job("r1")).toMatchObject({ status: "queued", conclusion: null, steps: "[]", restarts: 1 }); |
| 1372 | expect(swept.at(-1)).toEqual(["r1", true]); |
| 1373 | 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\)/) }]); |
| 1374 | // It still stops after MAX_RUNNER_RESTARTS, wherever the platform stopped it. |
| 1375 | for (let i = 1; i <= MAX_RUNNER_RESTARTS; i++) { |
| 1376 | await actions.relaunchJob("r1"); |
| 1377 | running("r1", "diffinery-runner:new", [0, 1, 2], install); |
| 1378 | await actions.runnerComplete("r1", started.at(-1)!.token, replaced); |
| 1379 | } |
| 1380 | expect(job("r1")).toMatchObject({ status: "completed", conclusion: "failure", restarts: MAX_RUNNER_RESTARTS }); |
| 1381 | // (On the deployed image no deploy replaced it, so the message says the platform stopped it.) |
| 1382 | 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."); |
| 1383 | // 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. |
| 1384 | for (const id of ["r2", "r3", "r4", "r5"]) { |
| 1385 | await launch(id); |
| 1386 | running(id, "diffinery-runner:new", [0, 1, 2], install); |
| 1387 | await actions.runnerComplete(id, started.at(-1)!.token, replaced); |
| 1388 | expect(job(id)).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 }); |
| 1389 | 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/); |
| 1390 | } |
| 1391 | // Right after a deploy this object can still have the previous RUNNER_IMAGE: the image of the |
| 1392 | // Worker that took the report decides whether a deploy replaced the runner. |
| 1393 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('r6', 'run1', 'r1', 'r6', 'queued'), ('r7', 'run1', 'r1', 'r7', 'queued')"); |
| 1394 | 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) { |
| 1395 | await launch(id); |
| 1396 | running(id, "diffinery-runner:next", [0, 1], install); |
| 1397 | await actions.runnerComplete(id, started.at(-1)!.token, replaced, deployed); |
| 1398 | expect(job(id)).toMatchObject({ status: "queued", restarts: 1 }); |
| 1399 | expect(logs(id)[0]!.text).toMatch(new RegExp(`^##\\[warning\\]${says} while step "Run pnpm install" ran`)); |
| 1400 | } |
| 1401 | |
| 1402 | // A job that finished despite the replacement, or a runner on the deployed image that stops, completes as before. |
| 1403 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j3', 'run1', 'k3', 'k3', 'queued'), ('j4', 'run1', 'k4', 'k4', 'queued')"); |
| 1404 | await launch("j3"); |
| 1405 | running("j3", "diffinery-runner:old", [0, 1]); |
| 1406 | await actions.runnerComplete("j3", started.at(-1)!.token, { conclusion: "success", outputs: {}, replaced: true }); |
| 1407 | expect(job("j3")).toMatchObject({ status: "completed", conclusion: "success" }); |
| 1408 | await launch("j4"); |
| 1409 | running("j4", "diffinery-runner:new", []); |
| 1410 | await actions.containerStopped("j4", 137, "job-j4"); |
| 1411 | expect(job("j4")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 }); |
| 1412 | expect(JSON.parse(job("j4")!.failure as string).message).toBe("The runner stopped unexpectedly (exit code 137)."); |
| 1413 | |
| 1414 | // A container on the deployed image that the platform stops or loses mid-job (no exit code, not |
| 1415 | // stopped by Diffinery) counts as a replacement too: a job that can only read starts again... |
| 1416 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('s1', 'run1', 'r1', 's1', 'queued'), ('s2', 'run1', 'k3', 's2', 'queued')"); |
| 1417 | await launch("s1"); |
| 1418 | running("s1", "diffinery-runner:new", [0, 1], install); |
| 1419 | await actions.containerStopped("s1", 0, "job-s1"); |
| 1420 | expect(job("s1")).toMatchObject({ status: "queued", conclusion: null, restarts: 1 }); |
| 1421 | expect(swept.at(-1)).toEqual(["s1", true]); |
| 1422 | 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\)\.$/) }]); |
| 1423 | // ...a late stop of that container changes nothing, and the restarts stop at MAX_RUNNER_RESTARTS. |
| 1424 | await actions.containerStopped("s1", null, "job-s1"); |
| 1425 | expect(job("s1")).toMatchObject({ status: "queued", restarts: 1 }); |
| 1426 | for (let i = 1; i <= MAX_RUNNER_RESTARTS; i++) { |
| 1427 | await actions.relaunchJob("s1"); |
| 1428 | running("s1", "diffinery-runner:new", [0, 1], install); |
| 1429 | await actions.containerStopped("s1", null, started.at(-1)!.container); |
| 1430 | } |
| 1431 | expect(job("s1")).toMatchObject({ status: "completed", conclusion: "failure", restarts: MAX_RUNNER_RESTARTS }); |
| 1432 | 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."); |
| 1433 | // One that could change something outside the run fails, naming the step. |
| 1434 | await launch("s2"); |
| 1435 | running("s2", "diffinery-runner:new", [0, 1]); |
| 1436 | await actions.containerStopped("s2", 0, "job-s2"); |
| 1437 | expect(job("s2")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 }); |
| 1438 | 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/); |
| 1439 | // Diffinery's own stop (a cancel, a timeout, fail-fast) never starts a job again, and a runner that crashes isn't retried. |
| 1440 | // (Each added after the one before failed, so fail-fast doesn't cancel it.) |
| 1441 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('s3', 'run1', 'r1', 's3', 'queued')"); |
| 1442 | await launch("s3"); |
| 1443 | running("s3", "diffinery-runner:new", [0, 1], install); |
| 1444 | await actions.containerStopped("s3", 0, "job-s3", true); |
| 1445 | expect(job("s3")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 }); |
| 1446 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('s4', 'run1', 'r1', 's4', 'queued')"); |
| 1447 | await launch("s4"); |
| 1448 | running("s4", "diffinery-runner:new", [0, 1], install); |
| 1449 | await actions.containerStopped("s4", 1, "job-s4"); |
| 1450 | expect(job("s4")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 }); |
| 1451 | }); |
| 1452 | }); |
| 1453 | |
| 1454 | describe("runner info", () => { |
| 1455 | it("keeps only the runner's short image, location and instance strings", () => { |
| 1456 | 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" }); |
| 1457 | expect(parseRunnerInfo(JSON.stringify({ image: "x".repeat(300) }))?.image).toHaveLength(200); |
| 1458 | for (const bad of [null, "", "not json", "{}", JSON.stringify({ image: 5 }), "x".repeat(2000)]) expect(parseRunnerInfo(bad)).toBeNull(); |
| 1459 | }); |
| 1460 | }); |
| 1461 | |
| 1462 | describe("unfinished runs", () => { |
| 1463 | it("past MAX_ACTIVE_RUNS a dispatch is refused and other events start nothing", async () => { |
| 1464 | const sql = sqlStorage(); |
| 1465 | sql.exec(REPO_SCHEMA); |
| 1466 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 1467 | 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); |
| 1468 | const t = new Date().toISOString(); |
| 1469 | 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); |
| 1470 | const wf = { path: ".github/workflows/w.yml", on: {}, jobs: {} } as never; |
| 1471 | const caller = { id: "u1", login: "ada", kind: "user" as const }; |
| 1472 | await expect(actions.createRun(wf, { event: "workflow_dispatch", ref: "refs/heads/main", sha: "abc", payload: {}, caller, title: "w" })).rejects.toThrow(/500 unfinished workflow runs/); |
| 1473 | expect(await actions.createRun(wf, { event: "push", ref: "refs/heads/main", sha: "abc", payload: {}, caller, title: "w" })).toBeNull(); |
| 1474 | }); |
| 1475 | }); |