| 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, JOB_DEADLINE_GRACE_MS, MAX_ACTIVE_RUNS, MAX_PLACEMENT_TRIES, MAX_RUNNER_RESTARTS, parseRunnerInfo, placementDelay, QUICK_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 } 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("merges many small blocks in rounds so that no step exceeds its calls", async () => { |
| 280 | // 10 calls per step: 300 blocks need two rounds of merging before the parts |
| 281 | const r = await run(Array.from({ length: 300 }, (_, i) => 40 * 1024 + i), { stepOps: 10, maxMergeBytes: 1 * MiB }); |
| 282 | expect(r.maxStepOps).toBeLessThanOrEqual(10); |
| 283 | expect(r.multipart).toBe(1); |
| 284 | }); |
| 285 | |
| 286 | it("keeps a 50,000-block list (Azure's maximum) within the Workers subrequest limit", async () => { |
| 287 | // Real data: 50k tiny blocks, as buildkit's gha cache can send. |
| 288 | const r = await run(Array.from({ length: 50_000 }, (_, i) => 16 + (i % 7))); |
| 289 | expect(r.maxStepOps).toBeLessThanOrEqual(STEP_OPS); |
| 290 | expect(r.steps + r.workerOps).toBeLessThan(1000); |
| 291 | // Planned only (no bytes): 50k blocks of 200 KiB (10 GB), and of 1 MiB with 4 MiB blocks mixed in. |
| 292 | for (const sizes of [Array.from({ length: 50_000 }, () => 200 * 1024), Array.from({ length: 50_000 }, (_, i) => (i % 10 ? MiB : 4 * MiB))]) { |
| 293 | const plan = planCompose(sizes.map((size, i) => ({ key: `k${i}`, size })), "s/"); |
| 294 | const steps = [...plan.rounds.flat(), ...batchParts(plan.parts)]; |
| 295 | for (const s of steps) { |
| 296 | const ops = "kind" in s ? s.slices.length + 1 : s.reduce((n, p) => n + p.slices.length + 1, 0); |
| 297 | expect(ops).toBeLessThanOrEqual(STEP_OPS); |
| 298 | } |
| 299 | // list pages and deletes of 50k staged blocks (50 each), the steps, create and complete |
| 300 | expect(50 + 50 + steps.length + 2).toBeLessThan(1000); |
| 301 | expect(plan.parts.length).toBeLessThanOrEqual(10_000); |
| 302 | expect(plan.parts.slice(0, -1).every((p) => p.reduce((n, s) => n + s.length, 0) === plan.partSize)).toBe(true); |
| 303 | } |
| 304 | // Writes and composes 50k objects: about 1.5s alone, and several times that on a busy CI runner. |
| 305 | }, 20_000); |
| 306 | }); |
| 307 | |
| 308 | describe("deleting a repository's transient objects", () => { |
| 309 | it("removes every object under its prefixes, a page at a time, and nothing else", async () => { |
| 310 | const m = memoryBucket(); |
| 311 | 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")]; |
| 312 | for (const k of keys) await m.bucket.put(k, new Uint8Array(1)); |
| 313 | for (const k of ["actions-cache/r10/1", "repos/r1/objects/ab/cd", "artifacts/r2/1"]) await m.bucket.put(k, new Uint8Array(1)); |
| 314 | m.stats.ops = 0; |
| 315 | expect(transientPrefixes("r1").some((p) => p.startsWith("repos/"))).toBe(false); |
| 316 | expect(await deletePrefixes(m.bucket, transientPrefixes("r1"))).toBe(true); |
| 317 | expect([...m.objects.keys()].sort()).toEqual(["actions-cache/r10/1", "artifacts/r2/1", "repos/r1/objects/ab/cd"]); |
| 318 | // 3 pages of caches (list + delete each), then a page for each other prefix |
| 319 | expect(m.stats.ops).toBe(6 + 4 * 2); |
| 320 | // stops within its budget and says so |
| 321 | for (let i = 0; i < 3000; i++) await m.bucket.put(`logs/r3/${i}`, new Uint8Array(1)); |
| 322 | expect(await deletePrefixes(m.bucket, transientPrefixes("r3"), 4)).toBe(false); |
| 323 | }); |
| 324 | }); |
| 325 | |
| 326 | describe("streaming bodies of unknown length", () => { |
| 327 | const stream = (chunks: number[]) => |
| 328 | new ReadableStream<Uint8Array>({ |
| 329 | start(ctl) { |
| 330 | chunks.forEach((n, i) => ctl.enqueue(filled(n, i))); |
| 331 | ctl.close(); |
| 332 | }, |
| 333 | }); |
| 334 | |
| 335 | it("stores small bodies in one put and large ones as parts, a part in memory at a time", async () => { |
| 336 | const m = memoryBucket(); |
| 337 | expect(await putStream(m.bucket, "a", stream([10, 20]), 100)).toBe(30); |
| 338 | expect(m.stats.multipart).toBe(0); |
| 339 | const MiB = 1024 * 1024; |
| 340 | 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); |
| 341 | expect(m.stats.parts).toEqual([5 * MiB, 5 * MiB, 2 * MiB + 1]); |
| 342 | expect(m.objects.get("b")!.data.length).toBe(12 * MiB + 1); |
| 343 | }); |
| 344 | |
| 345 | it("refuses more than the limit and stores nothing", async () => { |
| 346 | const m = memoryBucket(); |
| 347 | await expect(putStream(m.bucket, "c", stream([60, 60]), 100)).rejects.toBeInstanceOf(BodyTooLarge); |
| 348 | await expect(putStream(m.bucket, "d", stream([6 * 1024 * 1024, 6 * 1024 * 1024]), 10 * 1024 * 1024, undefined, 5 * 1024 * 1024)).rejects.toBeInstanceOf(BodyTooLarge); |
| 349 | expect(m.objects.has("c")).toBe(false); |
| 350 | expect(m.objects.has("d")).toBe(false); |
| 351 | }); |
| 352 | }); |
| 353 | |
| 354 | describe("event payloads", () => { |
| 355 | const meta: RepoMeta = { repoId: "r1", ownerId: "o1", owner: "acme", name: "app", defaultBranch: "main", ownerKind: "org" }; |
| 356 | const site = { publicUrl: "https://d.example", meta }; |
| 357 | |
| 358 | it("adds repository, sender and organization like GitHub", () => { |
| 359 | const p = completePayload(site, { ref: "refs/heads/main", sender: { login: "ada" } }, { id: "u1", login: "ada", kind: "user" }); |
| 360 | const repo = p.repository as Record<string, unknown>; |
| 361 | 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" }); |
| 362 | expect(repo.owner).toMatchObject({ login: "acme", type: "Organization" }); |
| 363 | expect(p.sender).toMatchObject({ login: "ada", id: numericId("u1"), type: "User" }); |
| 364 | expect(p.organization).toMatchObject({ login: "acme" }); |
| 365 | expect(p.ref).toBe("refs/heads/main"); |
| 366 | }); |
| 367 | |
| 368 | it("completes pull_request with head.repo and keeps trigger fields", () => { |
| 369 | 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" }); |
| 370 | const pr = p.pull_request as Record<string, any>; |
| 371 | expect(pr.head).toMatchObject({ ref: "feat", sha: "abc", label: "acme:feat" }); |
| 372 | expect(pr.head.repo.full_name).toBe("acme/app"); |
| 373 | expect(pr.base.repo.full_name).toBe("acme/app"); |
| 374 | // Numeric ids, as on GitHub and in the REST API. |
| 375 | expect(pr.head.repo.id).toBe(numericId("r1")); |
| 376 | expect(pr.base.repo.id).toBe(numericId("r1")); |
| 377 | expect(typeof pr.user.id).toBe("number"); |
| 378 | expect((p.organization as Record<string, unknown>).id).toBe(numericId("o1")); |
| 379 | expect(pr).toMatchObject({ number: 3, title: "T", state: "open", html_url: "https://d.example/acme/app/pull/3" }); |
| 380 | expect(pr.user.login).toBe("bob"); |
| 381 | expect((p.sender as Record<string, unknown>).type).toBe("Bot"); |
| 382 | }); |
| 383 | |
| 384 | it("builds issue and release objects", () => { |
| 385 | 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 }); |
| 386 | expect(issue).toMatchObject({ number: 5, title: "Bug", labels: [{ name: "bug" }], comments: 2, html_url: "https://d.example/acme/app/issues/5" }); |
| 387 | expect(issue.pull_request).toBeUndefined(); |
| 388 | 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" } }); |
| 389 | 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}" }); |
| 390 | }); |
| 391 | }); |
| 392 | |
| 393 | /** The Durable Object's SqlStorage over node:sqlite: exec(query, ...bindings) with toArray() and one(). */ |
| 394 | function sqlStorage() { |
| 395 | const db = new DatabaseSync(":memory:"); |
| 396 | return { |
| 397 | exec(query: string, ...args: unknown[]) { |
| 398 | const params = args.map((a) => (a === undefined ? null : a)) as Array<string | number | null>; |
| 399 | let rows: unknown[] = []; |
| 400 | if (!params.length && query.trim().replace(/;\s*$/, "").includes(";")) db.exec(query); |
| 401 | else rows = db.prepare(query).all(...params); |
| 402 | return { |
| 403 | toArray: () => rows, |
| 404 | one: () => { |
| 405 | if (rows.length !== 1) throw new Error(`expected one row, got ${rows.length}`); |
| 406 | return rows[0]; |
| 407 | }, |
| 408 | }; |
| 409 | }, |
| 410 | }; |
| 411 | } |
| 412 | |
| 413 | /** A Runtime over a fresh repository schema, an in-memory bucket, and jobs running on given refs. */ |
| 414 | function runtimeFixture() { |
| 415 | const sql = sqlStorage(); |
| 416 | sql.exec(REPO_SCHEMA); |
| 417 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 418 | const m = memoryBucket(); |
| 419 | const repo = { sql, meta: { repoId: "r", defaultBranch: "main" } } as unknown as RepoDO & { runtime: Runtime }; |
| 420 | const env = { BUCKET: m.bucket } as unknown as Env; |
| 421 | const rt = new Runtime(repo, env); |
| 422 | repo.runtime = rt; |
| 423 | // The runner API's side of older runners' caches and artifacts, with every job token taken as valid. |
| 424 | const actions = new Actions(repo, env); |
| 425 | Object.defineProperty(actions, "authJob", { value: async (id: string) => sql.exec("SELECT * FROM jobs WHERE id = ?", id).one() }); |
| 426 | let n = 0; |
| 427 | /** A running job of a run on `ref`; `payload` is the run's event payload. */ |
| 428 | const job = (ref: string, payload: Record<string, unknown> = {}) => { |
| 429 | n++; |
| 430 | const t = new Date().toISOString(); |
| 431 | sql.exec( |
| 432 | "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', ?, ?, ?)", |
| 433 | `run${n}`, |
| 434 | n, |
| 435 | ref, |
| 436 | t, |
| 437 | t, |
| 438 | JSON.stringify(payload), |
| 439 | ); |
| 440 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES (?, ?, 'j', 'j', 'in_progress')", `job${n}`, `run${n}`); |
| 441 | return `job${n}`; |
| 442 | }; |
| 443 | const version = "ab".repeat(32); |
| 444 | /** Saves `key` from `jobId` (reserve, then commit). */ |
| 445 | const save = async (jobId: string, key: string) => { |
| 446 | const r = await rt.cacheReserve(jobId, key, version); |
| 447 | expect(r).not.toBeNull(); |
| 448 | await rt.cacheCommit(jobId, r!.id, 10, "etag"); |
| 449 | }; |
| 450 | const restore = (jobId: string, ...keys: string[]) => rt.cacheLookup(jobId, keys, version); |
| 451 | return { sql, rt, actions, m, job, save, restore }; |
| 452 | } |
| 453 | |
| 454 | describe("cache scopes (as on GitHub)", () => { |
| 455 | it("restores from the job's own ref, then the default branch; never another branch's", async () => { |
| 456 | const f = runtimeFixture(); |
| 457 | const main = f.job("refs/heads/main"); |
| 458 | await f.save(main, "deps-main"); |
| 459 | const a = f.job("refs/heads/a"); |
| 460 | await f.save(a, "deps-a"); |
| 461 | // a restores its own entry first, else main's |
| 462 | expect(f.restore(a, "deps-a")?.scope).toBe("refs/heads/a"); |
| 463 | expect(f.restore(a, "nothing", "deps-")?.key).toBe("deps-a"); |
| 464 | expect(f.restore(a, "deps-main")?.scope).toBe("refs/heads/main"); |
| 465 | // b sees main's, not a's newer one, even by prefix |
| 466 | const b = f.job("refs/heads/b"); |
| 467 | expect(f.restore(b, "deps-a")).toBeNull(); |
| 468 | expect(f.restore(b, "deps-")?.key).toBe("deps-main"); |
| 469 | // the default branch never sees a branch's caches |
| 470 | const main2 = f.job("refs/heads/main"); |
| 471 | expect(f.restore(main2, "deps-a")).toBeNull(); |
| 472 | expect(f.restore(main2, "deps-")?.key).toBe("deps-main"); |
| 473 | // saves go to the job's own ref: b may save a key that a already has |
| 474 | await f.save(b, "deps-a"); |
| 475 | expect(f.restore(b, "deps-a")?.scope).toBe("refs/heads/b"); |
| 476 | }); |
| 477 | |
| 478 | it("gives pull requests their base branch's caches but not other pull requests'", async () => { |
| 479 | const f = runtimeFixture(); |
| 480 | await f.save(f.job("refs/heads/main"), "k-main"); |
| 481 | await f.save(f.job("refs/heads/release"), "k-release"); |
| 482 | const pr1 = f.job("refs/pull/1/merge", { pull_request: { number: 1, base: { ref: "release" } } }); |
| 483 | expect(f.restore(pr1, "k-")?.key).toBe("k-release"); |
| 484 | await f.save(pr1, "k-pr1"); |
| 485 | expect(f.restore(pr1, "k-")?.key).toBe("k-pr1"); |
| 486 | const pr2 = f.job("refs/pull/2/merge", { pull_request: { number: 2, base: { ref: "main" } } }); |
| 487 | expect(f.restore(pr2, "k-pr1")).toBeNull(); |
| 488 | expect(f.restore(pr2, "k-release")).toBeNull(); |
| 489 | expect(f.restore(pr2, "k-")?.key).toBe("k-main"); |
| 490 | // nor the base branch or the default branch theirs |
| 491 | expect(f.restore(f.job("refs/heads/release"), "k-pr1")).toBeNull(); |
| 492 | expect(f.restore(f.job("refs/heads/main"), "k-pr")).toBeNull(); |
| 493 | // a merge group reads its target branch's caches |
| 494 | const group = f.job("refs/heads/gh-readonly-queue/release/pr-1-abc", { merge_group: { base_ref: "refs/heads/release" } }); |
| 495 | expect(f.restore(group, "k-")?.key).toBe("k-release"); |
| 496 | }); |
| 497 | }); |
| 498 | |
| 499 | describe("artifact retention", () => { |
| 500 | const DAY = 86400_000; |
| 501 | async function upload(f: ReturnType<typeof runtimeFixture>, jobId: string, name: string, expiresAt?: string) { |
| 502 | const { id } = await f.rt.artifactCreate(jobId, name, undefined, expiresAt); |
| 503 | await f.m.bucket.put(finalKey({ repo: "r", kind: "artifact", id }), new Uint8Array(100)); |
| 504 | await f.rt.stage("artifact", id, 100); |
| 505 | return f.rt.artifactFinalize(jobId, name, 100, undefined, "etag"); |
| 506 | } |
| 507 | const days = (iso: string) => Math.round((Date.parse(iso) - Date.now()) / DAY); |
| 508 | |
| 509 | it("honours retention-days, clamped to 1..90 days", async () => { |
| 510 | const f = runtimeFixture(); |
| 511 | const j = f.job("refs/heads/main"); |
| 512 | expect(days((await upload(f, j, "five", new Date(Date.now() + 5 * DAY).toISOString())).expiresAt)).toBe(5); |
| 513 | expect(days((await upload(f, j, "default")).expiresAt)).toBe(90); |
| 514 | expect(days((await upload(f, j, "too-long", new Date(Date.now() + 400 * DAY).toISOString())).expiresAt)).toBe(90); |
| 515 | expect(days((await upload(f, j, "too-short", new Date(Date.now() - DAY).toISOString())).expiresAt)).toBe(1); |
| 516 | expect(days((await upload(f, j, "garbage", "next tuesday")).expiresAt)).toBe(90); |
| 517 | }); |
| 518 | |
| 519 | it("deletes an expired artifact's content and lists it as expired", async () => { |
| 520 | const f = runtimeFixture(); |
| 521 | const j = f.job("refs/heads/main"); |
| 522 | const a = await upload(f, j, "old", new Date(Date.now() + 2 * DAY).toISOString()); |
| 523 | const keep = await upload(f, j, "new", new Date(Date.now() + 3 * DAY).toISOString()); |
| 524 | expect(a.expired).toBe(false); |
| 525 | f.sql.exec("UPDATE rt_artifacts SET expires_at = ? WHERE id = ?", new Date(Date.now() - 1000).toISOString(), a.id); |
| 526 | // Already shown as expired (and not handed to the toolkit) before the sweep runs. |
| 527 | expect(f.rt.artifactGet(a.id)).toBeNull(); |
| 528 | expect(f.rt.artifactGet(a.id, { includeExpired: true })?.expired).toBe(true); |
| 529 | // A job ending sweeps: the content goes, the listing stays. |
| 530 | await f.rt.sweepJob(j); |
| 531 | expect(f.m.objects.has(finalKey({ repo: "r", kind: "artifact", id: a.id }))).toBe(false); |
| 532 | expect(f.m.objects.has(finalKey({ repo: "r", kind: "artifact", id: keep.id }))).toBe(true); |
| 533 | expect(f.rt.runArtifacts("run1").map((x) => x.name)).toEqual(["new"]); |
| 534 | expect(f.rt.runArtifacts("run1", { includeExpired: true }).map((x) => [x.name, x.expired])).toEqual([ |
| 535 | ["new", false], |
| 536 | ["old", true], |
| 537 | ]); |
| 538 | const listed = f.rt.repoArtifacts({}); |
| 539 | expect(listed.artifacts.find((x) => x.id === a.id)?.expired).toBe(true); |
| 540 | // Expired artifacts no longer count against the quota. |
| 541 | expect(f.sql.exec("SELECT status, staged FROM rt_artifacts WHERE id = ?", a.id).one()).toEqual({ status: "expired", staged: 0 }); |
| 542 | // The name can be used again, and 90 days on the listing goes too. |
| 543 | const again = f.job("refs/heads/main"); |
| 544 | f.sql.exec("UPDATE jobs SET run_id = 'run1' WHERE id = ?", again); |
| 545 | expect((await upload(f, again, "old")).expired).toBe(false); |
| 546 | f.sql.exec("UPDATE rt_artifacts SET expires_at = ? WHERE id = ?", new Date(Date.now() - 91 * DAY).toISOString(), keep.id); |
| 547 | await f.rt.expireArtifacts(); |
| 548 | await f.rt.expireArtifacts(); |
| 549 | expect(f.rt.artifactGet(keep.id, { includeExpired: true })).toBeNull(); |
| 550 | }); |
| 551 | |
| 552 | it("a job starting again from the beginning drops what its stopped attempt uploaded, and nothing else", async () => { |
| 553 | const f = runtimeFixture(); |
| 554 | const j = f.job("refs/heads/main"); |
| 555 | const other = f.job("refs/heads/main"); |
| 556 | const mine = await upload(f, j, "report"); |
| 557 | const theirs = await upload(f, other, "report"); |
| 558 | await f.save(j, "deps-1"); |
| 559 | await f.save(other, "deps-2"); |
| 560 | // A job's end keeps what it finished uploading. |
| 561 | await f.rt.sweepJob(j); |
| 562 | expect(f.rt.artifactGet(mine.id)).not.toBeNull(); |
| 563 | // Its restart doesn't: the new attempt uploads its own, under the same names. |
| 564 | await f.rt.sweepJob(j, { restart: true }); |
| 565 | expect(f.rt.artifactGet(mine.id, { includeExpired: true })).toBeNull(); |
| 566 | expect(f.m.objects.has(finalKey({ repo: "r", kind: "artifact", id: mine.id }))).toBe(false); |
| 567 | expect(f.restore(j, "deps-1")).toBeNull(); |
| 568 | expect(f.rt.artifactGet(theirs.id)).not.toBeNull(); |
| 569 | expect(f.restore(other, "deps-2")).not.toBeNull(); |
| 570 | expect((await upload(f, j, "report")).expired).toBe(false); |
| 571 | }); |
| 572 | }); |
| 573 | |
| 574 | describe("artifacts from older runners", () => { |
| 575 | const DAY = 86400_000; |
| 576 | /** What the runner API does with an older runner's upload: check the name, store the tarball, record it. */ |
| 577 | async function legacyUpload(f: ReturnType<typeof runtimeFixture>, jobId: string, name: string, size: number) { |
| 578 | const { r2Keys } = await f.actions.artifactUploadKeys(jobId, "djt_tok", name); |
| 579 | await f.m.bucket.put(r2Keys[0]!, new Uint8Array(size)); |
| 580 | await f.actions.artifactSaved(jobId, "djt_tok", name, size); |
| 581 | return r2Keys; |
| 582 | } |
| 583 | |
| 584 | it("are listed with numeric ids, and can be read and deleted like the others", async () => { |
| 585 | const f = runtimeFixture(); |
| 586 | const j = f.job("refs/heads/main"); |
| 587 | const keys = await legacyUpload(f, j, "logs", 40); |
| 588 | expect(keys).toEqual(legacyArtifactKeys("r", "run1", "logs")); |
| 589 | const listed = f.rt.repoArtifacts({}); |
| 590 | expect(listed.total).toBe(1); |
| 591 | const a = listed.artifacts[0]!; |
| 592 | expect(a).toMatchObject({ name: "logs", size: 40, expired: false, runId: "run1", runNumber: 1, mimeType: "application/gzip" }); |
| 593 | expect(Number.isSafeInteger(a.id) && a.id > 0).toBe(true); |
| 594 | expect(f.rt.runArtifacts("run1", { includeExpired: true }).map((x) => x.id)).toEqual([a.id]); |
| 595 | expect(f.rt.runArtifacts("1", { includeExpired: true }).map((x) => x.id)).toEqual([a.id]); |
| 596 | expect(f.rt.artifactGet(a.id, { includeExpired: true })?.name).toBe("logs"); |
| 597 | // Its content stays where the runner put it, and downloads as a tarball. |
| 598 | expect(artifactObjectKeys("r", a)).toEqual(keys); |
| 599 | expect(artifactFileName(a)).toBe("logs.tar.gz"); |
| 600 | expect(f.actions.artifacts("run1")).toMatchObject([{ id: a.id, name: "logs", r2Keys: keys, fileName: "logs.tar.gz", contentType: "application/gzip" }]); |
| 601 | // It counts against the quota, and is kept 90 days (upload-artifact's default). |
| 602 | expect(await f.rt.artifactRoom()).toBe(DEFAULT_ARTIFACT_QUOTA - 40); |
| 603 | expect(Math.round((Date.parse(a.expiresAt) - Date.now()) / DAY)).toBe(90); |
| 604 | // Uploading the name again replaces it. |
| 605 | await legacyUpload(f, j, "logs", 50); |
| 606 | const again = f.rt.repoArtifacts({}).artifacts; |
| 607 | expect(again.map((x) => [x.name, x.size])).toEqual([["logs", 50]]); |
| 608 | // Deleting it deletes its content. |
| 609 | expect((await f.rt.artifactDelete(again[0]!.id))?.name).toBe("logs"); |
| 610 | expect(f.m.objects.has(keys[0]!)).toBe(false); |
| 611 | expect(f.rt.repoArtifacts({}).total).toBe(0); |
| 612 | expect(await f.rt.artifactRoom()).toBe(DEFAULT_ARTIFACT_QUOTA); |
| 613 | }); |
| 614 | |
| 615 | it("expire like the others", async () => { |
| 616 | const f = runtimeFixture(); |
| 617 | const j = f.job("refs/heads/main"); |
| 618 | const [key] = await legacyUpload(f, j, "old", 10); |
| 619 | const a = f.rt.repoArtifacts({}).artifacts[0]!; |
| 620 | f.sql.exec("UPDATE rt_artifacts SET expires_at = ? WHERE id = ?", new Date(Date.now() - 1000).toISOString(), a.id); |
| 621 | await f.rt.expireArtifacts(); |
| 622 | expect(f.m.objects.has(key!)).toBe(false); |
| 623 | expect(f.rt.artifactGet(a.id, { includeExpired: true })?.expired).toBe(true); |
| 624 | expect(await f.rt.artifactRoom()).toBe(DEFAULT_ARTIFACT_QUOTA); |
| 625 | }); |
| 626 | |
| 627 | it("share names with the artifact service's uploads, which don't list them", async () => { |
| 628 | const f = runtimeFixture(); |
| 629 | const j = f.job("refs/heads/main"); |
| 630 | await legacyUpload(f, j, "tarball", 10); |
| 631 | // The toolkit talks to GitHub's v4 service, which doesn't list v3 artifacts either. |
| 632 | expect(f.rt.jobArtifacts(j)).toEqual([]); |
| 633 | await expect(f.rt.artifactCreate(j, "tarball", undefined, undefined)).rejects.toThrow(/already exists/); |
| 634 | const { id } = await f.rt.artifactCreate(j, "zip", undefined, undefined); |
| 635 | await f.m.bucket.put(finalKey({ repo: "r", kind: "artifact", id }), new Uint8Array(5)); |
| 636 | await f.rt.artifactFinalize(j, "zip", 5, undefined, "etag"); |
| 637 | await expect(f.actions.artifactUploadKeys(j, "djt_tok", "zip")).rejects.toThrow(/already exists/); |
| 638 | }); |
| 639 | |
| 640 | it("recorded before they were listed are moved over, keeping their age", () => { |
| 641 | const f = runtimeFixture(); |
| 642 | f.job("refs/heads/main"); |
| 643 | const created = new Date(Date.now() - 10 * DAY).toISOString(); |
| 644 | f.sql.exec("INSERT INTO artifacts (id, run_id, name, size, created_at) VALUES ('x', 'run1', 'old', 7, ?)", created); |
| 645 | migrateRepoSchema(f.sql as unknown as SqlStorage); |
| 646 | const [a] = f.rt.repoArtifacts({}).artifacts; |
| 647 | expect(a).toMatchObject({ name: "old", size: 7, createdAt: created, legacy: true, expired: false }); |
| 648 | expect(Math.round((Date.parse(a!.expiresAt) - Date.now()) / DAY)).toBe(80); |
| 649 | expect(f.sql.exec("SELECT count(*) AS n FROM artifacts").one()).toEqual({ n: 0 }); |
| 650 | }); |
| 651 | }); |
| 652 | |
| 653 | describe("workflow permission settings", () => { |
| 654 | it("audit each change, marking those made through the API", () => { |
| 655 | const sql = sqlStorage(); |
| 656 | sql.exec(REPO_SCHEMA); |
| 657 | const events: Array<{ action: string; detail: Record<string, unknown> }> = []; |
| 658 | const repo = { sql, meta: { owner: "acme", name: "app" }, emit: (e: { action: string; detail: Record<string, unknown> }) => events.push(e) }; |
| 659 | const policy = new TokenPolicy(repo as unknown as RepoDO); |
| 660 | const admin = { id: "u1", login: "ada", kind: "user", role: "admin" } as Caller; |
| 661 | policy.setSettings({ defaultPermissions: "read" }, admin, { via: "api" }); |
| 662 | // The web settings page calls it without `via`. |
| 663 | policy.setSettings({ canApprovePullRequests: true }, admin); |
| 664 | expect(events.map((e) => [e.action, e.detail])).toEqual([ |
| 665 | ["repo.actions_workflow_settings", { defaultPermissions: "read", canApprovePullRequests: false, agentRunSecrets: false, via: "api" }], |
| 666 | ["repo.actions_workflow_settings", { defaultPermissions: "read", canApprovePullRequests: true, agentRunSecrets: false }], |
| 667 | ]); |
| 668 | }); |
| 669 | }); |
| 670 | |
| 671 | describe("an older runner's cache", () => { |
| 672 | const DAY = 86400_000; |
| 673 | it("drops entries unused for a week, with their objects", async () => { |
| 674 | const f = runtimeFixture(); |
| 675 | const j = f.job("refs/heads/main"); |
| 676 | expect(await f.actions.cacheSaved(j, "djt_tok", "old", 10)).toEqual([]); |
| 677 | expect(await f.actions.cacheSaved(j, "djt_tok", "recent", 10)).toEqual([]); |
| 678 | f.sql.exec("UPDATE caches SET last_used = ? WHERE key = ?", new Date(Date.now() - 8 * DAY).toISOString(), "refs/heads/main\nold"); |
| 679 | f.sql.exec("UPDATE caches SET last_used = ? WHERE key = ?", new Date(Date.now() - 6 * DAY).toISOString(), "refs/heads/main\nrecent"); |
| 680 | expect(await f.actions.cacheSaved(j, "djt_tok", "new", 10)).toEqual(legacyCacheKeys("r", "refs/heads/main\nold")); |
| 681 | expect(f.sql.exec("SELECT key FROM caches ORDER BY key").toArray()).toEqual([{ key: "refs/heads/main\nnew" }, { key: "refs/heads/main\nrecent" }]); |
| 682 | }); |
| 683 | }); |
| 684 | |
| 685 | describe("hosted runner slots", () => { |
| 686 | it("a repository runs at most SHARE hosted jobs at once; the rest wait their turn", async () => { |
| 687 | const sql = sqlStorage(); |
| 688 | sql.exec(REPO_SCHEMA); |
| 689 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 690 | const started: string[] = []; |
| 691 | const repo = { |
| 692 | sql, |
| 693 | meta: { repoId: "r", owner: "acme", name: "app", defaultBranch: "main" }, |
| 694 | upsertCheck: () => {}, |
| 695 | broadcast: () => {}, |
| 696 | deploys: { jobCompleted: () => {} }, |
| 697 | runtime: { sweepJob: async () => {} }, |
| 698 | } as unknown as RepoDO; |
| 699 | const env = { |
| 700 | PUBLIC_URL: "https://d.test", |
| 701 | BUCKET: memoryBucket().bucket, |
| 702 | RUNNER: { idFromName: (n: string) => n, get: (n: string) => ({ startJob: async () => void started.push(n), stopJob: async () => {} }) }, |
| 703 | RUN_WORKFLOW: { get: async () => ({ sendEvent: async () => {} }) }, |
| 704 | } as unknown as Env; |
| 705 | const actions = new Actions(repo, env); |
| 706 | const t = new Date().toISOString(); |
| 707 | 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); |
| 708 | 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}`); |
| 709 | const launch = (id: string) => (actions as unknown as { launch(id: string): Promise<void> }).launch(id); |
| 710 | for (let i = 0; i < SHARE + 2; i++) await launch(`j${i}`); |
| 711 | expect(started).toHaveLength(SHARE); |
| 712 | const held = () => sql.exec("SELECT id FROM jobs WHERE fleet_held = 1 AND status = 'queued' ORDER BY rowid").toArray().map((r) => r.id); |
| 713 | expect(held()).toEqual([`j${SHARE}`, `j${SHARE + 1}`]); |
| 714 | // A finished job hands its slot to the oldest one waiting. |
| 715 | await (actions as unknown as { completeJob(id: string, c: unknown): Promise<void> }).completeJob("j0", { conclusion: "success", outputs: {} }); |
| 716 | expect(started).toHaveLength(SHARE + 1); |
| 717 | expect(started.at(-1)).toBe(`job-j${SHARE}`); |
| 718 | expect(held()).toEqual([`j${SHARE + 1}`]); |
| 719 | }); |
| 720 | |
| 721 | it("a job whose container gets no machine waits and starts again, and fails only after MAX_PLACEMENT_TRIES", async () => { |
| 722 | const sql = sqlStorage(); |
| 723 | sql.exec(REPO_SCHEMA); |
| 724 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 725 | let full = true; |
| 726 | let startError = "there is no container instance that can be provided to this durable object"; |
| 727 | const events: { type: string; payload: Record<string, unknown> }[] = []; |
| 728 | const repo = { |
| 729 | sql, |
| 730 | meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" }, |
| 731 | upsertCheck: () => {}, |
| 732 | broadcast: () => {}, |
| 733 | deploys: { jobCompleted: () => {} }, |
| 734 | runtime: { sweepJob: async () => {} }, |
| 735 | } as unknown as RepoDO; |
| 736 | const DB = testD1(); |
| 737 | DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString()); |
| 738 | const running = () => DB.sqlite.prepare("SELECT id FROM compute_usage WHERE ended_at IS NULL").all(); |
| 739 | const env = { |
| 740 | DB, |
| 741 | PUBLIC_URL: "https://d.test", |
| 742 | BUCKET: memoryBucket().bucket, |
| 743 | RUNNER: { |
| 744 | idFromName: (n: string) => n, |
| 745 | get: () => ({ |
| 746 | startJob: async () => { |
| 747 | if (full) throw new Error(startError); |
| 748 | }, |
| 749 | stopJob: async () => {}, |
| 750 | }), |
| 751 | }, |
| 752 | RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) }, |
| 753 | } as unknown as Env; |
| 754 | const actions = new Actions(repo, env); |
| 755 | const t = new Date().toISOString(); |
| 756 | 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); |
| 757 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j1', 'run1', 'k', 'k', 'queued'), ('j2', 'run1', 'k2', 'k2', 'queued')"); |
| 758 | const launch = (id: string) => (actions as unknown as { launch(id: string): Promise<void> }).launch(id); |
| 759 | const job = (id: string) => sql.exec("SELECT status, conclusion, container, placement_tries FROM jobs WHERE id = ?", id).toArray()[0]; |
| 760 | |
| 761 | // No machine: the job goes back in line and the run's workflow is asked to start it again after a pause. |
| 762 | await launch("j1"); |
| 763 | expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, container: null, placement_tries: 1 }); |
| 764 | expect(events).toEqual([{ type: "job-j1", payload: { retryInSeconds: placementDelay(1) } }]); |
| 765 | // Its log says why it waits. |
| 766 | const log = (id: string) => sql.exec<{ text: string }>("SELECT text FROM logs WHERE job_id = ? ORDER BY seq", id).toArray().map((l) => l.text); |
| 767 | 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}).`]); |
| 768 | // While it waits it isn't running, so it holds none of the account's hosted jobs (lib/quotas.ts). |
| 769 | expect(running()).toEqual([]); |
| 770 | |
| 771 | // Its container stopping before the runner reported in counts the same way; a started job's loss doesn't. |
| 772 | full = false; |
| 773 | await actions.relaunchJob("j1"); |
| 774 | expect(job("j1")).toMatchObject({ status: "waiting", container: "job-j1" }); |
| 775 | expect(running()).toEqual([{ id: "j1" }]); |
| 776 | // The first such stop starts it again after QUICK_RETRY_SECONDS, uncounted; another one waits as usual. |
| 777 | await actions.containerStopped("j1", null); |
| 778 | expect(job("j1")).toMatchObject({ status: "queued", placement_tries: 1 }); |
| 779 | expect(events.at(-1)).toEqual({ type: "job-j1", payload: { retryInSeconds: QUICK_RETRY_SECONDS } }); |
| 780 | expect(log("j1").at(-1)).toBe(`##[warning]No hosted runner could start the job yet: the container stopped before the runner started. Diffinery tries again in ${QUICK_RETRY_SECONDS} s (not counted: the first stop like this is retried at once).`); |
| 781 | expect(running()).toEqual([]); |
| 782 | await actions.relaunchJob("j1"); |
| 783 | expect(job("j1")).toMatchObject({ status: "waiting", container: "job-j1" }); |
| 784 | await actions.containerStopped("j1", 0); |
| 785 | expect(job("j1")).toMatchObject({ status: "queued", placement_tries: 2 }); |
| 786 | expect(events.at(-1)).toEqual({ type: "job-j1", payload: { retryInSeconds: placementDelay(2) } }); |
| 787 | expect(log("j1").at(-1)).toMatch(/: the container stopped before the runner started\. Diffinery tries again in 60 s \(wait 2 of \d+\)\.$/); |
| 788 | expect(running()).toEqual([]); |
| 789 | // The same when the runner exits with a code before it fetched the job; and a container that |
| 790 | // exits after the job went back in line (its token revoked) leaves the job in line. |
| 791 | await actions.relaunchJob("j1"); |
| 792 | await actions.containerStopped("j1", 2); |
| 793 | expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 3 }); |
| 794 | await actions.containerStopped("j1", 2); |
| 795 | expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 3 }); |
| 796 | await launch("j2"); |
| 797 | sql.exec("UPDATE jobs SET status = 'in_progress' WHERE id = 'j2'"); |
| 798 | await actions.containerStopped("j2", 1); |
| 799 | expect(job("j2")).toMatchObject({ status: "completed", conclusion: "failure", placement_tries: 0 }); |
| 800 | |
| 801 | // A container sidecar restarting under the start counts the same way; an error that won't go away fails the job. |
| 802 | full = true; |
| 803 | startError = "Container sidecar is shutting down"; |
| 804 | await actions.relaunchJob("j1"); |
| 805 | expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, placement_tries: 4 }); |
| 806 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j3', 'run1', 'k3', 'k3', 'queued')"); |
| 807 | startError = "image not found"; |
| 808 | await launch("j3"); |
| 809 | expect(job("j3")).toMatchObject({ status: "completed", conclusion: "failure", placement_tries: 0 }); |
| 810 | expect(sql.exec("SELECT failure FROM jobs WHERE id = 'j3'").one().failure).toMatch(/Could not start a runner: image not found/); |
| 811 | |
| 812 | // Past the limit it fails, and says why. |
| 813 | startError = "there is no container instance that can be provided to this durable object"; |
| 814 | for (let i = 4; i < MAX_PLACEMENT_TRIES; i++) await actions.relaunchJob("j1"); |
| 815 | expect(job("j1")).toMatchObject({ status: "queued", placement_tries: MAX_PLACEMENT_TRIES }); |
| 816 | await actions.relaunchJob("j1"); |
| 817 | expect(job("j1")).toMatchObject({ status: "completed", conclusion: "failure" }); |
| 818 | expect(sql.exec("SELECT failure FROM jobs WHERE id = 'j1'").one().failure).toMatch(/No hosted runner could be started/); |
| 819 | expect([1, 2, 3, 4, 5, 6].map(placementDelay)).toEqual([30, 60, 120, 240, 300, 300]); |
| 820 | }); |
| 821 | |
| 822 | it("a matrix leg back in line after getting no machine waits out its pause when a sibling finishes", async () => { |
| 823 | const sql = sqlStorage(); |
| 824 | sql.exec(REPO_SCHEMA); |
| 825 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 826 | let full = true; |
| 827 | const started: string[] = []; |
| 828 | const events: { type: string; payload: Record<string, unknown> }[] = []; |
| 829 | const repo = { |
| 830 | sql, |
| 831 | meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" }, |
| 832 | upsertCheck: () => {}, |
| 833 | broadcast: () => {}, |
| 834 | deploys: { jobCompleted: () => {} }, |
| 835 | runtime: { sweepJob: async () => {} }, |
| 836 | } as unknown as RepoDO; |
| 837 | const DB = testD1(); |
| 838 | DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString()); |
| 839 | const env = { |
| 840 | DB, |
| 841 | PUBLIC_URL: "https://d.test", |
| 842 | BUCKET: memoryBucket().bucket, |
| 843 | RUNNER: { |
| 844 | idFromName: (n: string) => n, |
| 845 | get: (n: string) => ({ |
| 846 | startJob: async () => { |
| 847 | started.push(n); |
| 848 | if (full) throw new Error("there is no container instance that can be provided to this durable object"); |
| 849 | }, |
| 850 | stopJob: async () => {}, |
| 851 | }), |
| 852 | }, |
| 853 | RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) }, |
| 854 | } as unknown as Env; |
| 855 | const actions = new Actions(repo, env); |
| 856 | const a = actions as unknown as { launch(id: string): Promise<void>; completeJob(id: string, c: unknown): Promise<void> }; |
| 857 | const t = new Date().toISOString(); |
| 858 | 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); |
| 859 | // Legs of one matrix: m1 and m2 started, m3 not yet (max-parallel). |
| 860 | 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')"); |
| 861 | // m4 is back in line after a safe restart (its runner was replaced before its steps ran). |
| 862 | sql.exec("UPDATE jobs SET restarts = 1 WHERE id = 'm4'"); |
| 863 | const job = (id: string) => sql.exec("SELECT status, placement_tries FROM jobs WHERE id = ?", id).toArray()[0]; |
| 864 | await a.launch("m1"); |
| 865 | expect(job("m1")).toMatchObject({ status: "queued", placement_tries: 1 }); |
| 866 | full = false; |
| 867 | await a.launch("m2"); |
| 868 | expect(job("m2")).toMatchObject({ status: "waiting" }); |
| 869 | started.length = 0; |
| 870 | |
| 871 | // m2 finishing starts the leg that never started, not the ones back in line: those are started |
| 872 | // again by the run's workflow, so each try still waits its turn (placementDelay). |
| 873 | await a.completeJob("m2", { conclusion: "success", outputs: {} }); |
| 874 | expect(started).toEqual(["job-m3"]); |
| 875 | expect(job("m1")).toMatchObject({ status: "queued", placement_tries: 1 }); |
| 876 | expect(job("m4")).toMatchObject({ status: "queued" }); |
| 877 | expect(events.filter((e) => e.type === "job-m1")).toHaveLength(1); |
| 878 | }); |
| 879 | |
| 880 | it("a hosted runner on an image other than the deployed one is turned away and the job placed again, up to STALE_IMAGE_TRIES", async () => { |
| 881 | const sql = sqlStorage(); |
| 882 | sql.exec(REPO_SCHEMA); |
| 883 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 884 | const events: { type: string; payload: Record<string, unknown> }[] = []; |
| 885 | const repo = { |
| 886 | sql, |
| 887 | meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" }, |
| 888 | upsertCheck: () => {}, |
| 889 | broadcast: () => {}, |
| 890 | deploys: { jobCompleted: () => {} }, |
| 891 | runtime: { sweepJob: async () => {} }, |
| 892 | } as unknown as RepoDO; |
| 893 | const DB = testD1(); |
| 894 | DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString()); |
| 895 | const env = { |
| 896 | DB, |
| 897 | PUBLIC_URL: "https://d.test", |
| 898 | RUNNER_IMAGE: "diffinery-runner:new", |
| 899 | RUNNER: { idFromName: (n: string) => n, get: () => ({ startJob: async () => {}, stopJob: async () => {} }) }, |
| 900 | RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) }, |
| 901 | } as unknown as Env; |
| 902 | const actions = new Actions(repo, env); |
| 903 | const t = new Date().toISOString(); |
| 904 | 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); |
| 905 | 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')"); |
| 906 | sql.exec("UPDATE jobs SET labels = '[\"self-hosted\"]' WHERE id = 's1'"); |
| 907 | const a = actions as unknown as { launch(id: string): Promise<void>; refuseStaleRunner(job: unknown, runner: unknown, deployed?: string): Promise<void>; jobRow(id: string): unknown }; |
| 908 | const refuse = (id: string, image?: string, deployed?: string) => a.refuseStaleRunner(a.jobRow(id), image ? { image } : null, deployed); |
| 909 | const job = (id = "j1") => sql.exec("SELECT status, placement_tries FROM jobs WHERE id = ?", id).toArray()[0]; |
| 910 | |
| 911 | await a.launch("j1"); |
| 912 | // The deployed image runs the job. |
| 913 | await expect(refuse("j1", "diffinery-runner:new")).resolves.toBeUndefined(); |
| 914 | expect(job()).toMatchObject({ status: "waiting", placement_tries: 0 }); |
| 915 | // The previous image (or one that doesn't say) is turned away, and the job waits to start again. |
| 916 | for (const [i, image] of ["diffinery-runner:old", undefined, "diffinery-runner:old", "diffinery-runner:old"].entries()) { |
| 917 | await expect(refuse("j1", image)).rejects.toMatchObject({ status: 409 }); |
| 918 | expect(job()).toMatchObject({ status: "queued", placement_tries: i + 1 }); |
| 919 | expect(events.at(-1)).toEqual({ type: "job-j1", payload: { retryInSeconds: placementDelay(i + 1) } }); |
| 920 | await actions.relaunchJob("j1"); |
| 921 | } |
| 922 | // A rollout that doesn't finish doesn't hold the job up past STALE_IMAGE_TRIES. |
| 923 | expect(job()).toMatchObject({ status: "waiting", placement_tries: STALE_IMAGE_TRIES }); |
| 924 | await expect(refuse("j1", "diffinery-runner:old")).resolves.toBeUndefined(); |
| 925 | // Self-hosted runners run any image. |
| 926 | await expect(refuse("s1", "my-image")).resolves.toBeUndefined(); |
| 927 | |
| 928 | // Right after a deploy this object can still run the previous version (its RUNNER_IMAGE is the |
| 929 | // old one) while the Worker that took the runner's request has the new one: the Worker's wins. |
| 930 | await a.launch("j2"); |
| 931 | await expect(refuse("j2", "diffinery-runner:next", "diffinery-runner:next")).resolves.toBeUndefined(); |
| 932 | expect(job("j2")).toMatchObject({ status: "waiting", placement_tries: 0 }); |
| 933 | await expect(refuse("j2", "diffinery-runner:new", "diffinery-runner:next")).rejects.toMatchObject({ status: 409 }); |
| 934 | expect(job("j2")).toMatchObject({ status: "queued", placement_tries: 1 }); |
| 935 | }); |
| 936 | |
| 937 | 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 () => { |
| 938 | const sql = sqlStorage(); |
| 939 | sql.exec(REPO_SCHEMA); |
| 940 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 941 | const events: { type: string; payload: Record<string, unknown> }[] = []; |
| 942 | const stopped: string[] = []; |
| 943 | const repo = { |
| 944 | sql, |
| 945 | meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" }, |
| 946 | upsertCheck: () => {}, |
| 947 | broadcast: () => {}, |
| 948 | deploys: { jobCompleted: () => {} }, |
| 949 | runtime: { sweepJob: async () => {} }, |
| 950 | } as unknown as RepoDO; |
| 951 | const env = { |
| 952 | PUBLIC_URL: "https://d.test", |
| 953 | BUCKET: memoryBucket().bucket, |
| 954 | RUNNER: { idFromName: (n: string) => n, get: (n: string) => ({ startJob: async () => {}, stopJob: async () => void stopped.push(n) }) }, |
| 955 | RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) }, |
| 956 | } as unknown as Env; |
| 957 | const actions = new Actions(repo, env); |
| 958 | const t0 = Date.parse("2026-10-16T12:00:00Z"); |
| 959 | const at = (ms: number) => new Date(t0 + ms).toISOString(); |
| 960 | 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)); |
| 961 | // Each runner fetched its job at t0 (started_at). f: frozen, timeout 10 min; h: healthy, timeout 30; |
| 962 | // d: no timeout of its own (6 hours); s: on a self-hosted runner, frozen too; q: still in line; c: finished. |
| 963 | const add = (id: string, minutes: number | null, o: { status?: string; container?: string | null; started?: string | null } = {}) => |
| 964 | sql.exec( |
| 965 | "INSERT INTO jobs (id, run_id, job_key, name, status, timeout_minutes, container, started_at, strategy) VALUES (?, 'run1', ?, ?, ?, ?, ?, ?, '{\"fail-fast\":false}')", |
| 966 | id, id, id, o.status ?? "in_progress", minutes, o.container === undefined ? `job-${id}` : o.container, o.started === undefined ? at(0) : o.started, |
| 967 | ); |
| 968 | add("f", 10); |
| 969 | add("h", 30); |
| 970 | add("d", 0); |
| 971 | add("s", 10, { container: "self-hosted:box" }); |
| 972 | add("q", 10, { status: "queued", container: null, started: null }); |
| 973 | add("c", 10, { status: "completed" }); |
| 974 | const job = (id: string) => sql.exec<{ status: string; conclusion: string | null; failure: string | null }>("SELECT status, conclusion, failure FROM jobs WHERE id = ?", id).one(); |
| 975 | const min = 60_000; |
| 976 | expect(actions.nextJobDeadline()).toBe(t0 + 10 * min + JOB_DEADLINE_GRACE_MS); |
| 977 | |
| 978 | // Past f's timeout but within the grace its runner has to stop the steps and report: nothing happens. |
| 979 | await actions.enforceJobDeadlines(t0 + 10 * min + JOB_DEADLINE_GRACE_MS - 1); |
| 980 | expect(["f", "h", "d", "s"].map((id) => job(id).status)).toEqual(["in_progress", "in_progress", "in_progress", "in_progress"]); |
| 981 | expect(stopped).toEqual([]); |
| 982 | |
| 983 | // At the deadline the frozen ones end: the hosted container is stopped, both fail as timed out. |
| 984 | await actions.enforceJobDeadlines(t0 + 10 * min + JOB_DEADLINE_GRACE_MS); |
| 985 | for (const id of ["f", "s"]) { |
| 986 | expect(job(id)).toMatchObject({ status: "completed", conclusion: "timed_out" }); |
| 987 | 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.`); |
| 988 | } |
| 989 | expect(stopped).toEqual(["job-f"]); |
| 990 | expect(events.filter((e) => e.type === "job-f")).toEqual([{ type: "job-f", payload: { conclusion: "timed_out" } }]); |
| 991 | // The others are untouched, and the next wake-up is h's deadline. |
| 992 | expect([job("h").status, job("d").status, job("q").status, job("c").status]).toEqual(["in_progress", "in_progress", "queued", "completed"]); |
| 993 | expect(actions.nextJobDeadline()).toBe(t0 + 30 * min + JOB_DEADLINE_GRACE_MS); |
| 994 | // A job without timeout-minutes gets GitHub's 6 hours. |
| 995 | await actions.enforceJobDeadlines(t0 + 360 * min + JOB_DEADLINE_GRACE_MS - 1); |
| 996 | expect(job("d").status).toBe("in_progress"); |
| 997 | await actions.enforceJobDeadlines(t0 + 360 * min + JOB_DEADLINE_GRACE_MS); |
| 998 | expect(job("d")).toMatchObject({ status: "completed", conclusion: "timed_out" }); |
| 999 | expect(actions.nextJobDeadline()).toBeNull(); |
| 1000 | }); |
| 1001 | |
| 1002 | 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 () => { |
| 1003 | const sql = sqlStorage(); |
| 1004 | sql.exec(REPO_SCHEMA); |
| 1005 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 1006 | const events: { type: string; payload: Record<string, unknown> }[] = []; |
| 1007 | const swept: [string, boolean][] = []; |
| 1008 | const repo = { |
| 1009 | sql, |
| 1010 | meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" }, |
| 1011 | upsertCheck: () => {}, |
| 1012 | broadcast: () => {}, |
| 1013 | deploys: { jobCompleted: () => {} }, |
| 1014 | runtime: { sweepJob: async (id: string, o?: { restart?: boolean }) => void swept.push([id, !!o?.restart]), grantFor: () => ({ role: "read", permissions: {} }) }, |
| 1015 | policy: { settings: () => ({ defaultPermissions: "write" }), isForkPull: () => false }, |
| 1016 | } as unknown as RepoDO; |
| 1017 | const DB = testD1(); |
| 1018 | DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString()); |
| 1019 | const started: { container: string; token: string }[] = []; |
| 1020 | const env = { |
| 1021 | DB, |
| 1022 | PUBLIC_URL: "https://d.test", |
| 1023 | BUCKET: memoryBucket().bucket, |
| 1024 | RUNNER_IMAGE: "diffinery-runner:new", |
| 1025 | RUNNER: { idFromName: (n: string) => n, get: () => ({ startJob: async (i: { container: string; token: string }) => void started.push(i), stopJob: async () => {} }) }, |
| 1026 | RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) }, |
| 1027 | } as unknown as Env; |
| 1028 | const actions = new Actions(repo, env); |
| 1029 | // The job's steps: checkout, then one whose effects running it twice would repeat. |
| 1030 | // 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. |
| 1031 | (actions as unknown as { expanded: () => Promise<unknown> }).expanded = async () => { |
| 1032 | const steps = [{ uses: "actions/checkout@v4" }, { run: "make deploy" }]; |
| 1033 | const read = { contents: "read" }; |
| 1034 | const install = [{ uses: "actions/checkout@v4" }, { run: "pnpm install" }, { run: "pnpm test" }]; |
| 1035 | const jobs: Record<string, unknown> = { |
| 1036 | k: { steps }, |
| 1037 | k2: { steps }, |
| 1038 | k3: { steps }, |
| 1039 | k4: { steps }, |
| 1040 | r1: { steps: install, permissions: read, env: { GH_TOKEN: "${{ secrets.GITHUB_TOKEN }}" } }, |
| 1041 | r2: { steps: [...install, { run: "npm publish", env: { NODE_AUTH_TOKEN: "${{ secrets.NPM_TOKEN }}" } }], permissions: read }, |
| 1042 | r3: { steps: install, permissions: read, environment: { name: "production" } }, |
| 1043 | r4: { steps: install, permissions: { contents: "read", "id-token": "write" } }, |
| 1044 | r5: { steps: [...install, { run: "echo ${{ toJSON(secrets) }}" }], permissions: read }, |
| 1045 | }; |
| 1046 | const workflow = { env: {}, jobs }; |
| 1047 | return { workflow, calls: new Map(), parentOf: new Map(), needs: new Map(), workflowOf: new Map(Object.keys(jobs).map((key) => [key, workflow])) }; |
| 1048 | }; |
| 1049 | const t = new Date().toISOString(); |
| 1050 | 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); |
| 1051 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j1', 'run1', 'k', 'k', 'queued'), ('j2', 'run1', 'k2', 'k2', 'queued')"); |
| 1052 | const launch = (id: string) => (actions as unknown as { launch(id: string): Promise<void> }).launch(id); |
| 1053 | const job = (id: string) => sql.exec("SELECT status, conclusion, container, token_hash, steps, restarts, failure FROM jobs WHERE id = ?", id).toArray()[0]; |
| 1054 | const logs = (id: string) => sql.exec("SELECT seq, step, text FROM logs WHERE job_id = ? ORDER BY seq", id).toArray(); |
| 1055 | // The runner fetched the job on `image` and reported these steps as started. |
| 1056 | const running = (id: string, image: string, steps: number[], names = ["Run actions/checkout@v4", "Deploy"]) => { |
| 1057 | const reports = steps.map((index) => ({ index, name: names[index] ?? `Step ${index}`, status: "in_progress", startedAt: t })); |
| 1058 | sql.exec("UPDATE jobs SET status = 'in_progress', started_at = ?, runner_info = ?, steps = ? WHERE id = ?", t, JSON.stringify({ image }), JSON.stringify(reports), id); |
| 1059 | sql.exec("INSERT INTO logs (job_id, seq, t, step, text) VALUES (?, 0, ?, -1, 'set up'), (?, 1, ?, 0, 'checking out')", id, t, id, t); |
| 1060 | }; |
| 1061 | const replaced = { conclusion: "cancelled" as const, outputs: {}, replaced: true }; |
| 1062 | |
| 1063 | // The runner reports the replacement during checkout: the job goes back in line, its log starts over with a line that says why. |
| 1064 | await launch("j1"); |
| 1065 | const first = started.at(-1)!; |
| 1066 | expect(first.container).toBe("job-j1"); |
| 1067 | running("j1", "diffinery-runner:old", [0]); |
| 1068 | await actions.runnerComplete("j1", first.token, replaced); |
| 1069 | expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, container: null, token_hash: null, steps: "[]", restarts: 1 }); |
| 1070 | expect(events.at(-1)).toEqual({ type: "job-j1", payload: { retryInSeconds: 10 } }); |
| 1071 | 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\)/) }]); |
| 1072 | // The replaced runner can't report any more, and its container stopping changes nothing. |
| 1073 | await expect(actions.runnerComplete("j1", first.token, replaced)).rejects.toMatchObject({ status: 401 }); |
| 1074 | await actions.containerStopped("j1", 143, "job-j1"); |
| 1075 | expect(job("j1")).toMatchObject({ status: "queued", restarts: 1 }); |
| 1076 | |
| 1077 | // It starts in a new container; the old one stopping late doesn't touch it. |
| 1078 | await actions.relaunchJob("j1"); |
| 1079 | expect(started.at(-1)!.container).toBe("job-j1-r1"); |
| 1080 | await actions.containerStopped("j1", 143, "job-j1"); |
| 1081 | expect(job("j1")).toMatchObject({ status: "waiting", container: "job-j1-r1" }); |
| 1082 | |
| 1083 | // A container on a replaced image that stops without reporting counts the same. |
| 1084 | running("j1", "diffinery-runner:old", []); |
| 1085 | await actions.containerStopped("j1", 143, "job-j1-r1"); |
| 1086 | expect(job("j1")).toMatchObject({ status: "queued", restarts: 2 }); |
| 1087 | |
| 1088 | // Past MAX_RUNNER_RESTARTS the job fails, and says why. |
| 1089 | await actions.relaunchJob("j1"); |
| 1090 | expect(started.at(-1)!.container).toBe("job-j1-r2"); |
| 1091 | running("j1", "diffinery-runner:old", []); |
| 1092 | await actions.runnerComplete("j1", started.at(-1)!.token, replaced); |
| 1093 | expect(job("j1")).toMatchObject({ status: "completed", conclusion: "failure", restarts: MAX_RUNNER_RESTARTS }); |
| 1094 | 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\./); |
| 1095 | |
| 1096 | // Once a step other than checkout ran, the job isn't started again: it fails and names the step. |
| 1097 | await launch("j2"); |
| 1098 | running("j2", "diffinery-runner:old", [0, 1]); |
| 1099 | await actions.runnerComplete("j2", started.at(-1)!.token, replaced); |
| 1100 | expect(job("j2")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 }); |
| 1101 | 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/); |
| 1102 | |
| 1103 | // A job that can only read starts again even after its steps ran (here the platform stopped |
| 1104 | // `pnpm install`), and what its stopped attempt uploaded goes. Each restart is counted. |
| 1105 | 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')"); |
| 1106 | const install = ["Run actions/checkout@v4", "Run pnpm install", "Run pnpm test"]; |
| 1107 | await launch("r1"); |
| 1108 | running("r1", "diffinery-runner:new", [0, 1], install); |
| 1109 | await actions.runnerComplete("r1", started.at(-1)!.token, replaced); |
| 1110 | expect(job("r1")).toMatchObject({ status: "queued", conclusion: null, steps: "[]", restarts: 1 }); |
| 1111 | expect(swept.at(-1)).toEqual(["r1", true]); |
| 1112 | 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\)/) }]); |
| 1113 | // It still stops after MAX_RUNNER_RESTARTS, wherever the platform stopped it. |
| 1114 | for (let i = 1; i <= MAX_RUNNER_RESTARTS; i++) { |
| 1115 | await actions.relaunchJob("r1"); |
| 1116 | running("r1", "diffinery-runner:new", [0, 1, 2], install); |
| 1117 | await actions.runnerComplete("r1", started.at(-1)!.token, replaced); |
| 1118 | } |
| 1119 | expect(job("r1")).toMatchObject({ status: "completed", conclusion: "failure", restarts: MAX_RUNNER_RESTARTS }); |
| 1120 | // (On the deployed image no deploy replaced it, so the message says the platform stopped it.) |
| 1121 | 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."); |
| 1122 | // 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. |
| 1123 | for (const id of ["r2", "r3", "r4", "r5"]) { |
| 1124 | await launch(id); |
| 1125 | running(id, "diffinery-runner:new", [0, 1, 2], install); |
| 1126 | await actions.runnerComplete(id, started.at(-1)!.token, replaced); |
| 1127 | expect(job(id)).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 }); |
| 1128 | 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/); |
| 1129 | } |
| 1130 | // Right after a deploy this object can still have the previous RUNNER_IMAGE: the image of the |
| 1131 | // Worker that took the report decides whether a deploy replaced the runner. |
| 1132 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('r6', 'run1', 'r1', 'r6', 'queued'), ('r7', 'run1', 'r1', 'r7', 'queued')"); |
| 1133 | 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) { |
| 1134 | await launch(id); |
| 1135 | running(id, "diffinery-runner:next", [0, 1], install); |
| 1136 | await actions.runnerComplete(id, started.at(-1)!.token, replaced, deployed); |
| 1137 | expect(job(id)).toMatchObject({ status: "queued", restarts: 1 }); |
| 1138 | expect(logs(id)[0]!.text).toMatch(new RegExp(`^##\\[warning\\]${says} while step "Run pnpm install" ran`)); |
| 1139 | } |
| 1140 | |
| 1141 | // A job that finished despite the replacement, or a runner on the deployed image that stops, completes as before. |
| 1142 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j3', 'run1', 'k3', 'k3', 'queued'), ('j4', 'run1', 'k4', 'k4', 'queued')"); |
| 1143 | await launch("j3"); |
| 1144 | running("j3", "diffinery-runner:old", [0, 1]); |
| 1145 | await actions.runnerComplete("j3", started.at(-1)!.token, { conclusion: "success", outputs: {}, replaced: true }); |
| 1146 | expect(job("j3")).toMatchObject({ status: "completed", conclusion: "success" }); |
| 1147 | await launch("j4"); |
| 1148 | running("j4", "diffinery-runner:new", []); |
| 1149 | await actions.containerStopped("j4", 137, "job-j4"); |
| 1150 | expect(job("j4")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 }); |
| 1151 | expect(JSON.parse(job("j4")!.failure as string).message).toBe("The runner stopped unexpectedly (exit code 137)."); |
| 1152 | |
| 1153 | // A container on the deployed image that the platform stops or loses mid-job (no exit code, not |
| 1154 | // stopped by Diffinery) counts as a replacement too: a job that can only read starts again... |
| 1155 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('s1', 'run1', 'r1', 's1', 'queued'), ('s2', 'run1', 'k3', 's2', 'queued')"); |
| 1156 | await launch("s1"); |
| 1157 | running("s1", "diffinery-runner:new", [0, 1], install); |
| 1158 | await actions.containerStopped("s1", 0, "job-s1"); |
| 1159 | expect(job("s1")).toMatchObject({ status: "queued", conclusion: null, restarts: 1 }); |
| 1160 | expect(swept.at(-1)).toEqual(["s1", true]); |
| 1161 | 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\)\.$/) }]); |
| 1162 | // ...a late stop of that container changes nothing, and the restarts stop at MAX_RUNNER_RESTARTS. |
| 1163 | await actions.containerStopped("s1", null, "job-s1"); |
| 1164 | expect(job("s1")).toMatchObject({ status: "queued", restarts: 1 }); |
| 1165 | for (let i = 1; i <= MAX_RUNNER_RESTARTS; i++) { |
| 1166 | await actions.relaunchJob("s1"); |
| 1167 | running("s1", "diffinery-runner:new", [0, 1], install); |
| 1168 | await actions.containerStopped("s1", null, started.at(-1)!.container); |
| 1169 | } |
| 1170 | expect(job("s1")).toMatchObject({ status: "completed", conclusion: "failure", restarts: MAX_RUNNER_RESTARTS }); |
| 1171 | 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."); |
| 1172 | // One that could change something outside the run fails, naming the step. |
| 1173 | await launch("s2"); |
| 1174 | running("s2", "diffinery-runner:new", [0, 1]); |
| 1175 | await actions.containerStopped("s2", 0, "job-s2"); |
| 1176 | expect(job("s2")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 }); |
| 1177 | 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/); |
| 1178 | // Diffinery's own stop (a cancel, a timeout, fail-fast) never starts a job again, and a runner that crashes isn't retried. |
| 1179 | // (Each added after the one before failed, so fail-fast doesn't cancel it.) |
| 1180 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('s3', 'run1', 'r1', 's3', 'queued')"); |
| 1181 | await launch("s3"); |
| 1182 | running("s3", "diffinery-runner:new", [0, 1], install); |
| 1183 | await actions.containerStopped("s3", 0, "job-s3", true); |
| 1184 | expect(job("s3")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 }); |
| 1185 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('s4', 'run1', 'r1', 's4', 'queued')"); |
| 1186 | await launch("s4"); |
| 1187 | running("s4", "diffinery-runner:new", [0, 1], install); |
| 1188 | await actions.containerStopped("s4", 1, "job-s4"); |
| 1189 | expect(job("s4")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 }); |
| 1190 | }); |
| 1191 | }); |
| 1192 | |
| 1193 | describe("runner info", () => { |
| 1194 | it("keeps only the runner's short image, location and instance strings", () => { |
| 1195 | 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" }); |
| 1196 | expect(parseRunnerInfo(JSON.stringify({ image: "x".repeat(300) }))?.image).toHaveLength(200); |
| 1197 | for (const bad of [null, "", "not json", "{}", JSON.stringify({ image: 5 }), "x".repeat(2000)]) expect(parseRunnerInfo(bad)).toBeNull(); |
| 1198 | }); |
| 1199 | }); |
| 1200 | |
| 1201 | describe("unfinished runs", () => { |
| 1202 | it("past MAX_ACTIVE_RUNS a dispatch is refused and other events start nothing", async () => { |
| 1203 | const sql = sqlStorage(); |
| 1204 | sql.exec(REPO_SCHEMA); |
| 1205 | migrateRepoSchema(sql as unknown as SqlStorage); |
| 1206 | 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); |
| 1207 | const t = new Date().toISOString(); |
| 1208 | 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); |
| 1209 | const wf = { path: ".github/workflows/w.yml", on: {}, jobs: {} } as never; |
| 1210 | const caller = { id: "u1", login: "ada", kind: "user" as const }; |
| 1211 | await expect(actions.createRun(wf, { event: "workflow_dispatch", ref: "refs/heads/main", sha: "abc", payload: {}, caller, title: "w" })).rejects.toThrow(/500 unfinished workflow runs/); |
| 1212 | expect(await actions.createRun(wf, { event: "push", ref: "refs/heads/main", sha: "abc", payload: {}, caller, title: "w" })).toBeNull(); |
| 1213 | }); |
| 1214 | }); |