| 1 | import { ZERO_OID, diffTrees, fromUtf8, isTreeMode, readCommit, readTree, resolvePath } from "@diffinery/git-core"; |
| 2 | import type { Actor, Conclusion, Json, JobCompletion, JobSpec, LogLine, NeedsResult, StepReport } from "@diffinery/shared"; |
| 3 | import { |
| 4 | evaluateIf, |
| 5 | evaluateValue, |
| 6 | expandMatrix, |
| 7 | interpolate, |
| 8 | isWarning, |
| 9 | jobOrder, |
| 10 | matchesTrigger, |
| 11 | parseWorkflow, |
| 12 | type ExprContext, |
| 13 | type Job, |
| 14 | type StatusFns, |
| 15 | type TriggerEvent, |
| 16 | type Workflow, |
| 17 | } from "@diffinery/workflow-yaml"; |
| 18 | import { decryptSecret, encryptSecret } from "../lib/crypto.js"; |
| 19 | import { Directory } from "../lib/directory.js"; |
| 20 | import { getFirst, headFirst, legacyArtifactKeys, legacyCacheKeys, logKeys } from "../lib/storage-keys.js"; |
| 21 | import { permissionsForJob, type JobPermissions } from "../lib/job-permissions.js"; |
| 22 | import { signRuntimeToken } from "../lib/runtime-auth.js"; |
| 23 | import { ghRepo, numericId } from "../http/gh.js"; |
| 24 | import { HttpError, newId, now, randomToken, sha256 } from "../lib/util.js"; |
| 25 | import type { Env } from "../env.js"; |
| 26 | import type { RepoDO } from "./RepoDO.js"; |
| 27 | import { completePayload } from "./event-payloads.js"; |
| 28 | import { CACHE_IDLE_MS, CACHE_LIMIT_BYTES, artifactFileName, artifactObjectKeys } from "./runtime.js"; |
| 29 | import { atLeast, type Caller } from "./types.js"; |
| 30 | import { isJobCaller } from "./token-policy.js"; |
| 31 | import { ReusableError, expandReusable, localKey, parseCallRef, type Expanded, type Source } from "./reusable.js"; |
| 32 | import { ancestors, combineResults, jobStatus, runnerLabels, type JobNode } from "./job-graph.js"; |
| 33 | |
| 34 | /** How long a sandbox token works after the sandbox starts. */ |
| 35 | export const SANDBOX_TOKEN_TTL_MS = 24 * 3600_000; |
| 36 | |
| 37 | const WORKFLOW_DIRS = [".github/workflows", ".diffinery/workflows"]; |
| 38 | const SUPPORTED_LABEL = /^(ubuntu-.*|linux|diffinery(-.*)?|self-hosted|x64|amd64)$/i; |
| 39 | |
| 40 | export type RunRow = { |
| 41 | id: string; |
| 42 | run_number: number; |
| 43 | workflow_path: string; |
| 44 | workflow_name: string; |
| 45 | event: string; |
| 46 | ref: string; |
| 47 | sha: string; |
| 48 | actor: string; |
| 49 | status: "queued" | "in_progress" | "completed" | "waiting"; |
| 50 | conclusion: Conclusion | "timed_out" | null; |
| 51 | created_at: string; |
| 52 | updated_at: string; |
| 53 | pr_number: number | null; |
| 54 | attempt: number; |
| 55 | event_payload: string; |
| 56 | inputs: string; |
| 57 | concurrency_group: string | null; |
| 58 | display_title: string; |
| 59 | /** Commit the run reports checks on, when it differs from `sha` (pull request runs). */ |
| 60 | head_sha: string | null; |
| 61 | /** 1 when the code the run executes came from an agent; its jobs then act as an agent. */ |
| 62 | agent_origin?: number; |
| 63 | }; |
| 64 | |
| 65 | export type JobRow = { |
| 66 | id: string; |
| 67 | run_id: string; |
| 68 | job_key: string; |
| 69 | name: string; |
| 70 | /** pending: held back by its concurrency group until the group's running job finishes. */ |
| 71 | status: "queued" | "pending" | "waiting" | "in_progress" | "completed"; |
| 72 | conclusion: Conclusion | "timed_out" | null; |
| 73 | matrix: string; |
| 74 | strategy: string; |
| 75 | started_at: string | null; |
| 76 | completed_at: string | null; |
| 77 | outputs: string; |
| 78 | steps: string; |
| 79 | failure: string | null; |
| 80 | token_hash: string | null; |
| 81 | summary: string | null; |
| 82 | container: string | null; |
| 83 | timeout_minutes: number; |
| 84 | /** runs-on labels for self-hosted jobs (JSON array); null for Diffinery-hosted containers. */ |
| 85 | labels: string | null; |
| 86 | /** `continue-on-error` as evaluated for this matrix leg (0 or 1). */ |
| 87 | continue_on_error: number; |
| 88 | /** Job-level `concurrency` group as evaluated for this leg, and its cancel-in-progress (0 or 1). */ |
| 89 | concurrency_group: string | null; |
| 90 | concurrency_cancel: number; |
| 91 | }; |
| 92 | |
| 93 | export type RunInfo = { |
| 94 | id: string; |
| 95 | number: number; |
| 96 | attempt: number; |
| 97 | workflowPath: string; |
| 98 | workflowName: string; |
| 99 | title: string; |
| 100 | event: string; |
| 101 | ref: string; |
| 102 | sha: string; |
| 103 | /** The commit checks are reported on: the pull request head for pull_request runs. */ |
| 104 | headSha: string; |
| 105 | /** Branch the run is about: the pull request's head branch for pull request runs. */ |
| 106 | headBranch: string; |
| 107 | actor: Actor; |
| 108 | status: RunRow["status"]; |
| 109 | conclusion: RunRow["conclusion"]; |
| 110 | createdAt: string; |
| 111 | updatedAt: string; |
| 112 | prNumber: number | null; |
| 113 | }; |
| 114 | |
| 115 | export type JobInfo = { |
| 116 | id: string; |
| 117 | runId: string; |
| 118 | key: string; |
| 119 | name: string; |
| 120 | status: JobRow["status"]; |
| 121 | conclusion: JobRow["conclusion"]; |
| 122 | matrix: Record<string, Json>; |
| 123 | startedAt: string | null; |
| 124 | completedAt: string | null; |
| 125 | outputs: Record<string, string>; |
| 126 | steps: Array<StepReport & { index: number }>; |
| 127 | failure: JobCompletion["failure"] | null; |
| 128 | summary: string | null; |
| 129 | }; |
| 130 | |
| 131 | const toRun = (r: RunRow): RunInfo => ({ |
| 132 | id: r.id, |
| 133 | number: r.run_number, |
| 134 | attempt: r.attempt, |
| 135 | workflowPath: r.workflow_path, |
| 136 | workflowName: r.workflow_name, |
| 137 | title: r.display_title, |
| 138 | event: r.event, |
| 139 | ref: r.ref, |
| 140 | sha: r.sha, |
| 141 | headSha: r.head_sha ?? r.sha, |
| 142 | headBranch: (r.pr_number ? (JSON.parse(r.event_payload) as { pull_request?: { head?: { ref?: string } } }).pull_request?.head?.ref : undefined) ?? r.ref.replace(/^refs\/(heads|tags)\//, ""), |
| 143 | actor: JSON.parse(r.actor), |
| 144 | status: r.status, |
| 145 | conclusion: r.conclusion, |
| 146 | createdAt: r.created_at, |
| 147 | updatedAt: r.updated_at, |
| 148 | prNumber: r.pr_number, |
| 149 | }); |
| 150 | |
| 151 | const toJob = (j: JobRow): JobInfo => ({ |
| 152 | id: j.id, |
| 153 | runId: j.run_id, |
| 154 | key: j.job_key, |
| 155 | name: j.name, |
| 156 | status: j.status, |
| 157 | conclusion: j.conclusion, |
| 158 | matrix: JSON.parse(j.matrix), |
| 159 | startedAt: j.started_at, |
| 160 | completedAt: j.completed_at, |
| 161 | outputs: JSON.parse(j.outputs), |
| 162 | steps: JSON.parse(j.steps), |
| 163 | failure: j.failure ? JSON.parse(j.failure) : null, |
| 164 | summary: j.summary, |
| 165 | }); |
| 166 | |
| 167 | /** |
| 168 | * GitHub Actions compatible run orchestration state. The Cloudflare Workflow |
| 169 | * (`RunWorkflow`) drives a run through these methods; the runner agent inside |
| 170 | * each job's container reports back through the runner* methods. |
| 171 | */ |
| 172 | export class Actions { |
| 173 | constructor( |
| 174 | private repo: RepoDO, |
| 175 | private env: Env, |
| 176 | ) {} |
| 177 | |
| 178 | private get sql() { |
| 179 | return this.repo.sql; |
| 180 | } |
| 181 | |
| 182 | // ---------------------------------------------------------------- workflow discovery |
| 183 | |
| 184 | async loadWorkflows(sha: string): Promise<Array<{ path: string; workflow?: Workflow; errors: string[] }>> { |
| 185 | const commit = await readCommit(this.repo.store, sha); |
| 186 | const out: Array<{ path: string; workflow?: Workflow; errors: string[] }> = []; |
| 187 | for (const dir of WORKFLOW_DIRS) { |
| 188 | const entry = await resolvePath(this.repo.store, commit.tree, dir); |
| 189 | if (!entry || !isTreeMode(entry.mode)) continue; |
| 190 | for (const f of await readTree(this.repo.store, entry.oid)) { |
| 191 | if (!/\.ya?ml$/.test(f.name) || isTreeMode(f.mode)) continue; |
| 192 | const path = `${dir}/${f.name}`; |
| 193 | const o = await this.repo.store.read(f.oid); |
| 194 | if (!o) continue; |
| 195 | const { workflow, errors } = parseWorkflow(fromUtf8(o.content), path); |
| 196 | out.push({ path, workflow, errors: errors.filter((e) => !isWarning(e)).map((e) => `${e.path}${e.line ? `:${e.line}` : ""}: ${e.message}`) }); |
| 197 | } |
| 198 | } |
| 199 | return out; |
| 200 | } |
| 201 | |
| 202 | async listWorkflows(): Promise<Array<{ path: string; name: string; errors: string[]; triggers: string[] }>> { |
| 203 | const sha = await this.repo.resolve("HEAD"); |
| 204 | if (!sha) return []; |
| 205 | return (await this.loadWorkflows(sha)).map((w) => ({ path: w.path, name: w.workflow?.name ?? w.path.split("/").pop()!, errors: w.errors, triggers: w.workflow ? Object.keys(w.workflow.on) : [] })); |
| 206 | } |
| 207 | |
| 208 | /** The `repository` object of event payloads: GitHub's shape, with the repository's real visibility. */ |
| 209 | async repositoryPayload(): Promise<Record<string, unknown>> { |
| 210 | const m = this.repo.meta; |
| 211 | const row = await new Directory(this.env).repoById(m.repoId); |
| 212 | if (row) return ghRepo(this.env.PUBLIC_URL, row); |
| 213 | return { id: numericId(m.repoId), name: m.name, full_name: `${m.owner}/${m.name}`, owner: { login: m.owner }, default_branch: m.defaultBranch, private: true, visibility: "private", html_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}`, clone_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git` }; |
| 214 | } |
| 215 | |
| 216 | // ---------------------------------------------------------------- triggers |
| 217 | |
| 218 | async onPush(ev: { ref: string; before: string; after: string; forced: boolean; created: boolean; deleted: boolean }, caller: Caller): Promise<void> { |
| 219 | if (ev.deleted) return; |
| 220 | const workflows = await this.loadWorkflows(ev.after); |
| 221 | let changedFiles: string[] | undefined; |
| 222 | if (ev.before !== ZERO_OID && this.repo.store.has(ev.before)) { |
| 223 | const [a, b] = await Promise.all([readCommit(this.repo.store, ev.before), readCommit(this.repo.store, ev.after)]); |
| 224 | changedFiles = (await diffTrees(this.repo.store, a.tree, b.tree)).map((c) => c.path); |
| 225 | } |
| 226 | const head = await readCommit(this.repo.store, ev.after); |
| 227 | const payload = { |
| 228 | ref: ev.ref, |
| 229 | before: ev.before, |
| 230 | after: ev.after, |
| 231 | created: ev.created, |
| 232 | deleted: ev.deleted, |
| 233 | forced: ev.forced, |
| 234 | repository: await this.repositoryPayload(), |
| 235 | pusher: { name: caller.login }, |
| 236 | sender: { login: caller.login, type: caller.kind === "agent" ? "Bot" : "User" }, |
| 237 | head_commit: { id: ev.after, message: head.message, timestamp: new Date(head.committer.timestamp * 1000).toISOString(), author: { name: head.author.name, email: head.author.email } }, |
| 238 | commits: [{ id: ev.after, message: head.message }], |
| 239 | }; |
| 240 | if (ev.ref === `refs/heads/${this.repo.meta.defaultBranch}`) await this.syncSchedules(workflows); |
| 241 | for (const w of workflows) { |
| 242 | if (!w.workflow) { |
| 243 | if (w.errors.length) await this.createFailedRun(w.path, "push", ev.ref, ev.after, caller, w.errors); |
| 244 | continue; |
| 245 | } |
| 246 | if (!matchesTrigger(w.workflow, { name: "push", ref: ev.ref, changedFiles })) continue; |
| 247 | await this.createRun(w.workflow, { event: "push", ref: ev.ref, sha: ev.after, payload, caller, title: head.message.split("\n")[0] ?? "" }); |
| 248 | } |
| 249 | } |
| 250 | |
| 251 | /** |
| 252 | * Pull request workflows, like GitHub's: `pull_request` runs the workflow from the test |
| 253 | * merge commit (`refs/pull/<n>/merge`, the head merged into the base; nothing runs while |
| 254 | * they conflict), and `pull_request_target` runs the base branch's own workflow on the |
| 255 | * base commit, so a pull request can never change what runs with the base's secrets. |
| 256 | * Both report their checks on the pull request's head commit. |
| 257 | */ |
| 258 | async onPullRequest(action: string, number: number, caller: Caller): Promise<void> { |
| 259 | const { issue } = this.repo.collab.get(number); |
| 260 | const pull = issue.pull!; |
| 261 | // Once merged, the base already holds the changes: keep the refs as they were and let path filters pass. |
| 262 | const { baseSha, mergeSha } = pull.merged ? { baseSha: this.repo.getRef(`refs/heads/${pull.base}`), mergeSha: null } : await this.repo.collab.refreshPullRefs(number); |
| 263 | const changedFiles = pull.merged ? undefined : (await this.repo.collab.pullFiles(number)).files.map((f) => f.path); |
| 264 | const m = this.repo.meta; |
| 265 | const repository = await this.repositoryPayload(); |
| 266 | const payload = { |
| 267 | action, |
| 268 | number, |
| 269 | pull_request: { |
| 270 | number, |
| 271 | id: number, |
| 272 | state: issue.state, |
| 273 | title: issue.title, |
| 274 | body: issue.body, |
| 275 | draft: pull.draft, |
| 276 | merged: pull.merged, |
| 277 | merged_at: pull.mergedAt, |
| 278 | merged_by: pull.mergedBy ? { login: pull.mergedBy } : null, |
| 279 | merge_commit_sha: pull.merged ? pull.mergeCommit : mergeSha, |
| 280 | user: { login: issue.author.login }, |
| 281 | labels: issue.labels.map((name) => ({ name })), |
| 282 | assignees: issue.assignees.map((login) => ({ login })), |
| 283 | head: { ref: pull.head, sha: pull.headSha, repo: repository }, |
| 284 | base: { ref: pull.base, sha: baseSha, repo: repository }, |
| 285 | html_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/pull/${number}`, |
| 286 | url: `${this.env.PUBLIC_URL}/api/v3/repos/${m.owner}/${m.name}/pulls/${number}`, |
| 287 | }, |
| 288 | repository, |
| 289 | sender: { login: caller.login }, |
| 290 | }; |
| 291 | const common = { payload, caller, prNumber: number, headSha: pull.headSha, headRef: pull.head, baseRef: pull.base, title: issue.title }; |
| 292 | // A merged pull request's runs use its merge commit, as on GitHub. |
| 293 | const prSha = pull.merged ? pull.mergeCommit : mergeSha; |
| 294 | if (prSha) { |
| 295 | for (const w of await this.loadWorkflows(prSha)) { |
| 296 | if (!w.workflow || !matchesTrigger(w.workflow, { name: "pull_request", action, baseRef: pull.base, changedFiles })) continue; |
| 297 | await this.createRun(w.workflow, { ...common, event: "pull_request", ref: `refs/pull/${number}/merge`, sha: prSha }); |
| 298 | } |
| 299 | } |
| 300 | if (baseSha) { |
| 301 | for (const w of await this.loadWorkflows(baseSha)) { |
| 302 | if (!w.workflow || !matchesTrigger(w.workflow, { name: "pull_request_target", action, baseRef: pull.base, changedFiles })) continue; |
| 303 | await this.createRun(w.workflow, { ...common, event: "pull_request_target", ref: `refs/heads/${pull.base}`, sha: baseSha }); |
| 304 | } |
| 305 | } |
| 306 | } |
| 307 | |
| 308 | /** `merge_group` workflows for a merge queue group, run from the group's commit. */ |
| 309 | async onMergeGroup(g: { number: number; base: string; ref: string; sha: string; parent: string; title: string; sender: Actor }): Promise<void> { |
| 310 | const head = await readCommit(this.repo.store, g.sha); |
| 311 | const payload = { |
| 312 | action: "checks_requested", |
| 313 | merge_group: { |
| 314 | head_sha: g.sha, |
| 315 | head_ref: g.ref, |
| 316 | base_sha: g.parent, |
| 317 | base_ref: `refs/heads/${g.base}`, |
| 318 | head_commit: { |
| 319 | id: g.sha, |
| 320 | tree_id: head.tree, |
| 321 | message: head.message, |
| 322 | timestamp: new Date(head.committer.timestamp * 1000).toISOString(), |
| 323 | author: { name: head.author.name, email: head.author.email }, |
| 324 | committer: { name: head.committer.name, email: head.committer.email }, |
| 325 | }, |
| 326 | }, |
| 327 | repository: await this.repositoryPayload(), |
| 328 | sender: { login: g.sender.login }, |
| 329 | }; |
| 330 | for (const w of await this.loadWorkflows(g.sha)) { |
| 331 | if (!w.workflow) { |
| 332 | if (w.errors.length) await this.createFailedRun(w.path, "merge_group", g.ref, g.sha, g.sender, w.errors); |
| 333 | continue; |
| 334 | } |
| 335 | if (!matchesTrigger(w.workflow, { name: "merge_group", action: "checks_requested", baseRef: g.base })) continue; |
| 336 | await this.createRun(w.workflow, { event: "merge_group", ref: g.ref, sha: g.sha, payload, caller: g.sender, title: `Merge group for #${g.number}: ${g.title}`, prNumbers: [g.number] }); |
| 337 | } |
| 338 | } |
| 339 | |
| 340 | /** Cancel unfinished runs on a ref (a merge group branch that is going away). */ |
| 341 | async cancelRunsOnRef(ref: string, reason: string): Promise<void> { |
| 342 | for (const r of this.sql.exec<{ id: string }>("SELECT id FROM runs WHERE ref = ? AND status != 'completed'", ref).toArray()) await this.cancelInternal(r.id, reason); |
| 343 | } |
| 344 | |
| 345 | async dispatch(workflowFile: string, ref: string, inputs: Record<string, unknown>, caller: Caller): Promise<RunInfo> { |
| 346 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "write access is required to run workflows"); |
| 347 | const sha = await this.repo.resolve(ref); |
| 348 | if (!sha) throw new HttpError(422, `unknown ref ${ref}`); |
| 349 | const all = await this.loadWorkflows(sha); |
| 350 | const w = all.find((x) => x.path === workflowFile || x.path.endsWith(`/${workflowFile}`)); |
| 351 | if (!w?.workflow) throw new HttpError(404, `workflow ${workflowFile} not found at ${ref}${w?.errors.length ? `: ${w.errors.join("; ")}` : ""}`); |
| 352 | if (!w.workflow.on.workflow_dispatch) throw new HttpError(422, "workflow does not have a workflow_dispatch trigger"); |
| 353 | const defs = w.workflow.on.workflow_dispatch.inputs ?? {}; |
| 354 | const resolved: Record<string, unknown> = {}; |
| 355 | for (const [k, d] of Object.entries(defs)) { |
| 356 | const v = inputs[k] ?? d.default; |
| 357 | if (d.required && (v === undefined || v === "")) throw new HttpError(422, `input '${k}' is required`); |
| 358 | if (d.type === "choice" && d.options && v !== undefined && !d.options.includes(String(v))) throw new HttpError(422, `input '${k}' must be one of ${d.options.join(", ")}`); |
| 359 | resolved[k] = d.type === "boolean" ? v === true || v === "true" : d.type === "number" ? Number(v) : v; |
| 360 | } |
| 361 | const fullRef = ref.startsWith("refs/") ? ref : this.repo.getRef(`refs/heads/${ref}`) ? `refs/heads/${ref}` : `refs/tags/${ref}`; |
| 362 | const payload = { inputs: resolved, ref: fullRef, repository: await this.repositoryPayload(), sender: { login: caller.login }, workflow: w.path }; |
| 363 | const m = this.repo.meta; |
| 364 | this.repo.emit({ type: "workflow_dispatch", repoId: m.repoId, owner: m.owner, repo: m.name, ref: fullRef, workflow: w.path, inputs: resolved as Record<string, Json>, actor: { id: caller.id, login: caller.login, kind: caller.kind, onBehalfOf: caller.onBehalfOf }, at: now() }); |
| 365 | return (await this.createRun(w.workflow, { event: "workflow_dispatch", ref: fullRef, sha, payload, caller, inputs: resolved, title: w.workflow.name ?? w.path }))!; |
| 366 | } |
| 367 | |
| 368 | async schedule(workflowPath: string, cron: string): Promise<void> { |
| 369 | const sha = await this.repo.resolve("HEAD"); |
| 370 | if (!sha) return; |
| 371 | const w = (await this.loadWorkflows(sha)).find((x) => x.path === workflowPath); |
| 372 | if (!w?.workflow || !matchesTrigger(w.workflow, { name: "schedule", cron })) return; |
| 373 | const system: Caller = { id: "system", login: "diffinery-scheduler", kind: "system", role: "write" }; |
| 374 | await this.createRun(w.workflow, { event: "schedule", ref: this.repo.headRef, sha, payload: { schedule: cron, repository: await this.repositoryPayload() }, caller: system, title: w.workflow.name ?? w.path }); |
| 375 | } |
| 376 | |
| 377 | private async syncSchedules(workflows: Array<{ path: string; workflow?: Workflow }>): Promise<void> { |
| 378 | const repoId = this.repo.meta.repoId; |
| 379 | const stmts = [this.env.DB.prepare("DELETE FROM schedules WHERE repo_id = ?").bind(repoId)]; |
| 380 | for (const w of workflows) for (const s of w.workflow?.on.schedule ?? []) stmts.push(this.env.DB.prepare("INSERT OR IGNORE INTO schedules (repo_id, workflow_path, cron) VALUES (?, ?, ?)").bind(repoId, w.path, s.cron)); |
| 381 | await this.env.DB.batch(stmts); |
| 382 | } |
| 383 | |
| 384 | // ---------------------------------------------------------------- runs |
| 385 | |
| 386 | private nextRunNumber(): number { |
| 387 | this.sql.exec("INSERT INTO counters (name, value) VALUES ('run', 1) ON CONFLICT(name) DO UPDATE SET value = value + 1"); |
| 388 | return this.sql.exec<{ value: number }>("SELECT value FROM counters WHERE name = 'run'").one().value; |
| 389 | } |
| 390 | |
| 391 | private githubContext(run: RunRow, jobKey: string, token: string): Record<string, unknown> { |
| 392 | const m = this.repo.meta; |
| 393 | const payload = JSON.parse(run.event_payload) as Record<string, unknown>; |
| 394 | const actor = JSON.parse(run.actor) as Actor; |
| 395 | const refName = run.ref.replace(/^refs\/(heads|tags)\//, "").replace(/^refs\/pull\/(\d+)\/(head|merge)$/, "$1/merge"); |
| 396 | const pr = payload.pull_request as { head?: { ref?: string }; base?: { ref?: string } } | undefined; |
| 397 | return { |
| 398 | event_name: run.event, |
| 399 | event: payload, |
| 400 | sha: run.sha, |
| 401 | ref: run.ref, |
| 402 | ref_name: refName, |
| 403 | ref_type: run.ref.startsWith("refs/tags/") ? "tag" : "branch", |
| 404 | ref_protected: false, |
| 405 | head_ref: pr?.head?.ref ?? "", |
| 406 | base_ref: pr?.base?.ref ?? "", |
| 407 | repository: `${m.owner}/${m.name}`, |
| 408 | repository_id: String(numericId(m.repoId)), |
| 409 | repository_owner: m.owner, |
| 410 | repositoryUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, |
| 411 | actor: actor.login, |
| 412 | triggering_actor: actor.login, |
| 413 | // Numeric like GitHub's (actions parse it, e.g. download-artifact's run-id); the API accepts it as a run id. |
| 414 | run_id: String(run.run_number), |
| 415 | run_number: String(run.run_number), |
| 416 | run_attempt: String(run.attempt), |
| 417 | workflow: run.workflow_name, |
| 418 | workflow_ref: `${m.owner}/${m.name}/${run.workflow_path}@${run.ref}`, |
| 419 | job: jobKey, |
| 420 | server_url: this.env.PUBLIC_URL, |
| 421 | api_url: `${this.env.PUBLIC_URL}/api/v3`, |
| 422 | graphql_url: `${this.env.PUBLIC_URL}/api/graphql`, |
| 423 | token, |
| 424 | retention_days: "90", |
| 425 | secret_source: "Actions", |
| 426 | }; |
| 427 | } |
| 428 | |
| 429 | private async createFailedRun(path: string, event: string, ref: string, sha: string, caller: Actor, errors: string[]): Promise<void> { |
| 430 | const id = newId(); |
| 431 | const t = now(); |
| 432 | this.sql.exec( |
| 433 | "INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, conclusion, created_at, updated_at, event_payload, display_title) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'completed', 'failure', ?, ?, ?, ?)", |
| 434 | id, |
| 435 | this.nextRunNumber(), |
| 436 | path, |
| 437 | path, |
| 438 | event, |
| 439 | ref, |
| 440 | sha, |
| 441 | JSON.stringify({ id: caller.id, login: caller.login, kind: caller.kind }), |
| 442 | t, |
| 443 | t, |
| 444 | "{}", |
| 445 | `Invalid workflow file: ${errors[0] ?? path}`, |
| 446 | ); |
| 447 | const jobId = newId(); |
| 448 | this.sql.exec( |
| 449 | "INSERT INTO jobs (id, run_id, job_key, name, status, conclusion, started_at, completed_at, failure) VALUES (?, ?, 'invalid', 'Invalid workflow file', 'completed', 'failure', ?, ?, ?)", |
| 450 | jobId, |
| 451 | id, |
| 452 | t, |
| 453 | t, |
| 454 | JSON.stringify({ message: errors.join("\n") }), |
| 455 | ); |
| 456 | } |
| 457 | |
| 458 | async createRun( |
| 459 | wf: Workflow, |
| 460 | o: { |
| 461 | event: string; |
| 462 | ref: string; |
| 463 | sha: string; |
| 464 | headSha?: string; |
| 465 | payload: Record<string, unknown>; |
| 466 | caller: Caller | { id: string; login: string; kind: Actor["kind"] }; |
| 467 | prNumber?: number; |
| 468 | inputs?: Record<string, unknown>; |
| 469 | headRef?: string; |
| 470 | baseRef?: string; |
| 471 | title: string; |
| 472 | /** Started by a run whose code came from an agent (a re-run, or a workflow_run follower). */ |
| 473 | agentOrigin?: boolean; |
| 474 | /** Pull requests whose code the run includes (a merge group's). */ |
| 475 | prNumbers?: number[]; |
| 476 | }, |
| 477 | ): Promise<RunInfo | null> { |
| 478 | // Like GitHub, what a job does with its GITHUB_TOKEN (pushes, pull requests) starts no new |
| 479 | // runs, so workflows cannot trigger themselves in a loop; workflow_dispatch and |
| 480 | // repository_dispatch are the exceptions. |
| 481 | if (isJobCaller(o.caller) && o.event !== "workflow_dispatch" && o.event !== "repository_dispatch") return null; |
| 482 | const id = newId(); |
| 483 | const t = now(); |
| 484 | const run: RunRow = { |
| 485 | id, |
| 486 | run_number: this.nextRunNumber(), |
| 487 | workflow_path: wf.path, |
| 488 | workflow_name: wf.name ?? wf.path, |
| 489 | event: o.event, |
| 490 | ref: o.ref, |
| 491 | sha: o.sha, |
| 492 | actor: JSON.stringify({ id: o.caller.id, login: o.caller.login, kind: o.caller.kind }), |
| 493 | status: "queued", |
| 494 | conclusion: null, |
| 495 | created_at: t, |
| 496 | updated_at: t, |
| 497 | pr_number: o.prNumber ?? null, |
| 498 | attempt: 1, |
| 499 | event_payload: JSON.stringify(completePayload({ publicUrl: this.env.PUBLIC_URL, meta: this.repo.meta, visibility: (o.payload.repository as { visibility?: string } | undefined)?.visibility === "public" ? "public" : "private" }, o.payload, o.caller)), |
| 500 | inputs: JSON.stringify(o.inputs ?? {}), |
| 501 | concurrency_group: null, |
| 502 | display_title: o.title, |
| 503 | head_sha: o.headSha && o.headSha !== o.sha ? o.headSha : null, |
| 504 | agent_origin: this.repo.policy.agentOrigin({ |
| 505 | caller: o.caller, |
| 506 | ref: o.ref, |
| 507 | prNumbers: [...(o.prNumber !== undefined ? [o.prNumber] : []), ...(o.prNumbers ?? [])], |
| 508 | inherited: o.agentOrigin, |
| 509 | }) |
| 510 | ? 1 |
| 511 | : 0, |
| 512 | }; |
| 513 | const ctx: ExprContext = { github: this.githubContext(run, "", ""), inputs: o.inputs ?? {}, vars: this.variables() }; |
| 514 | try { |
| 515 | if (wf.runName) run.display_title = interpolate(wf.runName, ctx); |
| 516 | if (wf.concurrency) run.concurrency_group = interpolate(wf.concurrency.group, ctx); |
| 517 | } catch (e) { |
| 518 | console.warn("run-name/concurrency evaluation failed", e); |
| 519 | } |
| 520 | this.sql.exec( |
| 521 | "INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, pr_number, attempt, event_payload, inputs, concurrency_group, display_title, head_sha, agent_origin) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", |
| 522 | run.id, |
| 523 | run.run_number, |
| 524 | run.workflow_path, |
| 525 | run.workflow_name, |
| 526 | run.event, |
| 527 | run.ref, |
| 528 | run.sha, |
| 529 | run.actor, |
| 530 | run.status, |
| 531 | run.created_at, |
| 532 | run.updated_at, |
| 533 | run.pr_number, |
| 534 | run.attempt, |
| 535 | run.event_payload, |
| 536 | run.inputs, |
| 537 | run.concurrency_group, |
| 538 | run.display_title, |
| 539 | run.head_sha, |
| 540 | run.agent_origin ?? 0, |
| 541 | ); |
| 542 | if (run.concurrency_group && wf.concurrency) { |
| 543 | const cancelInProgress = evaluateValue(wf.concurrency.cancelInProgress, ctx); |
| 544 | const others = this.sql.exec<{ id: string }>("SELECT id FROM runs WHERE concurrency_group = ? AND status != 'completed' AND id != ?", run.concurrency_group, id).toArray(); |
| 545 | for (const other of others) if (cancelInProgress === true || cancelInProgress === "true") await this.cancelInternal(other.id, "cancelled by a newer run in the same concurrency group"); |
| 546 | } |
| 547 | await this.env.RUN_WORKFLOW.create({ id, params: { repoId: this.repo.meta.repoId, runId: id } }); |
| 548 | const m = this.repo.meta; |
| 549 | this.repo.emit({ type: "workflow_run", action: "requested", repoId: m.repoId, owner: m.owner, repo: m.name, runId: id, actor: JSON.parse(run.actor), at: t }); |
| 550 | this.repo.broadcast("runs", { action: "created", runId: id }); |
| 551 | await this.onWorkflowRun(run, "requested"); |
| 552 | return toRun(run); |
| 553 | } |
| 554 | |
| 555 | /** A run by id, or by its number (github.run_id; the latest attempt). */ |
| 556 | private runRow(id: string): RunRow { |
| 557 | const r = /^\d+$/.test(id) |
| 558 | ? this.sql.exec<RunRow>("SELECT * FROM runs WHERE run_number = ? ORDER BY attempt DESC, created_at DESC LIMIT 1", Number(id)).toArray()[0] |
| 559 | : this.sql.exec<RunRow>("SELECT * FROM runs WHERE id = ?", id).toArray()[0]; |
| 560 | if (!r) throw new HttpError(404, "run not found"); |
| 561 | return r; |
| 562 | } |
| 563 | |
| 564 | private jobRow(id: string): JobRow { |
| 565 | const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", id).toArray()[0]; |
| 566 | if (!j) throw new HttpError(404, "job not found"); |
| 567 | return j; |
| 568 | } |
| 569 | |
| 570 | listRuns(opts: { workflow?: string; branch?: string; status?: string; event?: string; sha?: string; prNumber?: number; limit?: number; offset?: number } = {}): { runs: RunInfo[]; total: number } { |
| 571 | const where: string[] = []; |
| 572 | const args: unknown[] = []; |
| 573 | if (opts.workflow) { |
| 574 | where.push("(workflow_path = ? OR workflow_path LIKE ?)"); |
| 575 | args.push(opts.workflow, `%/${opts.workflow}`); |
| 576 | } |
| 577 | if (opts.branch) { |
| 578 | where.push("ref = ?"); |
| 579 | args.push(`refs/heads/${opts.branch}`); |
| 580 | } |
| 581 | if (opts.status) { |
| 582 | where.push(opts.status === "completed" || opts.status === "queued" || opts.status === "in_progress" ? "status = ?" : "conclusion = ?"); |
| 583 | args.push(opts.status); |
| 584 | } |
| 585 | if (opts.event) { |
| 586 | where.push("event = ?"); |
| 587 | args.push(opts.event); |
| 588 | } |
| 589 | if (opts.sha) { |
| 590 | where.push("(sha = ? OR head_sha = ?)"); |
| 591 | args.push(opts.sha, opts.sha); |
| 592 | } |
| 593 | if (opts.prNumber) { |
| 594 | where.push("pr_number = ?"); |
| 595 | args.push(opts.prNumber); |
| 596 | } |
| 597 | const w = where.length ? `WHERE ${where.join(" AND ")}` : ""; |
| 598 | const total = this.sql.exec<{ n: number }>(`SELECT count(*) AS n FROM runs ${w}`, ...args).one().n; |
| 599 | const rows = this.sql.exec<RunRow>(`SELECT * FROM runs ${w} ORDER BY created_at DESC LIMIT ? OFFSET ?`, ...args, Math.min(opts.limit ?? 25, 100), opts.offset ?? 0).toArray(); |
| 600 | return { runs: rows.map(toRun), total }; |
| 601 | } |
| 602 | |
| 603 | getRun(id: string): RunInfo & { jobs: JobInfo[] } { |
| 604 | const run = this.runRow(id); |
| 605 | return { ...toRun(run), jobs: this.jobs(run.id) }; |
| 606 | } |
| 607 | |
| 608 | jobs(runId: string): JobInfo[] { |
| 609 | if (/^\d+$/.test(runId)) runId = this.runRow(runId).id; |
| 610 | return this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? ORDER BY rowid", runId).toArray().map(toJob); |
| 611 | } |
| 612 | |
| 613 | getJob(id: string): JobInfo { |
| 614 | return toJob(this.jobRow(id)); |
| 615 | } |
| 616 | |
| 617 | async logs(jobId: string, afterSeq = -1): Promise<{ lines: Array<LogLine & { seq: number }>; complete: boolean }> { |
| 618 | const j = this.jobRow(jobId); |
| 619 | const rows = this.sql.exec<LogLine & { seq: number }>("SELECT seq, t, step, text FROM logs WHERE job_id = ? AND seq > ? ORDER BY seq LIMIT 5000", jobId, afterSeq).toArray(); |
| 620 | if (rows.length === 0 && j.status === "completed") { |
| 621 | const archived = await getFirst(this.env.BUCKET, logKeys(this.repo.meta.repoId, jobId)); |
| 622 | if (archived && afterSeq < 0) return { lines: (await archived.json()) as Array<LogLine & { seq: number }>, complete: true }; |
| 623 | } |
| 624 | return { lines: rows, complete: j.status === "completed" }; |
| 625 | } |
| 626 | |
| 627 | private logKey(jobId: string) { |
| 628 | return logKeys(this.repo.meta.repoId, jobId)[0]; |
| 629 | } |
| 630 | |
| 631 | // ---------------------------------------------------------------- orchestration (called by RunWorkflow) |
| 632 | |
| 633 | /** |
| 634 | * The run's job graph: each job with the jobs it waits for (reusable workflow calls |
| 635 | * expanded). RunWorkflow starts a job as soon as its own needs have finished. |
| 636 | */ |
| 637 | async plan(runId: string): Promise<{ jobs: JobNode[]; error?: string; concurrencyBlocked: boolean }> { |
| 638 | const run = this.runRow(runId); |
| 639 | if (run.status === "completed") return { jobs: [], error: "run already completed", concurrencyBlocked: false }; |
| 640 | let exp: Expanded; |
| 641 | try { |
| 642 | exp = await this.expanded(run); |
| 643 | } catch (e) { |
| 644 | return { jobs: [], error: (e as Error).message, concurrencyBlocked: false }; |
| 645 | } |
| 646 | const { order, errors } = jobOrder(exp.workflow); |
| 647 | if (errors.length) return { jobs: [], error: errors.map((e) => e.message).join("; "), concurrencyBlocked: false }; |
| 648 | const blocked = |
| 649 | !!run.concurrency_group && |
| 650 | this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM runs WHERE concurrency_group = ? AND status = 'in_progress' AND id != ?", run.concurrency_group, runId).one().n > 0; |
| 651 | if (!blocked) this.sql.exec("UPDATE runs SET status = 'in_progress', updated_at = ? WHERE id = ? AND status = 'queued'", now(), runId); |
| 652 | return { jobs: order.flat().map((key) => ({ key, needs: exp.workflow.jobs[key]!.needs })), concurrencyBlocked: blocked }; |
| 653 | } |
| 654 | |
| 655 | /** The run's workflow with reusable workflow calls expanded (see ./reusable.ts). */ |
| 656 | private async expanded(run: RunRow): Promise<Expanded> { |
| 657 | const wf = (await this.loadWorkflows(run.sha)).find((w) => w.path === run.workflow_path); |
| 658 | if (!wf?.workflow) throw new ReusableError(`workflow file not found or invalid: ${wf?.errors.join("; ") || run.workflow_path}`); |
| 659 | return expandReusable(wf.workflow, { repoId: null, sha: run.sha }, (uses, from) => this.loadCalled(uses, from)); |
| 660 | } |
| 661 | |
| 662 | /** Load a called workflow: `./path` from the same commit, or `owner/repo/path@ref` from a repository on this instance. */ |
| 663 | private async loadCalled(uses: string, from: Source): Promise<{ workflow: Workflow; source: Source }> { |
| 664 | const ref = parseCallRef(uses); |
| 665 | const m = this.repo.meta; |
| 666 | let source: Source; |
| 667 | let text: string | null; |
| 668 | if (ref.kind === "local" && from.repoId === null) { |
| 669 | source = from; |
| 670 | const b = await this.repo.blob(from.sha, ref.path); |
| 671 | text = b && !b.binary ? fromUtf8(b.content) : null; |
| 672 | } else { |
| 673 | let repoId: string; |
| 674 | if (ref.kind === "local") repoId = from.repoId!; |
| 675 | else if (ref.owner.toLowerCase() === m.owner.toLowerCase() && ref.repo.toLowerCase() === m.name.toLowerCase()) repoId = m.repoId; |
| 676 | else { |
| 677 | const dir = new Directory(this.env); |
| 678 | // A transferred or renamed repository's workflows are found under the old name too. |
| 679 | const target = (await dir.repo(ref.owner, ref.repo)) ?? (await dir.movedRepo(ref.owner, ref.repo)); |
| 680 | // Like GitHub: public workflows can be shared; private ones only within the same owner. |
| 681 | if (!target || (target.visibility === "private" && target.owner_id !== m.ownerId)) throw new Error(`repository ${ref.owner}/${ref.repo} not found, or its workflows are not shared with ${m.owner}/${m.name}`); |
| 682 | repoId = target.id; |
| 683 | } |
| 684 | if (repoId === m.repoId) { |
| 685 | const sha = ref.kind === "local" ? from.sha : await this.repo.resolve(ref.ref); |
| 686 | if (!sha) throw new Error(`unknown ref ${ref.kind === "remote" ? ref.ref : ""}`); |
| 687 | source = { repoId: null, sha }; |
| 688 | const b = await this.repo.blob(sha, ref.path); |
| 689 | text = b && !b.binary ? fromUtf8(b.content) : null; |
| 690 | } else { |
| 691 | const stub = this.env.REPO.get(this.env.REPO.idFromName(repoId)); |
| 692 | const sha = ref.kind === "local" ? from.sha : await stub.resolve(ref.ref); |
| 693 | if (!sha) throw new Error(`unknown ref ${ref.kind === "remote" ? ref.ref : ""} in ${ref.kind === "remote" ? `${ref.owner}/${ref.repo}` : "the called repository"}`); |
| 694 | source = { repoId, sha }; |
| 695 | const b = await stub.blob(sha, ref.path); |
| 696 | text = b && !b.binary ? fromUtf8(b.content) : null; |
| 697 | } |
| 698 | } |
| 699 | if (text === null) throw new Error(`${uses} not found`); |
| 700 | const { workflow, errors } = parseWorkflow(text, ref.path); |
| 701 | const fatal = errors.filter((e) => !isWarning(e)); |
| 702 | if (!workflow || fatal.length) throw new Error(`${uses} is invalid: ${fatal.map((e) => `${e.path}: ${e.message}`).join("; ")}`); |
| 703 | return { workflow, source }; |
| 704 | } |
| 705 | |
| 706 | /** `needs` context for a job or call: keys are as written inside its own workflow. */ |
| 707 | private async needsResults(run: RunRow, exp: Expanded, keys: string[], scope: string | null): Promise<Record<string, NeedsResult>> { |
| 708 | const out: Record<string, NeedsResult> = {}; |
| 709 | for (const key of keys) out[localKey(key, scope)] = await this.resultOf(run, exp, key); |
| 710 | return out; |
| 711 | } |
| 712 | |
| 713 | /** Status functions for the `if:` of a job or call: from the results of all its ancestors, like GitHub. */ |
| 714 | private statusOf(run: RunRow, exp: Expanded, key: string): StatusFns { |
| 715 | const results = ancestors(exp.needs, exp.needs.get(key) ?? []).map((k) => combineResults(this.rowsOf(run, exp, k))); |
| 716 | return jobStatus(results, run.conclusion === "cancelled"); |
| 717 | } |
| 718 | |
| 719 | /** The job rows of a job, or of every job inside a reusable workflow call. */ |
| 720 | private rowsOf(run: RunRow, exp: Expanded, key: string): Array<JobRow & { continueOnError: boolean }> { |
| 721 | const rows = exp.calls.has(key) |
| 722 | ? this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND substr(job_key, 1, ?) = ?", run.id, key.length + 1, `${key}/`).toArray() |
| 723 | : this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND job_key = ?", run.id, key).toArray(); |
| 724 | return rows.map((r) => ({ ...r, continueOnError: !!r.continue_on_error })); |
| 725 | } |
| 726 | |
| 727 | /** Result and outputs of a job, or of a reusable workflow call (from its `on.workflow_call.outputs`). */ |
| 728 | private async resultOf(run: RunRow, exp: Expanded, key: string): Promise<NeedsResult> { |
| 729 | const call = exp.calls.get(key); |
| 730 | const rows = this.rowsOf(run, exp, key); |
| 731 | const result = combineResults(rows); |
| 732 | let outputs: Record<string, string> = {}; |
| 733 | if (!call) for (const r of rows) Object.assign(outputs, JSON.parse(r.outputs)); |
| 734 | if (call && Object.keys(call.workflow.on.workflow_call?.outputs ?? {}).length) { |
| 735 | const jobs: Record<string, NeedsResult> = {}; |
| 736 | for (const k of Object.keys(call.workflow.jobs)) jobs[k] = await this.resultOf(run, exp, `${key}/${k}`); |
| 737 | const { inputs } = await this.callContext(run, exp, key, false); |
| 738 | const ctx: ExprContext = { github: this.githubContext(run, localKey(key, call.parent), ""), jobs, inputs, vars: this.variables() }; |
| 739 | outputs = {}; |
| 740 | for (const [name, d] of Object.entries(call.workflow.on.workflow_call!.outputs!)) { |
| 741 | try { |
| 742 | const v = evaluateValue(d.value, ctx); |
| 743 | outputs[name] = v === undefined || v === null ? "" : typeof v === "object" ? JSON.stringify(v) : String(v); |
| 744 | } catch { |
| 745 | outputs[name] = ""; |
| 746 | } |
| 747 | } |
| 748 | } |
| 749 | return { result, outputs }; |
| 750 | } |
| 751 | |
| 752 | /** |
| 753 | * `inputs` and `secrets` inside a reusable workflow call (or at the top level when |
| 754 | * `callKey` is null): inputs from the caller's `with` and the declared defaults, |
| 755 | * secrets from `secrets: inherit` or the caller's mapping. |
| 756 | */ |
| 757 | private async callContext(run: RunRow, exp: Expanded, callKey: string | null, withSecrets: boolean): Promise<{ inputs: Record<string, unknown>; secrets: Record<string, string> }> { |
| 758 | if (callKey === null) return { inputs: JSON.parse(run.inputs), secrets: withSecrets ? { ...(await this.orgSecrets()), ...(await this.repoSecrets()) } : {} }; |
| 759 | const call = exp.calls.get(callKey)!; |
| 760 | const parent = await this.callContext(run, exp, call.parent, withSecrets); |
| 761 | const ctx: ExprContext = { github: this.githubContext(run, localKey(callKey, call.parent), ""), needs: await this.needsResults(run, exp, call.needs, call.parent), inputs: parent.inputs, vars: this.variables() }; |
| 762 | const inputs: Record<string, unknown> = {}; |
| 763 | for (const [k, d] of Object.entries(call.workflow.on.workflow_call?.inputs ?? {})) { |
| 764 | let v = call.job.with && k in call.job.with ? evaluateValue(call.job.with[k], ctx) : evaluateValue(d.default, ctx); |
| 765 | if (v === undefined || v === null || v === "") { |
| 766 | if (d.required) throw new ReusableError(`jobs.${callKey}: input '${k}' is required by ${call.job.uses}`); |
| 767 | v = d.type === "boolean" ? false : d.type === "number" ? 0 : ""; |
| 768 | } |
| 769 | inputs[k] = d.type === "boolean" ? v === true || v === "true" : d.type === "number" ? Number(v) : v; |
| 770 | } |
| 771 | const secrets: Record<string, string> = {}; |
| 772 | if (withSecrets) { |
| 773 | if (call.job.secrets === "inherit") Object.assign(secrets, parent.secrets); |
| 774 | else for (const [name, expr] of Object.entries(call.job.secrets ?? {})) secrets[name.toUpperCase()] = String(evaluateValue(expr, { ...ctx, secrets: parent.secrets }) ?? ""); |
| 775 | for (const [name, d] of Object.entries(call.workflow.on.workflow_call?.secrets ?? {})) { |
| 776 | if (d.required && !secrets[name.toUpperCase()]) throw new ReusableError(`jobs.${callKey}: secret '${name}' is required by ${call.job.uses}`); |
| 777 | } |
| 778 | } |
| 779 | return { inputs, secrets }; |
| 780 | } |
| 781 | |
| 782 | /** |
| 783 | * Start every matrix instance of a job. Returns the ids the workflow must wait |
| 784 | * for (instances beyond max-parallel are queued and started as others finish). |
| 785 | */ |
| 786 | async startJobKey(runId: string, jobKey: string): Promise<{ jobIds: string[] }> { |
| 787 | const run = this.runRow(runId); |
| 788 | const exp = await this.expanded(run); |
| 789 | const job = exp.workflow.jobs[jobKey]!; |
| 790 | const owner = exp.workflowOf.get(jobKey)!; |
| 791 | const parent = exp.parentOf.get(jobKey) ?? null; |
| 792 | const chain: string[] = []; |
| 793 | for (let c = parent; c; c = exp.calls.get(c)!.parent) chain.unshift(c); |
| 794 | const prefix = chain.map((c) => exp.calls.get(c)!.job.name ?? localKey(c, exp.calls.get(c)!.parent)).join(" / "); |
| 795 | const label = (name: string) => (prefix ? `${prefix} / ${name}` : name); |
| 796 | const fail = (name: string, message: string, conclusion: Conclusion = "failure") => { |
| 797 | const id = newId(); |
| 798 | const t = now(); |
| 799 | this.sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status, conclusion, started_at, completed_at, failure) VALUES (?, ?, ?, ?, 'completed', ?, ?, ?, ?)", id, runId, jobKey, name, conclusion, t, t, conclusion === "failure" ? JSON.stringify({ message }) : null); |
| 800 | this.repo.broadcast(`run:${runId}`, { action: "job", jobId: id }); |
| 801 | return { jobIds: [] as string[] }; |
| 802 | }; |
| 803 | const short = localKey(jobKey, parent); |
| 804 | let needs: Record<string, NeedsResult>; |
| 805 | let baseCtx: ExprContext; |
| 806 | let shouldRun: boolean; |
| 807 | try { |
| 808 | // The callers decide first: a skipped call, or one whose ancestors failed or were skipped, skips everything inside it. |
| 809 | for (const c of chain) { |
| 810 | const call = exp.calls.get(c)!; |
| 811 | const callNeeds = await this.needsResults(run, exp, call.needs, call.parent); |
| 812 | const outer = await this.callContext(run, exp, call.parent, false); |
| 813 | const callCtx: ExprContext = { github: this.githubContext(run, localKey(c, call.parent), ""), needs: callNeeds, inputs: outer.inputs, vars: this.variables() }; |
| 814 | if (!evaluateIf(call.job.if, callCtx, this.statusOf(run, exp, c))) return fail(label(job.name ?? short), "", "skipped"); |
| 815 | } |
| 816 | // Resolve the call's inputs and secrets now, so a missing required one fails here rather than in the runner. |
| 817 | if (parent) await this.callContext(run, exp, parent, true); |
| 818 | needs = await this.needsResults(run, exp, exp.needs.get(jobKey) ?? [], parent); |
| 819 | baseCtx = { github: this.githubContext(run, short, ""), needs, inputs: (await this.callContext(run, exp, parent, false)).inputs, vars: this.variables(), env: owner.env }; |
| 820 | shouldRun = evaluateIf(job.if, baseCtx, this.statusOf(run, exp, jobKey)); |
| 821 | } catch (e) { |
| 822 | return fail(label(job.name ?? short), e instanceof ReusableError ? e.message : `Error evaluating 'if': ${(e as Error).message}`); |
| 823 | } |
| 824 | if (!shouldRun) return fail(label(job.name ?? short), "", "skipped"); |
| 825 | |
| 826 | let combos: Array<Record<string, unknown>>; |
| 827 | try { |
| 828 | combos = expandMatrix(job.strategy?.matrix, baseCtx); |
| 829 | } catch (e) { |
| 830 | return fail(label(job.name ?? short), `Error expanding matrix: ${(e as Error).message}`); |
| 831 | } |
| 832 | if (combos.length > 256) return fail(label(job.name ?? short), "A matrix may produce at most 256 jobs."); |
| 833 | const failFast = evaluateValue(job.strategy?.failFast ?? true, baseCtx) !== false; |
| 834 | const maxParallel = Number(evaluateValue(job.strategy?.maxParallel ?? combos.length, baseCtx)) || combos.length; |
| 835 | const ids: string[] = []; |
| 836 | combos.forEach((matrix, i) => { |
| 837 | // Like GitHub, name, runs-on, timeout-minutes, continue-on-error and concurrency are evaluated for each matrix leg. |
| 838 | const ctx = { ...baseCtx, matrix, strategy: { "fail-fast": failFast, "job-index": i, "job-total": combos.length, "max-parallel": maxParallel } }; |
| 839 | let name = short; |
| 840 | if (!job.name && Object.keys(matrix).length) name = `${short} (${Object.values(matrix).map((v) => (typeof v === "object" ? JSON.stringify(v) : String(v))).join(", ")})`; |
| 841 | let leg: { labels: string[]; timeout: number; continueOnError: boolean; group: string | null; cancel: boolean }; |
| 842 | try { |
| 843 | if (job.name) name = interpolate(job.name, ctx); |
| 844 | const continueOnError = evaluateValue(job.continueOnError, ctx); |
| 845 | const cancel = job.concurrency ? evaluateValue(job.concurrency.cancelInProgress, ctx) : false; |
| 846 | leg = { |
| 847 | labels: runnerLabels(job.runsOn, ctx), |
| 848 | timeout: Number(evaluateValue(job.timeoutMinutes, ctx)) || 360, |
| 849 | continueOnError: continueOnError === true || continueOnError === "true", |
| 850 | group: (job.concurrency ? interpolate(job.concurrency.group, ctx).trim() : "") || null, |
| 851 | cancel: cancel === true || cancel === "true", |
| 852 | }; |
| 853 | } catch (e) { |
| 854 | fail(label(name), `Error evaluating the job for this matrix combination: ${(e as Error).message}`); |
| 855 | return; |
| 856 | } |
| 857 | const unsupported = leg.labels.filter((l) => !SUPPORTED_LABEL.test(l)); |
| 858 | const selfHosted = leg.labels.some((l) => l.toLowerCase() === "self-hosted"); |
| 859 | if (!leg.labels.length || (unsupported.length && !selfHosted)) { |
| 860 | fail(label(name), `No Diffinery runner matches runs-on: ${unsupported.join(", ") || "(no labels)"}. Linux (ubuntu-*) runners are hosted; anything else needs runs-on: [self-hosted, ...] and a registered runner.`); |
| 861 | return; |
| 862 | } |
| 863 | const id = newId(); |
| 864 | this.sql.exec( |
| 865 | "INSERT INTO jobs (id, run_id, job_key, name, status, matrix, strategy, timeout_minutes, labels, continue_on_error, concurrency_group, concurrency_cancel) VALUES (?, ?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?)", |
| 866 | id, |
| 867 | runId, |
| 868 | jobKey, |
| 869 | label(name), |
| 870 | JSON.stringify(matrix), |
| 871 | JSON.stringify(ctx.strategy), |
| 872 | Math.min(leg.timeout, 360), |
| 873 | selfHosted ? JSON.stringify(leg.labels.map((l) => l.toLowerCase())) : null, |
| 874 | leg.continueOnError ? 1 : 0, |
| 875 | leg.group, |
| 876 | leg.cancel ? 1 : 0, |
| 877 | ); |
| 878 | ids.push(id); |
| 879 | }); |
| 880 | for (const id of ids.slice(0, maxParallel)) await this.launch(id); |
| 881 | this.repo.broadcast(`run:${runId}`, { action: "jobs" }); |
| 882 | return { jobIds: ids }; |
| 883 | } |
| 884 | |
| 885 | private checkName(run: RunRow, job: JobRow): string { |
| 886 | return `${run.workflow_name} / ${job.name}`; |
| 887 | } |
| 888 | |
| 889 | private async launch(jobId: string): Promise<void> { |
| 890 | const job = this.jobRow(jobId); |
| 891 | if (job.status === "completed" || !(await this.concurrencySlot(job))) return; |
| 892 | const run = this.runRow(job.run_id); |
| 893 | const m = this.repo.meta; |
| 894 | if (job.labels) { |
| 895 | // Self-hosted: park the job until a registered runner with matching labels claims it. |
| 896 | this.sql.exec("UPDATE jobs SET status = 'waiting', container = NULL WHERE id = ?", jobId); |
| 897 | this.repo.upsertCheck({ sha: run.head_sha ?? run.sha, name: this.checkName(run, job), status: "queued", runId: run.id, jobId, detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}` }); |
| 898 | return; |
| 899 | } |
| 900 | const token = `djt_${m.repoId}_${jobId}_${randomToken(24)}`; |
| 901 | const container = `job-${jobId}`; |
| 902 | this.sql.exec("UPDATE jobs SET status = 'waiting', token_hash = ?, container = ? WHERE id = ?", await sha256(token), container, jobId); |
| 903 | this.repo.upsertCheck({ sha: run.head_sha ?? run.sha, name: this.checkName(run, job), status: "queued", runId: run.id, jobId, detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}` }); |
| 904 | try { |
| 905 | const stub = this.env.RUNNER.get(this.env.RUNNER.idFromName(container)); |
| 906 | await stub.startJob({ repoId: m.repoId, jobId, token, apiUrl: this.env.PUBLIC_URL }); |
| 907 | } catch (e) { |
| 908 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `Could not start a runner: ${(e as Error).message}` } }); |
| 909 | } |
| 910 | } |
| 911 | |
| 912 | /** |
| 913 | * Job-level `concurrency`, like GitHub: at most one job of a group runs at a time in |
| 914 | * the repository. With cancel-in-progress the newcomer cancels the group's other jobs; |
| 915 | * without it the newcomer waits as `pending` and replaces any older pending job. |
| 916 | * Returns whether the job may start now. |
| 917 | */ |
| 918 | private async concurrencySlot(job: JobRow): Promise<boolean> { |
| 919 | if (!job.concurrency_group) return true; |
| 920 | const others = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE concurrency_group = ? AND id != ? AND status IN ('pending', 'waiting', 'in_progress') ORDER BY rowid", job.concurrency_group, job.id).toArray(); |
| 921 | const pending = others.filter((o) => o.status === "pending"); |
| 922 | const running = others.filter((o) => o.status !== "pending"); |
| 923 | const failure = { message: `Canceling since a higher priority waiting request for '${job.concurrency_group}' exists` }; |
| 924 | if (running.length && !job.concurrency_cancel) { |
| 925 | this.sql.exec("UPDATE jobs SET status = 'pending' WHERE id = ?", job.id); |
| 926 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId: job.id }); |
| 927 | for (const o of pending) await this.completeJob(o.id, { conclusion: "cancelled", outputs: {}, failure }); |
| 928 | return false; |
| 929 | } |
| 930 | // Take the group before cancelling anything, so a job that starts meanwhile waits for this one. |
| 931 | this.sql.exec("UPDATE jobs SET status = 'waiting' WHERE id = ?", job.id); |
| 932 | for (const o of [...pending, ...running]) { |
| 933 | await this.stopContainer(o); |
| 934 | await this.completeJob(o.id, { conclusion: "cancelled", outputs: {}, failure }); |
| 935 | } |
| 936 | return true; |
| 937 | } |
| 938 | |
| 939 | async jobTimedOut(jobId: string): Promise<void> { |
| 940 | const j = this.jobRow(jobId); |
| 941 | if (j.status === "completed") return; |
| 942 | await this.stopContainer(j); |
| 943 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The job exceeded its timeout of ${j.timeout_minutes} minutes.` } }, "timed_out"); |
| 944 | } |
| 945 | |
| 946 | async containerStopped(jobId: string, exitCode: number | null): Promise<void> { |
| 947 | const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", jobId).toArray()[0]; |
| 948 | if (!j || j.status === "completed") return; |
| 949 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The runner stopped unexpectedly (exit code ${exitCode ?? "unknown"}).` } }); |
| 950 | } |
| 951 | |
| 952 | /** |
| 953 | * A self-hosted runner asks for work. Returns the oldest waiting job whose |
| 954 | * runs-on labels are all offered by the runner, with a fresh job token. |
| 955 | */ |
| 956 | async claimJob(runnerName: string, runnerLabels: string[], caller: Caller): Promise<{ jobId: string; token: string; runId: string; name: string } | null> { |
| 957 | if (!atLeast(caller.role, "admin")) throw new HttpError(403, "admin access is required to run self-hosted jobs"); |
| 958 | const offered = new Set(["self-hosted", ...runnerLabels.map((l) => l.toLowerCase())]); |
| 959 | const waiting = this.sql |
| 960 | .exec<JobRow>("SELECT j.* FROM jobs j JOIN runs r ON r.id = j.run_id WHERE j.status = 'waiting' AND j.labels IS NOT NULL AND j.container IS NULL AND r.status != 'completed' ORDER BY r.created_at, j.rowid") |
| 961 | .toArray(); |
| 962 | const job = waiting.find((j) => (JSON.parse(j.labels!) as string[]).every((l) => offered.has(l))); |
| 963 | if (!job) return null; |
| 964 | const m = this.repo.meta; |
| 965 | const token = `djt_${m.repoId}_${job.id}_${randomToken(24)}`; |
| 966 | const container = `self-hosted:${runnerName.slice(0, 64)}`; |
| 967 | this.sql.exec("UPDATE jobs SET token_hash = ?, container = ? WHERE id = ? AND container IS NULL", await sha256(token), container, job.id); |
| 968 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId: job.id }); |
| 969 | return { jobId: job.id, token, runId: job.run_id, name: job.name }; |
| 970 | } |
| 971 | |
| 972 | private async stopContainer(j: JobRow): Promise<void> { |
| 973 | if (!j.container || j.container.startsWith("self-hosted:")) return; |
| 974 | try { |
| 975 | await this.env.RUNNER.get(this.env.RUNNER.idFromName(j.container)).stopJob(); |
| 976 | } catch (e) { |
| 977 | console.warn("stop container failed", e); |
| 978 | } |
| 979 | } |
| 980 | |
| 981 | async finishRun(runId: string): Promise<RunInfo> { |
| 982 | const run = this.runRow(runId); |
| 983 | const jobs = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ?", runId).toArray(); |
| 984 | let conclusion: RunRow["conclusion"] = run.conclusion === "cancelled" ? "cancelled" : "success"; |
| 985 | if (conclusion !== "cancelled") { |
| 986 | // A job with continue-on-error may fail without failing the run. |
| 987 | if (jobs.some((j) => (j.conclusion === "failure" || j.conclusion === "timed_out") && !j.continue_on_error)) conclusion = "failure"; |
| 988 | else if (jobs.some((j) => j.conclusion === "cancelled")) conclusion = "cancelled"; |
| 989 | } |
| 990 | const t = now(); |
| 991 | this.sql.exec("UPDATE runs SET status = 'completed', conclusion = ?, updated_at = ? WHERE id = ?", conclusion, t, runId); |
| 992 | const m = this.repo.meta; |
| 993 | this.repo.emit({ type: "workflow_run", action: "completed", repoId: m.repoId, owner: m.owner, repo: m.name, runId, conclusion: conclusion ?? undefined, actor: JSON.parse(run.actor), at: t }); |
| 994 | this.repo.broadcast(`run:${runId}`, { action: "completed", conclusion }); |
| 995 | this.repo.broadcast("runs", { action: "completed", runId }); |
| 996 | await this.onWorkflowRun(this.runRow(runId), "completed"); |
| 997 | return toRun(this.runRow(runId)); |
| 998 | } |
| 999 | |
| 1000 | /** |
| 1001 | * Starts the default branch's `on: workflow_run` workflows that follow `run`. Like |
| 1002 | * GitHub, they run on the default branch, must name the workflows they follow, and |
| 1003 | * chains stop after three levels (A -> B -> C -> D; D's runs trigger nothing). |
| 1004 | */ |
| 1005 | private async onWorkflowRun(run: RunRow, action: "requested" | "completed"): Promise<void> { |
| 1006 | if (this.chainDepth(run) >= 3) return; |
| 1007 | try { |
| 1008 | const sha = this.repo.getRef(this.repo.headRef); |
| 1009 | if (!sha) return; |
| 1010 | const followers = (await this.loadWorkflows(sha)).filter((w) => w.workflow?.on.workflow_run?.workflows?.length); |
| 1011 | if (!followers.length) return; |
| 1012 | const m = this.repo.meta; |
| 1013 | const payload = JSON.parse(run.event_payload) as { pull_request?: { head?: { ref?: string } } }; |
| 1014 | const branch = payload.pull_request?.head?.ref ?? run.ref.replace(/^refs\/(heads|tags)\//, ""); |
| 1015 | const actor = JSON.parse(run.actor) as Actor; |
| 1016 | const workflowRun = { |
| 1017 | id: run.run_number, |
| 1018 | name: run.workflow_name, |
| 1019 | display_title: run.display_title, |
| 1020 | path: run.workflow_path, |
| 1021 | run_number: run.run_number, |
| 1022 | run_attempt: run.attempt, |
| 1023 | event: run.event, |
| 1024 | status: run.status, |
| 1025 | conclusion: run.conclusion, |
| 1026 | head_branch: branch, |
| 1027 | head_sha: run.head_sha ?? run.sha, |
| 1028 | actor: { login: actor.login }, |
| 1029 | triggering_actor: { login: actor.login }, |
| 1030 | pull_requests: run.pr_number ? [{ number: run.pr_number }] : [], |
| 1031 | created_at: run.created_at, |
| 1032 | updated_at: run.updated_at, |
| 1033 | html_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}`, |
| 1034 | }; |
| 1035 | for (const w of followers) { |
| 1036 | if (!matchesTrigger(w.workflow!, { name: "workflow_run", action, workflow: run.workflow_name, branch })) continue; |
| 1037 | await this.createRun(w.workflow!, { |
| 1038 | event: "workflow_run", |
| 1039 | ref: this.repo.headRef, |
| 1040 | sha, |
| 1041 | payload: { action, workflow_run: workflowRun, workflow: { name: run.workflow_name, path: run.workflow_path }, repository: await this.repositoryPayload(), sender: { login: actor.login } }, |
| 1042 | caller: actor, |
| 1043 | title: w.workflow!.name ?? w.path, |
| 1044 | // A follower acts on what the run it follows produced (artifacts, outputs), so it stays an agent's. |
| 1045 | agentOrigin: !!run.agent_origin, |
| 1046 | }); |
| 1047 | } |
| 1048 | } catch (e) { |
| 1049 | console.warn("workflow_run trigger failed", e); |
| 1050 | } |
| 1051 | } |
| 1052 | |
| 1053 | /** How many workflow_run links led to `run` (0 for a run started by any other event). */ |
| 1054 | private chainDepth(run: RunRow): number { |
| 1055 | let depth = 0; |
| 1056 | for (let r: RunRow | undefined = run; r?.event === "workflow_run" && depth < 3; depth++) { |
| 1057 | const parent: string | number | undefined = (JSON.parse(r.event_payload) as { workflow_run?: { id?: string | number } }).workflow_run?.id; |
| 1058 | r = parent !== undefined ? this.sql.exec<RunRow>("SELECT * FROM runs WHERE id = ? OR run_number = ? ORDER BY attempt DESC LIMIT 1", String(parent), Number(parent) || -1).toArray()[0] : undefined; |
| 1059 | } |
| 1060 | return depth; |
| 1061 | } |
| 1062 | |
| 1063 | async plannedError(runId: string, message: string): Promise<void> { |
| 1064 | const t = now(); |
| 1065 | this.sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status, conclusion, started_at, completed_at, failure) VALUES (?, ?, 'setup', 'Set up workflow', 'completed', 'failure', ?, ?, ?)", newId(), runId, t, t, JSON.stringify({ message })); |
| 1066 | } |
| 1067 | |
| 1068 | // ---------------------------------------------------------------- cancellation and re-runs |
| 1069 | |
| 1070 | private async cancelInternal(runId: string, reason: string): Promise<void> { |
| 1071 | const run = this.runRow(runId); |
| 1072 | if (run.status === "completed") return; |
| 1073 | this.sql.exec("UPDATE runs SET conclusion = 'cancelled', updated_at = ? WHERE id = ?", now(), run.id); |
| 1074 | for (const j of this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND status != 'completed'", run.id).toArray()) { |
| 1075 | await this.stopContainer(j); |
| 1076 | await this.completeJob(j.id, { conclusion: "cancelled", outputs: {}, failure: { message: reason } }); |
| 1077 | } |
| 1078 | } |
| 1079 | |
| 1080 | async cancel(runId: string, caller: Caller): Promise<void> { |
| 1081 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "permission denied"); |
| 1082 | await this.cancelInternal(runId, `Cancelled by ${caller.login}`); |
| 1083 | } |
| 1084 | |
| 1085 | async rerun(runId: string, caller: Caller): Promise<RunInfo> { |
| 1086 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "permission denied"); |
| 1087 | if (caller.id.startsWith("job:")) throw new HttpError(403, "a workflow's GITHUB_TOKEN cannot re-run workflows"); |
| 1088 | const old = this.runRow(runId); |
| 1089 | const wf = (await this.loadWorkflows(old.sha)).find((w) => w.path === old.workflow_path)?.workflow; |
| 1090 | if (!wf) throw new HttpError(422, "workflow file no longer valid at this commit"); |
| 1091 | const run = await this.createRun(wf, { |
| 1092 | event: old.event, |
| 1093 | ref: old.ref, |
| 1094 | sha: old.sha, |
| 1095 | headSha: old.head_sha ?? undefined, |
| 1096 | payload: JSON.parse(old.event_payload), |
| 1097 | caller, |
| 1098 | prNumber: old.pr_number ?? undefined, |
| 1099 | inputs: JSON.parse(old.inputs), |
| 1100 | title: old.display_title, |
| 1101 | // Re-running code an agent wrote keeps it acting as the agent, whoever re-runs it. |
| 1102 | agentOrigin: !!old.agent_origin, |
| 1103 | }); |
| 1104 | this.sql.exec("UPDATE runs SET attempt = ?, run_number = ? WHERE id = ?", old.attempt + 1, old.run_number, run!.id); |
| 1105 | return toRun(this.runRow(run!.id)); |
| 1106 | } |
| 1107 | |
| 1108 | // ---------------------------------------------------------------- runner protocol |
| 1109 | |
| 1110 | /** Validate a job or sandbox token. Returns what it grants. */ |
| 1111 | async runnerAuth( |
| 1112 | token: string, |
| 1113 | ): Promise<{ kind: "job"; jobId: string; runId: string; role: Caller["role"]; permissions: JobPermissions; actorKind: Actor["kind"] } | { kind: "sandbox"; sandboxId: string; caller: Caller } | null> { |
| 1114 | const parts = token.split("_"); |
| 1115 | if (parts.length < 4) return null; |
| 1116 | const hash = await sha256(token); |
| 1117 | if (parts[0] === "djt") { |
| 1118 | const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", parts[2]!).toArray()[0]; |
| 1119 | if (!j || j.token_hash !== hash || j.status === "completed") return null; |
| 1120 | const grant = this.repo.runtime.grantFor(j.id); |
| 1121 | // Code an agent wrote, pushed or opened a pull request with runs as that agent (repo/token-policy.ts). |
| 1122 | const run = this.sql.exec<{ agent_origin: number }>("SELECT agent_origin FROM runs WHERE id = ?", j.run_id).toArray()[0]; |
| 1123 | const actorKind: Actor["kind"] = run?.agent_origin ? "agent" : "system"; |
| 1124 | return { kind: "job", jobId: j.id, runId: j.run_id, role: grant.role, permissions: grant.permissions, actorKind }; |
| 1125 | } |
| 1126 | if (parts[0] === "dst") { |
| 1127 | const s = this.sql.exec<{ id: string; secret: string; created_by: string; status: string; created_at: string }>("SELECT * FROM sandboxes WHERE id = ?", parts[2]!).toArray()[0]; |
| 1128 | if (!s || s.secret !== hash || s.status !== "running") return null; |
| 1129 | // A sandbox's token lasts a day at most, even if nobody closes the sandbox. |
| 1130 | if (Date.now() - Date.parse(s.created_at) > SANDBOX_TOKEN_TTL_MS) return null; |
| 1131 | return { kind: "sandbox", sandboxId: s.id, caller: JSON.parse(s.created_by) as Caller }; |
| 1132 | } |
| 1133 | return null; |
| 1134 | } |
| 1135 | |
| 1136 | private async authJob(jobId: string, token: string): Promise<JobRow> { |
| 1137 | const auth = await this.runnerAuth(token); |
| 1138 | if (!auth || auth.kind !== "job" || auth.jobId !== jobId) throw new HttpError(401, "invalid job token"); |
| 1139 | return this.jobRow(jobId); |
| 1140 | } |
| 1141 | |
| 1142 | /** Claims for an OIDC id token; side-effect free (unlike runnerSpec). */ |
| 1143 | async oidcClaims(jobId: string, token: string): Promise<{ allowed: boolean; github: Record<string, string> }> { |
| 1144 | const job = await this.authJob(jobId, token); |
| 1145 | if (job.status !== "in_progress") throw new HttpError(409, "job is not running"); |
| 1146 | const run = this.runRow(job.run_id); |
| 1147 | const exp = await this.expanded(run); |
| 1148 | const def: Job = exp.workflow.jobs[job.job_key]!; |
| 1149 | const wf = exp.workflowOf.get(job.job_key)!; |
| 1150 | const allowed = this.repo.runtime.grantFor(jobId).permissions["id-token"] === "write"; |
| 1151 | const ctx = this.githubContext(run, localKey(job.job_key, exp.parentOf.get(job.job_key) ?? null), token) as Record<string, Json>; |
| 1152 | const github: Record<string, string> = {}; |
| 1153 | for (const [k, v] of Object.entries(ctx)) if (typeof v === "string" || typeof v === "number") github[k] = String(v); |
| 1154 | return { allowed, github }; |
| 1155 | } |
| 1156 | |
| 1157 | /** |
| 1158 | * What a job token may do with container packages (the `packages` permission). |
| 1159 | * A workflow without a permissions block gets the repository's default: read in |
| 1160 | * repositories with read-only tokens (the default for new ones), write in those that |
| 1161 | * keep GitHub's old permissive default. A permissions map that leaves `packages` out gets none. |
| 1162 | */ |
| 1163 | async packagesPermission(token: string): Promise<"none" | "read" | "write"> { |
| 1164 | const auth = await this.runnerAuth(token); |
| 1165 | return auth?.kind === "job" ? auth.permissions.packages : "none"; |
| 1166 | } |
| 1167 | |
| 1168 | /** |
| 1169 | * A pull_request run for a pull request from another repository (a fork): its jobs |
| 1170 | * never get write. Decided from the pull request's row, not the event payload. |
| 1171 | */ |
| 1172 | private untrusted(run: RunRow): boolean { |
| 1173 | return run.event === "pull_request" && this.repo.policy.isForkPull(run.pr_number); |
| 1174 | } |
| 1175 | |
| 1176 | async runnerSpec(jobId: string, token: string): Promise<JobSpec> { |
| 1177 | const job = await this.authJob(jobId, token); |
| 1178 | const run = this.runRow(job.run_id); |
| 1179 | const exp = await this.expanded(run); |
| 1180 | const def: Job = exp.workflow.jobs[job.job_key]!; |
| 1181 | const wf = exp.workflowOf.get(job.job_key)!; |
| 1182 | const parent = exp.parentOf.get(job.job_key) ?? null; |
| 1183 | const short = localKey(job.job_key, parent); |
| 1184 | let scope: { inputs: Record<string, unknown>; secrets: Record<string, string> }; |
| 1185 | try { |
| 1186 | scope = await this.callContext(run, exp, parent, true); |
| 1187 | } catch (e) { |
| 1188 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: (e as Error).message } }); |
| 1189 | throw new HttpError(409, (e as Error).message); |
| 1190 | } |
| 1191 | const m = this.repo.meta; |
| 1192 | // Deployment environment (`environment:`): its branch policy, secrets and variables, and a deployment record. |
| 1193 | let environment: { secrets: Record<string, string>; vars: Record<string, string> } = { secrets: {}, vars: {} }; |
| 1194 | if (def.environment) { |
| 1195 | try { |
| 1196 | const ctx = { github: this.githubContext(run, short, ""), inputs: scope.inputs, vars: this.variables(), matrix: JSON.parse(job.matrix), strategy: JSON.parse(job.strategy), needs: await this.needsResults(run, exp, exp.needs.get(job.job_key) ?? [], parent) }; |
| 1197 | const name = interpolate(def.environment.name, ctx); |
| 1198 | const url = def.environment.url && !def.environment.url.includes("steps.") ? interpolate(def.environment.url, ctx) : undefined; |
| 1199 | const logUrl = `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}`; |
| 1200 | // Code an agent wrote deploys as that agent (environment rules hold it to the agent's limits). |
| 1201 | const actor = JSON.parse(run.actor) as Actor; |
| 1202 | environment = await this.repo.deploys.jobStarted({ id: jobId, ref: run.ref, sha: run.sha, logUrl, actor: run.agent_origin ? { ...actor, kind: "agent" } : actor }, { name, url }); |
| 1203 | } catch (e) { |
| 1204 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: (e as Error).message } }); |
| 1205 | throw new HttpError(409, (e as Error).message); |
| 1206 | } |
| 1207 | } |
| 1208 | const t = now(); |
| 1209 | this.sql.exec("UPDATE jobs SET status = 'in_progress', started_at = ? WHERE id = ?", t, jobId); |
| 1210 | this.repo.upsertCheck({ sha: run.head_sha ?? run.sha, name: this.checkName(run, job), status: "in_progress", runId: run.id, jobId, detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}` }); |
| 1211 | this.repo.broadcast(`run:${run.id}`, { action: "job", jobId }); |
| 1212 | const secrets: Record<string, string> = { ...scope.secrets, ...environment.secrets, GITHUB_TOKEN: token, DIFFINERY_TOKEN: token }; |
| 1213 | // The job token can do what the effective `permissions:` allow (job level wins, like GitHub). |
| 1214 | const permissions = permissionsForJob(exp, job.job_key, { untrusted: this.untrusted(run), repoDefault: this.repo.policy.settings().defaultPermissions }); |
| 1215 | this.repo.runtime.grant(jobId, permissions); |
| 1216 | const idToken = permissions["id-token"] === "write"; |
| 1217 | return { |
| 1218 | runId: run.id, |
| 1219 | jobId, |
| 1220 | jobKey: short, |
| 1221 | name: job.name, |
| 1222 | repo: { owner: m.owner, name: m.name, cloneUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, defaultBranch: m.defaultBranch }, |
| 1223 | sha: run.sha, |
| 1224 | ref: run.ref, |
| 1225 | workflowPath: run.workflow_path, |
| 1226 | github: this.githubContext(run, short, token), |
| 1227 | env: { ...wf.env, ...def.env }, |
| 1228 | defaults: { run: { ...wf.defaults.run, ...def.defaults.run } }, |
| 1229 | matrix: JSON.parse(job.matrix), |
| 1230 | strategy: JSON.parse(job.strategy), |
| 1231 | needs: await this.needsResults(run, exp, exp.needs.get(job.job_key) ?? [], parent), |
| 1232 | inputs: scope.inputs, |
| 1233 | vars: { ...this.variables(), ...environment.vars }, |
| 1234 | secrets, |
| 1235 | outputs: def.outputs, |
| 1236 | steps: def.steps.map((s) => ({ ...s, "timeout-minutes": s["timeout-minutes"] === undefined ? undefined : Number(s["timeout-minutes"]) })), |
| 1237 | container: def.container, |
| 1238 | services: def.services, |
| 1239 | timeoutMinutes: job.timeout_minutes, |
| 1240 | continueOnError: !!job.continue_on_error, |
| 1241 | idToken: !!idToken, |
| 1242 | apiUrl: this.env.PUBLIC_URL, |
| 1243 | token, |
| 1244 | runtime: await this.runtimeServices(run, job), |
| 1245 | }; |
| 1246 | } |
| 1247 | |
| 1248 | /** ACTIONS_RUNTIME_TOKEN and the service URLs; the token expires when the job would time out. */ |
| 1249 | private async runtimeServices(run: RunRow, job: JobRow): Promise<JobSpec["runtime"]> { |
| 1250 | if (!this.env.MASTER_KEY) return undefined; |
| 1251 | const origin = new URL(this.env.PUBLIC_URL).origin; |
| 1252 | const exp = Math.floor(Date.now() / 1000) + (job.timeout_minutes + 30) * 60; |
| 1253 | const { scopes } = this.repo.runtime.job(job.id); |
| 1254 | const token = await signRuntimeToken(this.env.MASTER_KEY, origin, { repo: this.repo.meta.repoId, run: run.id, job: job.id, exp }, scopes); |
| 1255 | return { token, resultsUrl: `${origin}/`, cacheUrl: `${origin}/_runtime/`, runtimeUrl: `${origin}/_runtime/` }; |
| 1256 | } |
| 1257 | |
| 1258 | async runnerLogs(jobId: string, token: string, lines: LogLine[]): Promise<void> { |
| 1259 | await this.authJob(jobId, token); |
| 1260 | let seq = this.sql.exec<{ n: number | null }>("SELECT max(seq) AS n FROM logs WHERE job_id = ?", jobId).one().n ?? -1; |
| 1261 | const out: Array<LogLine & { seq: number }> = []; |
| 1262 | for (const l of lines.slice(0, 5000)) { |
| 1263 | seq++; |
| 1264 | const text = l.text.length > 16384 ? `${l.text.slice(0, 16384)}…` : l.text; |
| 1265 | this.sql.exec("INSERT INTO logs (job_id, seq, t, step, text) VALUES (?, ?, ?, ?, ?)", jobId, seq, l.t, l.step, text); |
| 1266 | out.push({ ...l, text, seq }); |
| 1267 | } |
| 1268 | this.repo.broadcast(`job:${jobId}`, { action: "logs", lines: out }); |
| 1269 | } |
| 1270 | |
| 1271 | async runnerStep(jobId: string, token: string, index: number, report: StepReport): Promise<void> { |
| 1272 | const job = await this.authJob(jobId, token); |
| 1273 | const steps = JSON.parse(job.steps) as Array<StepReport & { index: number }>; |
| 1274 | const existing = steps.findIndex((s) => s.index === index); |
| 1275 | const next = { ...(existing >= 0 ? steps[existing] : {}), ...report, index }; |
| 1276 | if (existing >= 0) steps[existing] = next; |
| 1277 | else steps.push(next); |
| 1278 | steps.sort((a, b) => a.index - b.index); |
| 1279 | this.sql.exec("UPDATE jobs SET steps = ? WHERE id = ?", JSON.stringify(steps), jobId); |
| 1280 | this.repo.broadcast(`job:${jobId}`, { action: "step", step: next }); |
| 1281 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId }); |
| 1282 | } |
| 1283 | |
| 1284 | async runnerComplete(jobId: string, token: string, completion: JobCompletion): Promise<void> { |
| 1285 | await this.authJob(jobId, token); |
| 1286 | await this.completeJob(jobId, completion); |
| 1287 | } |
| 1288 | |
| 1289 | private async completeJob(jobId: string, completion: JobCompletion, override?: "timed_out"): Promise<void> { |
| 1290 | const job = this.jobRow(jobId); |
| 1291 | if (job.status === "completed") return; |
| 1292 | const run = this.runRow(job.run_id); |
| 1293 | const t = now(); |
| 1294 | const conclusion = override ?? completion.conclusion; |
| 1295 | this.sql.exec( |
| 1296 | "UPDATE jobs SET status = 'completed', conclusion = ?, completed_at = ?, outputs = ?, failure = ?, summary = ?, token_hash = NULL, started_at = COALESCE(started_at, ?) WHERE id = ?", |
| 1297 | conclusion, |
| 1298 | t, |
| 1299 | JSON.stringify(completion.outputs ?? {}), |
| 1300 | completion.failure ? JSON.stringify(completion.failure) : null, |
| 1301 | completion.summary ?? null, |
| 1302 | t, |
| 1303 | jobId, |
| 1304 | ); |
| 1305 | this.repo.deploys.jobCompleted(jobId, conclusion); |
| 1306 | const m = this.repo.meta; |
| 1307 | this.repo.upsertCheck({ |
| 1308 | sha: run.head_sha ?? run.sha, |
| 1309 | name: this.checkName(run, job), |
| 1310 | status: "completed", |
| 1311 | conclusion: conclusion === "timed_out" ? "timed_out" : conclusion, |
| 1312 | runId: run.id, |
| 1313 | jobId, |
| 1314 | detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}`, |
| 1315 | output: completion.failure ? { title: "Failed", summary: completion.failure.message } : { title: conclusion }, |
| 1316 | }); |
| 1317 | // Archive logs to R2 and free the Durable Object's storage. |
| 1318 | const lines = this.sql.exec<LogLine & { seq: number }>("SELECT seq, t, step, text FROM logs WHERE job_id = ? ORDER BY seq", jobId).toArray(); |
| 1319 | if (lines.length) { |
| 1320 | await this.env.BUCKET.put(this.logKey(jobId), JSON.stringify(lines), { httpMetadata: { contentType: "application/json" } }); |
| 1321 | this.sql.exec("DELETE FROM logs WHERE job_id = ?", jobId); |
| 1322 | } |
| 1323 | // Uploads the job left unfinished stop counting against the budgets, and their bytes go. |
| 1324 | await this.repo.runtime.sweepJob(jobId).catch((e) => console.error("runtime sweep failed", e)); |
| 1325 | this.repo.broadcast(`job:${jobId}`, { action: "completed", conclusion }); |
| 1326 | this.repo.broadcast(`run:${run.id}`, { action: "job", jobId }); |
| 1327 | |
| 1328 | // fail-fast: cancel unfinished siblings of a failed matrix job (unless the failed leg may continue on error). |
| 1329 | const siblings = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND job_key = ? AND id != ? AND status != 'completed'", run.id, job.job_key, jobId).toArray(); |
| 1330 | const strategy = JSON.parse(job.strategy || "{}") as { "fail-fast"?: boolean }; |
| 1331 | if ((conclusion === "failure" || conclusion === "timed_out") && !job.continue_on_error && strategy["fail-fast"] !== false && siblings.length) { |
| 1332 | for (const s of siblings) { |
| 1333 | await this.stopContainer(s); |
| 1334 | await this.completeJob(s.id, { conclusion: "cancelled", outputs: {}, failure: { message: `Cancelled because ${job.name} failed (fail-fast).` } }); |
| 1335 | } |
| 1336 | } else { |
| 1337 | const queued = siblings.find((s) => s.status === "queued"); |
| 1338 | if (queued) await this.launch(queued.id); |
| 1339 | } |
| 1340 | // A job that held its concurrency group hands it to the oldest pending job (of any run). |
| 1341 | if (job.concurrency_group && (job.status === "waiting" || job.status === "in_progress")) { |
| 1342 | const next = this.sql.exec<{ id: string }>("SELECT id FROM jobs WHERE concurrency_group = ? AND status = 'pending' ORDER BY rowid LIMIT 1", job.concurrency_group).toArray()[0]; |
| 1343 | if (next) await this.launch(next.id); |
| 1344 | } |
| 1345 | try { |
| 1346 | const instance = await this.env.RUN_WORKFLOW.get(run.id); |
| 1347 | await instance.sendEvent({ type: `job-${jobId}`, payload: { conclusion } }); |
| 1348 | } catch (e) { |
| 1349 | console.error("failed to notify workflow", e); |
| 1350 | } |
| 1351 | } |
| 1352 | |
| 1353 | // ---------------------------------------------------------------- secrets and variables |
| 1354 | |
| 1355 | private async repoSecrets(): Promise<Record<string, string>> { |
| 1356 | const out: Record<string, string> = {}; |
| 1357 | for (const r of this.sql.exec<{ name: string; ciphertext: string }>("SELECT name, ciphertext FROM secrets").toArray()) { |
| 1358 | out[r.name] = await decryptSecret(this.env.MASTER_KEY, `repo:${this.repo.meta.repoId}`, r.ciphertext); |
| 1359 | } |
| 1360 | return out; |
| 1361 | } |
| 1362 | |
| 1363 | private async orgSecrets(): Promise<Record<string, string>> { |
| 1364 | const m = this.repo.meta; |
| 1365 | if (m.ownerKind !== "org") return {}; |
| 1366 | const rows = await this.env.DB.prepare("SELECT name, ciphertext FROM org_secrets WHERE org_id = ?").bind(m.ownerId).all<{ name: string; ciphertext: string }>(); |
| 1367 | const out: Record<string, string> = {}; |
| 1368 | for (const r of rows.results) out[r.name] = await decryptSecret(this.env.MASTER_KEY, `org:${m.ownerId}`, r.ciphertext); |
| 1369 | return out; |
| 1370 | } |
| 1371 | |
| 1372 | async setSecret(name: string, value: string): Promise<void> { |
| 1373 | if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(name) || /^GITHUB_/i.test(name)) throw new HttpError(422, "invalid secret name"); |
| 1374 | this.sql.exec("INSERT OR REPLACE INTO secrets (name, ciphertext, updated_at) VALUES (?, ?, ?)", name.toUpperCase(), await encryptSecret(this.env.MASTER_KEY, `repo:${this.repo.meta.repoId}`, value), now()); |
| 1375 | } |
| 1376 | |
| 1377 | deleteSecret(name: string): void { |
| 1378 | this.sql.exec("DELETE FROM secrets WHERE name = ?", name.toUpperCase()); |
| 1379 | } |
| 1380 | |
| 1381 | secretNames(): Array<{ name: string; updatedAt: string }> { |
| 1382 | return this.sql.exec<{ name: string; updated_at: string }>("SELECT name, updated_at FROM secrets ORDER BY name").toArray().map((r) => ({ name: r.name, updatedAt: r.updated_at })); |
| 1383 | } |
| 1384 | |
| 1385 | setVariable(name: string, value: string | null): void { |
| 1386 | if (value === null) this.sql.exec("DELETE FROM variables WHERE name = ?", name.toUpperCase()); |
| 1387 | else this.sql.exec("INSERT OR REPLACE INTO variables (name, value, updated_at) VALUES (?, ?, ?)", name.toUpperCase(), value, now()); |
| 1388 | } |
| 1389 | |
| 1390 | /** Variables for `vars.*`: the organization's, overridden by the repository's. `withOrg = false` lists only the repository's own. */ |
| 1391 | variables(withOrg = true): Record<string, string> { |
| 1392 | const own = Object.fromEntries(this.sql.exec<{ name: string; value: string }>("SELECT name, value FROM variables").toArray().map((r) => [r.name, r.value])); |
| 1393 | if (!withOrg) return own; |
| 1394 | const org = Object.fromEntries(this.sql.exec<{ name: string; value: string }>("SELECT name, value FROM org_variables").toArray().map((r) => [r.name, r.value])); |
| 1395 | return { ...org, ...own }; |
| 1396 | } |
| 1397 | |
| 1398 | // ---------------------------------------------------------------- cache and artifacts |
| 1399 | |
| 1400 | /** |
| 1401 | * An older runner's cache (the runner API before the toolkit's cache service). |
| 1402 | * Entries are stored as `<ref>\n<key>` and scoped like the cache service's |
| 1403 | * (repo/runtime.ts): a job saves to its own ref and restores from its own ref, |
| 1404 | * then the pull request's base, then the default branch, so one branch or pull |
| 1405 | * request never sees another's caches. Entries saved before scoping (no ref) |
| 1406 | * are never restored, so they go a week after their last use. |
| 1407 | */ |
| 1408 | async cacheLookup(jobId: string, token: string, key: string, restoreKeys: string[]): Promise<{ key: string; stored: string } | null> { |
| 1409 | await this.authJob(jobId, token); |
| 1410 | const { scopes } = this.repo.runtime.job(jobId); |
| 1411 | // An entry whose object is gone is forgotten, and the lookup moves on. |
| 1412 | for (let attempt = 0; attempt < 5; attempt++) { |
| 1413 | const hit = this.legacyCacheMatch(scopes, key, restoreKeys); |
| 1414 | if (!hit) return null; |
| 1415 | if (!(await headFirst(this.env.BUCKET, this.cacheR2Keys(hit)))) { |
| 1416 | this.sql.exec("DELETE FROM caches WHERE key = ?", hit); |
| 1417 | continue; |
| 1418 | } |
| 1419 | this.sql.exec("UPDATE caches SET last_used = ? WHERE key = ?", now(), hit); |
| 1420 | return { key: hit.slice(hit.indexOf("\n") + 1), stored: hit }; |
| 1421 | } |
| 1422 | return null; |
| 1423 | } |
| 1424 | |
| 1425 | /** In each scope in turn: the exact key, then the newest entry for each restore key prefix. */ |
| 1426 | private legacyCacheMatch(scopes: string[], key: string, restoreKeys: string[]): string | null { |
| 1427 | for (const scope of scopes) { |
| 1428 | const exact = this.sql.exec<{ key: string }>("SELECT key FROM caches WHERE key = ?", `${scope}\n${key}`).toArray()[0]; |
| 1429 | if (exact) return exact.key; |
| 1430 | for (const prefix of restoreKeys) { |
| 1431 | const p = `${scope}\n${prefix}`; |
| 1432 | const r = this.sql.exec<{ key: string }>("SELECT key FROM caches WHERE substr(key, 1, ?) = ? ORDER BY created_at DESC LIMIT 1", p.length, p).toArray()[0]; |
| 1433 | if (r) return r.key; |
| 1434 | } |
| 1435 | } |
| 1436 | return null; |
| 1437 | } |
| 1438 | |
| 1439 | /** Where a job saves `key` (its own ref's scope): the stored key and its R2 keys. */ |
| 1440 | async cacheSaveKeys(jobId: string, token: string, key: string): Promise<{ stored: string; r2Keys: string[] }> { |
| 1441 | await this.authJob(jobId, token); |
| 1442 | const stored = `${this.repo.runtime.job(jobId).scopes[0]}\n${key}`; |
| 1443 | return { stored, r2Keys: this.cacheR2Keys(stored) }; |
| 1444 | } |
| 1445 | |
| 1446 | /** The R2 keys of entry `stored`, when the job may read its scope. */ |
| 1447 | async cacheReadKeys(jobId: string, token: string, stored: string): Promise<string[] | null> { |
| 1448 | await this.authJob(jobId, token); |
| 1449 | const nl = stored.indexOf("\n"); |
| 1450 | if (nl < 0 || !this.repo.runtime.job(jobId).scopes.includes(stored.slice(0, nl))) return null; |
| 1451 | return this.cacheR2Keys(stored); |
| 1452 | } |
| 1453 | |
| 1454 | /** Where an older runner's cache entry is stored: the current key, then the old layout's. */ |
| 1455 | cacheR2Keys(key: string): string[] { |
| 1456 | return legacyCacheKeys(this.repo.meta.repoId, key); |
| 1457 | } |
| 1458 | |
| 1459 | /** Records a saved entry; returns the R2 keys of the entries it evicted, for the caller to delete. */ |
| 1460 | async cacheSaved(jobId: string, token: string, key: string, size: number): Promise<string[]> { |
| 1461 | const { stored } = await this.cacheSaveKeys(jobId, token, key); |
| 1462 | this.sql.exec("INSERT OR REPLACE INTO caches (key, size, created_at, last_used) VALUES (?, ?, ?, ?)", stored, size, now(), now()); |
| 1463 | // Like the cache service (repo/runtime.ts): entries unused for a week go, then the |
| 1464 | // least recently used ones above 10 GB per repository. |
| 1465 | const stale = this.sql.exec<{ key: string }>("SELECT key FROM caches WHERE last_used < ?", new Date(Date.now() - CACHE_IDLE_MS).toISOString()).toArray().map((r) => r.key); |
| 1466 | for (const k of stale) this.sql.exec("DELETE FROM caches WHERE key = ?", k); |
| 1467 | const evicted = stale.flatMap((k) => this.cacheR2Keys(k)); |
| 1468 | let total = this.sql.exec<{ n: number | null }>("SELECT sum(size) AS n FROM caches").one().n ?? 0; |
| 1469 | while (total > CACHE_LIMIT_BYTES) { |
| 1470 | const old = this.sql.exec<{ key: string; size: number }>("SELECT key, size FROM caches ORDER BY last_used LIMIT 1").one(); |
| 1471 | this.sql.exec("DELETE FROM caches WHERE key = ?", old.key); |
| 1472 | evicted.push(...this.cacheR2Keys(old.key)); |
| 1473 | total -= old.size; |
| 1474 | } |
| 1475 | return evicted; |
| 1476 | } |
| 1477 | |
| 1478 | /** |
| 1479 | * Where job `jobId` (authenticated by its token) stores artifact `name` of its run, and how many bytes |
| 1480 | * it may store there (what is left of the artifact quota). Refused when the run |
| 1481 | * already has an artifact of that name from the artifact service. |
| 1482 | */ |
| 1483 | async artifactUploadKeys(jobId: string, token: string, name: string): Promise<{ r2Keys: string[]; room: number }> { |
| 1484 | const job = await this.authJob(jobId, token); |
| 1485 | this.repo.runtime.legacyArtifactCheck(job.run_id, name); |
| 1486 | return { r2Keys: this.artifactR2Keys(job.run_id, name), room: await this.repo.runtime.artifactRoom() }; |
| 1487 | } |
| 1488 | |
| 1489 | /** Records an older runner's artifact, stored at its legacy key, with the artifact service's (repo/runtime.ts). */ |
| 1490 | async artifactSaved(jobId: string, token: string, name: string, size: number): Promise<{ id: number }> { |
| 1491 | const job = await this.authJob(jobId, token); |
| 1492 | return { id: this.repo.runtime.legacyArtifactSaved(job.run_id, jobId, name, size).id }; |
| 1493 | } |
| 1494 | |
| 1495 | /** Where job `jobId` (authenticated by its token) reads artifact `name` of its run. */ |
| 1496 | async artifactKeysFor(jobId: string, token: string, name: string): Promise<string[]> { |
| 1497 | const job = await this.authJob(jobId, token); |
| 1498 | return this.artifactR2Keys(job.run_id, name); |
| 1499 | } |
| 1500 | |
| 1501 | /** Where an older runner's artifact is stored: the current key, then the old layout's. */ |
| 1502 | artifactR2Keys(runId: string, name: string): string[] { |
| 1503 | return legacyArtifactKeys(this.repo.meta.repoId, runId, name); |
| 1504 | } |
| 1505 | |
| 1506 | /** |
| 1507 | * A run's artifacts: those uploaded through the artifact service (zip files, or |
| 1508 | * single unarchived files) and older runners' tarballs. |
| 1509 | */ |
| 1510 | artifacts(runId: string): Array<{ name: string; size: number; createdAt: string; r2Keys: string[]; id: number; contentType: string; fileName: string }> { |
| 1511 | runId = this.runRow(runId).id; |
| 1512 | return this.repo.runtime |
| 1513 | .runArtifacts(runId) |
| 1514 | .map((a) => ({ |
| 1515 | name: a.name, |
| 1516 | size: a.size, |
| 1517 | createdAt: a.createdAt, |
| 1518 | r2Keys: artifactObjectKeys(this.repo.meta.repoId, a), |
| 1519 | id: a.id, |
| 1520 | contentType: a.mimeType, |
| 1521 | fileName: artifactFileName(a), |
| 1522 | })) |
| 1523 | .sort((a, b) => a.name.localeCompare(b.name)); |
| 1524 | } |
| 1525 | |
| 1526 | // ---------------------------------------------------------------- sandboxes for agents |
| 1527 | |
| 1528 | async createSandbox(caller: Caller, ref?: string): Promise<{ id: string; ref: string }> { |
| 1529 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "write access is required to create a sandbox"); |
| 1530 | const sha = await this.repo.resolve(ref ?? "HEAD"); |
| 1531 | if (!sha) throw new HttpError(422, "unknown ref"); |
| 1532 | const id = newId(); |
| 1533 | const m = this.repo.meta; |
| 1534 | const token = `dst_${m.repoId}_${id}_${randomToken(24)}`; |
| 1535 | const secret = randomToken(32); |
| 1536 | this.sql.exec("INSERT INTO sandboxes (id, secret, created_by, ref, created_at, status) VALUES (?, ?, ?, ?, ?, 'running')", id, await sha256(token), JSON.stringify(caller), ref ?? m.defaultBranch, now()); |
| 1537 | const stub = this.env.SANDBOX.get(this.env.SANDBOX.idFromName(id)); |
| 1538 | try { |
| 1539 | await stub.boot({ |
| 1540 | DIFFINERY_CLONE_URL: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, |
| 1541 | DIFFINERY_REF: ref ?? m.defaultBranch, |
| 1542 | DIFFINERY_JOB_TOKEN: token, |
| 1543 | DIFFINERY_SANDBOX_SECRET: secret, |
| 1544 | DIFFINERY_API_URL: this.env.PUBLIC_URL, |
| 1545 | GIT_AUTHOR_NAME: caller.login, |
| 1546 | GIT_AUTHOR_EMAIL: `${caller.login}@users.noreply.diffinery.com`, |
| 1547 | GIT_COMMITTER_NAME: caller.login, |
| 1548 | GIT_COMMITTER_EMAIL: `${caller.login}@users.noreply.diffinery.com`, |
| 1549 | }, secret); |
| 1550 | } catch (e) { |
| 1551 | console.error("sandbox boot failed", e); |
| 1552 | this.sql.exec("UPDATE sandboxes SET status = 'failed' WHERE id = ?", id); |
| 1553 | await stub.shutdown().catch(() => {}); |
| 1554 | throw new HttpError(503, "the sandbox container did not start; try again"); |
| 1555 | } |
| 1556 | return { id, ref: ref ?? m.defaultBranch }; |
| 1557 | } |
| 1558 | |
| 1559 | getSandbox(id: string, caller: Caller): { id: string; ref: string; status: string; createdAt: string } { |
| 1560 | const s = this.sql.exec<{ id: string; ref: string; status: string; created_at: string; created_by: string }>("SELECT * FROM sandboxes WHERE id = ?", id).toArray()[0]; |
| 1561 | if (!s) throw new HttpError(404, "sandbox not found"); |
| 1562 | const owner = JSON.parse(s.created_by) as Caller; |
| 1563 | if (owner.id !== caller.id && !atLeast(caller.role, "admin")) throw new HttpError(403, "not your sandbox"); |
| 1564 | return { id: s.id, ref: s.ref, status: s.status, createdAt: s.created_at }; |
| 1565 | } |
| 1566 | |
| 1567 | async closeSandbox(id: string, caller: Caller): Promise<void> { |
| 1568 | this.getSandbox(id, caller); |
| 1569 | this.sql.exec("UPDATE sandboxes SET status = 'closed' WHERE id = ?", id); |
| 1570 | await this.env.SANDBOX.get(this.env.SANDBOX.idFromName(id)).shutdown(); |
| 1571 | } |
| 1572 | } |