| 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 | import { schedulerStub } from "../lib/schedules.js"; |
| 34 | import { APPROVAL_DAYS, type Deployment, type DeploymentReview, type Reviewer } from "./deployments.js"; |
| 35 | import { referencedSecrets, RUNNER_SECRETS, secretsFor } from "./job-secrets.js"; |
| 36 | |
| 37 | /** How long a sandbox token works after the sandbox starts. */ |
| 38 | export const SANDBOX_TOKEN_TTL_MS = 24 * 3600_000; |
| 39 | /** Sandboxes one person or agent may have open in a repository at once: their containers are shared by the whole instance. */ |
| 40 | export const MAX_OPEN_SANDBOXES = 5; |
| 41 | |
| 42 | const WORKFLOW_DIRS = [".github/workflows", ".diffinery/workflows"]; |
| 43 | /** Characters of text a job's log keeps; the rest is dropped after a line saying so. */ |
| 44 | export const MAX_JOB_LOG = 16 * 1048576; |
| 45 | /** Every push reads and parses a commit's workflow files: at most this many, each at most this large. */ |
| 46 | export const MAX_WORKFLOW_FILES = 256; |
| 47 | export const MAX_WORKFLOW_BYTES = 512 * 1024; |
| 48 | /** Hosted jobs one repository runs at once (GitHub's free plan allows 20 per account); more wait their turn. */ |
| 49 | export const MAX_HOSTED_JOBS = 20; |
| 50 | /** |
| 51 | * A hosted job whose container never got a machine (the platform had none free, usually when many |
| 52 | * jobs start at once or right after a deploy) goes back in line and starts again later, this many |
| 53 | * times, before it fails. Only jobs whose runner never reported in: nothing of theirs has run. |
| 54 | */ |
| 55 | export const MAX_PLACEMENT_TRIES = 8; |
| 56 | /** Seconds before the next start of a job whose container got no machine: 30 s, doubling to 5 minutes. */ |
| 57 | export const placementDelay = (tries: number): number => Math.min(30 * 2 ** (tries - 1), 300); |
| 58 | const NO_INSTANCE = "no container instance"; |
| 59 | /** Runs one repository may have unfinished at once; past it, new ones aren't started. */ |
| 60 | export const MAX_ACTIVE_RUNS = 500; |
| 61 | /** Self-hosted runners one repository may have registered. */ |
| 62 | export const MAX_RUNNERS = 100; |
| 63 | /** How long a runner registration token works, as on GitHub. */ |
| 64 | const REGISTRATION_TTL_MS = 3600_000; |
| 65 | /** A runner that polled this recently shows as online. */ |
| 66 | const RUNNER_ONLINE_MS = 90_000; |
| 67 | |
| 68 | type RunnerRow = { id: number; name: string; labels: string; token_hash: string | null; ephemeral: number; created_by: string; created_at: string; last_seen_at: string | null }; |
| 69 | |
| 70 | export interface RunnerInfo { |
| 71 | id: number; |
| 72 | name: string; |
| 73 | os: string; |
| 74 | status: "online" | "offline"; |
| 75 | busy: boolean; |
| 76 | ephemeral: boolean; |
| 77 | labels: Array<{ id: number; name: string; type: "read-only" | "custom" }>; |
| 78 | } |
| 79 | const SUPPORTED_LABEL = /^(ubuntu-.*|linux|diffinery(-.*)?|self-hosted|x64|amd64)$/i; |
| 80 | |
| 81 | export type RunRow = { |
| 82 | id: string; |
| 83 | run_number: number; |
| 84 | workflow_path: string; |
| 85 | workflow_name: string; |
| 86 | event: string; |
| 87 | ref: string; |
| 88 | sha: string; |
| 89 | actor: string; |
| 90 | status: "queued" | "in_progress" | "completed" | "waiting"; |
| 91 | conclusion: Conclusion | "timed_out" | null; |
| 92 | created_at: string; |
| 93 | updated_at: string; |
| 94 | pr_number: number | null; |
| 95 | attempt: number; |
| 96 | event_payload: string; |
| 97 | inputs: string; |
| 98 | concurrency_group: string | null; |
| 99 | display_title: string; |
| 100 | /** Commit the run reports checks on, when it differs from `sha` (pull request runs). */ |
| 101 | head_sha: string | null; |
| 102 | /** 1 when the code the run executes came from an agent; its jobs then act as an agent. */ |
| 103 | agent_origin?: number; |
| 104 | /** Account ids that started earlier attempts of this run (JSON list), for prevent_self_review. */ |
| 105 | prior_actors?: string | null; |
| 106 | }; |
| 107 | |
| 108 | export type JobRow = { |
| 109 | id: string; |
| 110 | run_id: string; |
| 111 | job_key: string; |
| 112 | name: string; |
| 113 | /** |
| 114 | * pending: held back by its concurrency group until the group's running job finishes. |
| 115 | * approval: waiting on its environment's required reviewers or wait timer; nothing has started. |
| 116 | */ |
| 117 | status: "queued" | "pending" | "approval" | "waiting" | "in_progress" | "completed"; |
| 118 | conclusion: Conclusion | "timed_out" | null; |
| 119 | matrix: string; |
| 120 | strategy: string; |
| 121 | started_at: string | null; |
| 122 | completed_at: string | null; |
| 123 | outputs: string; |
| 124 | steps: string; |
| 125 | failure: string | null; |
| 126 | token_hash: string | null; |
| 127 | summary: string | null; |
| 128 | container: string | null; |
| 129 | timeout_minutes: number; |
| 130 | /** runs-on labels for self-hosted jobs (JSON array); null for Diffinery-hosted containers. */ |
| 131 | labels: string | null; |
| 132 | /** `continue-on-error` as evaluated for this matrix leg (0 or 1). */ |
| 133 | continue_on_error: number; |
| 134 | /** Job-level `concurrency` group as evaluated for this leg, and its cancel-in-progress (0 or 1). */ |
| 135 | concurrency_group: string | null; |
| 136 | concurrency_cancel: number; |
| 137 | log_chars: number; |
| 138 | fleet_held: number; |
| 139 | runner_id: number | null; |
| 140 | /** `environment:` as evaluated at launch (null for jobs without one, and for jobs launched before it was stored). */ |
| 141 | environment: string | null; |
| 142 | /** When the job started waiting on its environment's protection rules, and "released" once they passed. */ |
| 143 | gate_since: string | null; |
| 144 | gate_state: string | null; |
| 145 | placement_tries: number; |
| 146 | }; |
| 147 | |
| 148 | /** An environment a run's jobs wait on (GitHub's pending deployment). */ |
| 149 | export type PendingDeployment = { |
| 150 | environment: { id: number; name: string }; |
| 151 | waitTimer: number; |
| 152 | waitTimerStartedAt: string; |
| 153 | reviewers: Reviewer[]; |
| 154 | preventSelfReview: boolean; |
| 155 | /** Still waiting for a required reviewer (false once approved, or when only a timer applies). */ |
| 156 | needsReview: boolean; |
| 157 | currentUserCanApprove: boolean; |
| 158 | /** The viewer is listed but started the run, and the environment prevents self-review. */ |
| 159 | selfBlocked: boolean; |
| 160 | }; |
| 161 | |
| 162 | export type RunInfo = { |
| 163 | id: string; |
| 164 | number: number; |
| 165 | attempt: number; |
| 166 | workflowPath: string; |
| 167 | workflowName: string; |
| 168 | title: string; |
| 169 | event: string; |
| 170 | ref: string; |
| 171 | sha: string; |
| 172 | /** The commit checks are reported on: the pull request head for pull_request runs. */ |
| 173 | headSha: string; |
| 174 | /** Branch the run is about: the pull request's head branch for pull request runs. */ |
| 175 | headBranch: string; |
| 176 | actor: Actor; |
| 177 | status: RunRow["status"]; |
| 178 | conclusion: RunRow["conclusion"]; |
| 179 | createdAt: string; |
| 180 | updatedAt: string; |
| 181 | prNumber: number | null; |
| 182 | }; |
| 183 | |
| 184 | export type JobInfo = { |
| 185 | id: string; |
| 186 | runId: string; |
| 187 | key: string; |
| 188 | name: string; |
| 189 | status: JobRow["status"]; |
| 190 | conclusion: JobRow["conclusion"]; |
| 191 | matrix: Record<string, Json>; |
| 192 | startedAt: string | null; |
| 193 | completedAt: string | null; |
| 194 | outputs: Record<string, string>; |
| 195 | steps: Array<StepReport & { index: number }>; |
| 196 | failure: JobCompletion["failure"] | null; |
| 197 | summary: string | null; |
| 198 | }; |
| 199 | |
| 200 | const toRun = (r: RunRow): RunInfo => ({ |
| 201 | id: r.id, |
| 202 | number: r.run_number, |
| 203 | attempt: r.attempt, |
| 204 | workflowPath: r.workflow_path, |
| 205 | workflowName: r.workflow_name, |
| 206 | title: r.display_title, |
| 207 | event: r.event, |
| 208 | ref: r.ref, |
| 209 | sha: r.sha, |
| 210 | headSha: r.head_sha ?? r.sha, |
| 211 | 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)\//, ""), |
| 212 | actor: JSON.parse(r.actor), |
| 213 | status: r.status, |
| 214 | conclusion: r.conclusion, |
| 215 | createdAt: r.created_at, |
| 216 | updatedAt: r.updated_at, |
| 217 | prNumber: r.pr_number, |
| 218 | }); |
| 219 | |
| 220 | const toJob = (j: JobRow): JobInfo => ({ |
| 221 | id: j.id, |
| 222 | runId: j.run_id, |
| 223 | key: j.job_key, |
| 224 | name: j.name, |
| 225 | status: j.status, |
| 226 | conclusion: j.conclusion, |
| 227 | matrix: JSON.parse(j.matrix), |
| 228 | startedAt: j.started_at, |
| 229 | completedAt: j.completed_at, |
| 230 | outputs: JSON.parse(j.outputs), |
| 231 | steps: JSON.parse(j.steps), |
| 232 | failure: j.failure ? JSON.parse(j.failure) : null, |
| 233 | summary: j.summary, |
| 234 | }); |
| 235 | |
| 236 | /** |
| 237 | * GitHub Actions compatible run orchestration state. The Cloudflare Workflow |
| 238 | * (`RunWorkflow`) drives a run through these methods; the runner agent inside |
| 239 | * each job's container reports back through the runner* methods. |
| 240 | */ |
| 241 | export class Actions { |
| 242 | constructor( |
| 243 | private repo: RepoDO, |
| 244 | private env: Env, |
| 245 | ) {} |
| 246 | |
| 247 | private get sql() { |
| 248 | return this.repo.sql; |
| 249 | } |
| 250 | |
| 251 | // ---------------------------------------------------------------- workflow discovery |
| 252 | |
| 253 | async loadWorkflows(sha: string): Promise<Array<{ path: string; workflow?: Workflow; errors: string[] }>> { |
| 254 | const commit = await readCommit(this.repo.store, sha); |
| 255 | const out: Array<{ path: string; workflow?: Workflow; errors: string[] }> = []; |
| 256 | for (const dir of WORKFLOW_DIRS) { |
| 257 | const entry = await resolvePath(this.repo.store, commit.tree, dir); |
| 258 | if (!entry || !isTreeMode(entry.mode)) continue; |
| 259 | for (const f of await readTree(this.repo.store, entry.oid)) { |
| 260 | if (!/\.ya?ml$/.test(f.name) || isTreeMode(f.mode)) continue; |
| 261 | if (out.length >= MAX_WORKFLOW_FILES) break; |
| 262 | const path = `${dir}/${f.name}`; |
| 263 | if ((this.repo.store.size(f.oid) ?? 0) > MAX_WORKFLOW_BYTES) { |
| 264 | out.push({ path, errors: [`: the workflow file is larger than ${MAX_WORKFLOW_BYTES / 1024} KiB`] }); |
| 265 | continue; |
| 266 | } |
| 267 | const o = await this.repo.store.read(f.oid); |
| 268 | if (!o) continue; |
| 269 | const { workflow, errors } = parseWorkflow(fromUtf8(o.content), path); |
| 270 | out.push({ path, workflow, errors: errors.filter((e) => !isWarning(e)).map((e) => `${e.path}${e.line ? `:${e.line}` : ""}: ${e.message}`) }); |
| 271 | } |
| 272 | } |
| 273 | return out; |
| 274 | } |
| 275 | |
| 276 | async listWorkflows(): Promise<Array<{ path: string; name: string; errors: string[]; triggers: string[] }>> { |
| 277 | const sha = await this.repo.resolve("HEAD"); |
| 278 | if (!sha) return []; |
| 279 | 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) : [] })); |
| 280 | } |
| 281 | |
| 282 | /** The `repository` object of event payloads: GitHub's shape, with the repository's real visibility. */ |
| 283 | async repositoryPayload(): Promise<Record<string, unknown>> { |
| 284 | const m = this.repo.meta; |
| 285 | const row = await new Directory(this.env).repoById(m.repoId); |
| 286 | if (row) return ghRepo(this.env.PUBLIC_URL, row); |
| 287 | 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` }; |
| 288 | } |
| 289 | |
| 290 | // ---------------------------------------------------------------- triggers |
| 291 | |
| 292 | async onPush(ev: { ref: string; before: string; after: string; forced: boolean; created: boolean; deleted: boolean }, caller: Caller): Promise<void> { |
| 293 | if (ev.deleted) return; |
| 294 | const workflows = await this.loadWorkflows(ev.after); |
| 295 | let changedFiles: string[] | undefined; |
| 296 | if (ev.before !== ZERO_OID && this.repo.store.has(ev.before)) { |
| 297 | const [a, b] = await Promise.all([readCommit(this.repo.store, ev.before), readCommit(this.repo.store, ev.after)]); |
| 298 | changedFiles = (await diffTrees(this.repo.store, a.tree, b.tree)).map((c) => c.path); |
| 299 | } |
| 300 | const head = await readCommit(this.repo.store, ev.after); |
| 301 | const payload = { |
| 302 | ref: ev.ref, |
| 303 | before: ev.before, |
| 304 | after: ev.after, |
| 305 | created: ev.created, |
| 306 | deleted: ev.deleted, |
| 307 | forced: ev.forced, |
| 308 | repository: await this.repositoryPayload(), |
| 309 | pusher: { name: caller.login }, |
| 310 | sender: { login: caller.login, type: caller.kind === "agent" ? "Bot" : "User" }, |
| 311 | 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 } }, |
| 312 | commits: [{ id: ev.after, message: head.message }], |
| 313 | }; |
| 314 | if (ev.ref === `refs/heads/${this.repo.meta.defaultBranch}`) await this.syncSchedules(workflows); |
| 315 | for (const w of workflows) { |
| 316 | if (!w.workflow) { |
| 317 | if (w.errors.length) await this.createFailedRun(w.path, "push", ev.ref, ev.after, caller, w.errors); |
| 318 | continue; |
| 319 | } |
| 320 | if (!matchesTrigger(w.workflow, { name: "push", ref: ev.ref, changedFiles })) continue; |
| 321 | await this.createRun(w.workflow, { event: "push", ref: ev.ref, sha: ev.after, payload, caller, title: head.message.split("\n")[0] ?? "" }); |
| 322 | } |
| 323 | } |
| 324 | |
| 325 | /** |
| 326 | * Pull request workflows, like GitHub's: `pull_request` runs the workflow from the test |
| 327 | * merge commit (`refs/pull/<n>/merge`, the head merged into the base; nothing runs while |
| 328 | * they conflict), and `pull_request_target` runs the base branch's own workflow on the |
| 329 | * base commit, so a pull request can never change what runs with the base's secrets. |
| 330 | * Both report their checks on the pull request's head commit. |
| 331 | */ |
| 332 | async onPullRequest(action: string, number: number, caller: Caller): Promise<void> { |
| 333 | const { issue } = this.repo.collab.get(number); |
| 334 | const pull = issue.pull!; |
| 335 | // Once merged, the base already holds the changes: keep the refs as they were and let path filters pass. |
| 336 | const { baseSha, mergeSha } = pull.merged ? { baseSha: this.repo.getRef(`refs/heads/${pull.base}`), mergeSha: null } : await this.repo.collab.refreshPullRefs(number); |
| 337 | const changedFiles = pull.merged ? undefined : (await this.repo.collab.pullFiles(number)).files.map((f) => f.path); |
| 338 | const m = this.repo.meta; |
| 339 | const repository = await this.repositoryPayload(); |
| 340 | const payload = { |
| 341 | action, |
| 342 | number, |
| 343 | pull_request: { |
| 344 | number, |
| 345 | id: number, |
| 346 | state: issue.state, |
| 347 | title: issue.title, |
| 348 | body: issue.body, |
| 349 | draft: pull.draft, |
| 350 | merged: pull.merged, |
| 351 | merged_at: pull.mergedAt, |
| 352 | merged_by: pull.mergedBy ? { login: pull.mergedBy } : null, |
| 353 | merge_commit_sha: pull.merged ? pull.mergeCommit : mergeSha, |
| 354 | user: { login: issue.author.login }, |
| 355 | labels: issue.labels.map((name) => ({ name })), |
| 356 | assignees: issue.assignees.map((login) => ({ login })), |
| 357 | head: { ref: pull.head, sha: pull.headSha, repo: repository }, |
| 358 | base: { ref: pull.base, sha: baseSha, repo: repository }, |
| 359 | html_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/pull/${number}`, |
| 360 | url: `${this.env.PUBLIC_URL}/api/v3/repos/${m.owner}/${m.name}/pulls/${number}`, |
| 361 | }, |
| 362 | repository, |
| 363 | sender: { login: caller.login }, |
| 364 | }; |
| 365 | const common = { payload, caller, prNumber: number, headSha: pull.headSha, headRef: pull.head, baseRef: pull.base, title: issue.title }; |
| 366 | // A merged pull request's runs use its merge commit, as on GitHub. |
| 367 | const prSha = pull.merged ? pull.mergeCommit : mergeSha; |
| 368 | if (prSha) { |
| 369 | for (const w of await this.loadWorkflows(prSha)) { |
| 370 | if (!w.workflow || !matchesTrigger(w.workflow, { name: "pull_request", action, baseRef: pull.base, changedFiles })) continue; |
| 371 | await this.createRun(w.workflow, { ...common, event: "pull_request", ref: `refs/pull/${number}/merge`, sha: prSha }); |
| 372 | } |
| 373 | } |
| 374 | if (baseSha) { |
| 375 | for (const w of await this.loadWorkflows(baseSha)) { |
| 376 | if (!w.workflow || !matchesTrigger(w.workflow, { name: "pull_request_target", action, baseRef: pull.base, changedFiles })) continue; |
| 377 | await this.createRun(w.workflow, { ...common, event: "pull_request_target", ref: `refs/heads/${pull.base}`, sha: baseSha }); |
| 378 | } |
| 379 | } |
| 380 | } |
| 381 | |
| 382 | /** `merge_group` workflows for a merge queue group, run from the group's commit. */ |
| 383 | async onMergeGroup(g: { number: number; base: string; ref: string; sha: string; parent: string; title: string; sender: Actor }): Promise<void> { |
| 384 | const head = await readCommit(this.repo.store, g.sha); |
| 385 | const payload = { |
| 386 | action: "checks_requested", |
| 387 | merge_group: { |
| 388 | head_sha: g.sha, |
| 389 | head_ref: g.ref, |
| 390 | base_sha: g.parent, |
| 391 | base_ref: `refs/heads/${g.base}`, |
| 392 | head_commit: { |
| 393 | id: g.sha, |
| 394 | tree_id: head.tree, |
| 395 | message: head.message, |
| 396 | timestamp: new Date(head.committer.timestamp * 1000).toISOString(), |
| 397 | author: { name: head.author.name, email: head.author.email }, |
| 398 | committer: { name: head.committer.name, email: head.committer.email }, |
| 399 | }, |
| 400 | }, |
| 401 | repository: await this.repositoryPayload(), |
| 402 | sender: { login: g.sender.login }, |
| 403 | }; |
| 404 | for (const w of await this.loadWorkflows(g.sha)) { |
| 405 | if (!w.workflow) { |
| 406 | if (w.errors.length) await this.createFailedRun(w.path, "merge_group", g.ref, g.sha, g.sender, w.errors); |
| 407 | continue; |
| 408 | } |
| 409 | if (!matchesTrigger(w.workflow, { name: "merge_group", action: "checks_requested", baseRef: g.base })) continue; |
| 410 | 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] }); |
| 411 | } |
| 412 | } |
| 413 | |
| 414 | /** Cancel unfinished runs on a ref (a merge group branch that is going away). */ |
| 415 | async cancelRunsOnRef(ref: string, reason: string): Promise<void> { |
| 416 | 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); |
| 417 | } |
| 418 | |
| 419 | async dispatch(workflowFile: string, ref: string, inputs: Record<string, unknown>, caller: Caller): Promise<RunInfo> { |
| 420 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "write access is required to run workflows"); |
| 421 | const sha = await this.repo.resolve(ref); |
| 422 | if (!sha) throw new HttpError(422, `unknown ref ${ref}`); |
| 423 | const all = await this.loadWorkflows(sha); |
| 424 | const w = all.find((x) => x.path === workflowFile || x.path.endsWith(`/${workflowFile}`)); |
| 425 | if (!w?.workflow) throw new HttpError(404, `workflow ${workflowFile} not found at ${ref}${w?.errors.length ? `: ${w.errors.join("; ")}` : ""}`); |
| 426 | if (!w.workflow.on.workflow_dispatch) throw new HttpError(422, "workflow does not have a workflow_dispatch trigger"); |
| 427 | const defs = w.workflow.on.workflow_dispatch.inputs ?? {}; |
| 428 | const resolved: Record<string, unknown> = {}; |
| 429 | for (const [k, d] of Object.entries(defs)) { |
| 430 | const v = inputs[k] ?? d.default; |
| 431 | if (d.required && (v === undefined || v === "")) throw new HttpError(422, `input '${k}' is required`); |
| 432 | 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(", ")}`); |
| 433 | resolved[k] = d.type === "boolean" ? v === true || v === "true" : d.type === "number" ? Number(v) : v; |
| 434 | } |
| 435 | const fullRef = ref.startsWith("refs/") ? ref : this.repo.getRef(`refs/heads/${ref}`) ? `refs/heads/${ref}` : `refs/tags/${ref}`; |
| 436 | const payload = { inputs: resolved, ref: fullRef, repository: await this.repositoryPayload(), sender: { login: caller.login }, workflow: w.path }; |
| 437 | const m = this.repo.meta; |
| 438 | 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() }); |
| 439 | return (await this.createRun(w.workflow, { event: "workflow_dispatch", ref: fullRef, sha, payload, caller, inputs: resolved, title: w.workflow.name ?? w.path }))!; |
| 440 | } |
| 441 | |
| 442 | async schedule(workflowPath: string, cron: string): Promise<void> { |
| 443 | const sha = await this.repo.resolve("HEAD"); |
| 444 | if (!sha) return; |
| 445 | const w = (await this.loadWorkflows(sha)).find((x) => x.path === workflowPath); |
| 446 | if (!w?.workflow || !matchesTrigger(w.workflow, { name: "schedule", cron })) return; |
| 447 | const system: Caller = { id: "system", login: "diffinery-scheduler", kind: "system", role: "write" }; |
| 448 | 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 }); |
| 449 | } |
| 450 | |
| 451 | private async syncSchedules(workflows: Array<{ path: string; workflow?: Workflow }>): Promise<void> { |
| 452 | const repoId = this.repo.meta.repoId; |
| 453 | const stmts = [this.env.DB.prepare("DELETE FROM schedules WHERE repo_id = ?").bind(repoId)]; |
| 454 | 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)); |
| 455 | await this.env.DB.batch(stmts); |
| 456 | // The scheduler's alarm starts with the first schedule, without waiting for the cron trigger. |
| 457 | if (stmts.length > 1) await schedulerStub(this.env).ensure().catch((e) => console.error("scheduler alarm check failed", e)); |
| 458 | } |
| 459 | |
| 460 | // ---------------------------------------------------------------- runs |
| 461 | |
| 462 | private nextRunNumber(): number { |
| 463 | this.sql.exec("INSERT INTO counters (name, value) VALUES ('run', 1) ON CONFLICT(name) DO UPDATE SET value = value + 1"); |
| 464 | return this.sql.exec<{ value: number }>("SELECT value FROM counters WHERE name = 'run'").one().value; |
| 465 | } |
| 466 | |
| 467 | private githubContext(run: RunRow, jobKey: string, token: string): Record<string, unknown> { |
| 468 | const m = this.repo.meta; |
| 469 | const payload = JSON.parse(run.event_payload) as Record<string, unknown>; |
| 470 | const actor = JSON.parse(run.actor) as Actor; |
| 471 | const refName = run.ref.replace(/^refs\/(heads|tags)\//, "").replace(/^refs\/pull\/(\d+)\/(head|merge)$/, "$1/merge"); |
| 472 | const pr = payload.pull_request as { head?: { ref?: string }; base?: { ref?: string } } | undefined; |
| 473 | return { |
| 474 | event_name: run.event, |
| 475 | event: payload, |
| 476 | sha: run.sha, |
| 477 | ref: run.ref, |
| 478 | ref_name: refName, |
| 479 | ref_type: run.ref.startsWith("refs/tags/") ? "tag" : "branch", |
| 480 | ref_protected: false, |
| 481 | head_ref: pr?.head?.ref ?? "", |
| 482 | base_ref: pr?.base?.ref ?? "", |
| 483 | repository: `${m.owner}/${m.name}`, |
| 484 | repository_id: String(numericId(m.repoId)), |
| 485 | repository_owner: m.owner, |
| 486 | repositoryUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, |
| 487 | actor: actor.login, |
| 488 | triggering_actor: actor.login, |
| 489 | // Numeric like GitHub's (actions parse it, e.g. download-artifact's run-id); the API accepts it as a run id. |
| 490 | run_id: String(run.run_number), |
| 491 | run_number: String(run.run_number), |
| 492 | run_attempt: String(run.attempt), |
| 493 | workflow: run.workflow_name, |
| 494 | workflow_ref: `${m.owner}/${m.name}/${run.workflow_path}@${run.ref}`, |
| 495 | job: jobKey, |
| 496 | server_url: this.env.PUBLIC_URL, |
| 497 | api_url: `${this.env.PUBLIC_URL}/api/v3`, |
| 498 | graphql_url: `${this.env.PUBLIC_URL}/api/graphql`, |
| 499 | token, |
| 500 | retention_days: "90", |
| 501 | secret_source: "Actions", |
| 502 | }; |
| 503 | } |
| 504 | |
| 505 | private async createFailedRun(path: string, event: string, ref: string, sha: string, caller: Actor, errors: string[]): Promise<void> { |
| 506 | const id = newId(); |
| 507 | const t = now(); |
| 508 | this.sql.exec( |
| 509 | "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', ?, ?, ?, ?)", |
| 510 | id, |
| 511 | this.nextRunNumber(), |
| 512 | path, |
| 513 | path, |
| 514 | event, |
| 515 | ref, |
| 516 | sha, |
| 517 | JSON.stringify({ id: caller.id, login: caller.login, kind: caller.kind }), |
| 518 | t, |
| 519 | t, |
| 520 | "{}", |
| 521 | `Invalid workflow file: ${errors[0] ?? path}`, |
| 522 | ); |
| 523 | const jobId = newId(); |
| 524 | this.sql.exec( |
| 525 | "INSERT INTO jobs (id, run_id, job_key, name, status, conclusion, started_at, completed_at, failure) VALUES (?, ?, 'invalid', 'Invalid workflow file', 'completed', 'failure', ?, ?, ?)", |
| 526 | jobId, |
| 527 | id, |
| 528 | t, |
| 529 | t, |
| 530 | JSON.stringify({ message: errors.join("\n") }), |
| 531 | ); |
| 532 | } |
| 533 | |
| 534 | async createRun( |
| 535 | wf: Workflow, |
| 536 | o: { |
| 537 | event: string; |
| 538 | ref: string; |
| 539 | sha: string; |
| 540 | headSha?: string; |
| 541 | payload: Record<string, unknown>; |
| 542 | caller: Caller | { id: string; login: string; kind: Actor["kind"] }; |
| 543 | prNumber?: number; |
| 544 | inputs?: Record<string, unknown>; |
| 545 | headRef?: string; |
| 546 | baseRef?: string; |
| 547 | title: string; |
| 548 | /** Started by a run whose code came from an agent (a re-run, or a workflow_run follower). */ |
| 549 | agentOrigin?: boolean; |
| 550 | /** Pull requests whose code the run includes (a merge group's). */ |
| 551 | prNumbers?: number[]; |
| 552 | }, |
| 553 | ): Promise<RunInfo | null> { |
| 554 | // Like GitHub, what a job does with its GITHUB_TOKEN (pushes, pull requests) starts no new |
| 555 | // runs, so workflows cannot trigger themselves in a loop; workflow_dispatch and |
| 556 | // repository_dispatch are the exceptions. |
| 557 | if (isJobCaller(o.caller) && o.event !== "workflow_dispatch" && o.event !== "repository_dispatch") return null; |
| 558 | // A dispatch loop (or a flood of pushes) stops growing here. |
| 559 | const active = this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM runs WHERE status != 'completed'").one().n; |
| 560 | if (active >= MAX_ACTIVE_RUNS) { |
| 561 | if (o.event === "workflow_dispatch" || o.event === "repository_dispatch") throw new HttpError(429, `this repository has ${active} unfinished workflow runs, the most allowed; wait for some to finish`); |
| 562 | console.warn(`not starting ${wf.path}: ${active} unfinished runs`); |
| 563 | return null; |
| 564 | } |
| 565 | const id = newId(); |
| 566 | const t = now(); |
| 567 | const run: RunRow = { |
| 568 | id, |
| 569 | run_number: this.nextRunNumber(), |
| 570 | workflow_path: wf.path, |
| 571 | workflow_name: wf.name ?? wf.path, |
| 572 | event: o.event, |
| 573 | ref: o.ref, |
| 574 | sha: o.sha, |
| 575 | actor: JSON.stringify({ id: o.caller.id, login: o.caller.login, kind: o.caller.kind }), |
| 576 | status: "queued", |
| 577 | conclusion: null, |
| 578 | created_at: t, |
| 579 | updated_at: t, |
| 580 | pr_number: o.prNumber ?? null, |
| 581 | attempt: 1, |
| 582 | 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)), |
| 583 | inputs: JSON.stringify(o.inputs ?? {}), |
| 584 | concurrency_group: null, |
| 585 | display_title: o.title, |
| 586 | head_sha: o.headSha && o.headSha !== o.sha ? o.headSha : null, |
| 587 | agent_origin: this.repo.policy.agentOrigin({ |
| 588 | caller: o.caller, |
| 589 | ref: o.ref, |
| 590 | prNumbers: [...(o.prNumber !== undefined ? [o.prNumber] : []), ...(o.prNumbers ?? [])], |
| 591 | inherited: o.agentOrigin, |
| 592 | }) |
| 593 | ? 1 |
| 594 | : 0, |
| 595 | }; |
| 596 | const ctx: ExprContext = { github: this.githubContext(run, "", ""), inputs: o.inputs ?? {}, vars: this.variables() }; |
| 597 | try { |
| 598 | if (wf.runName) run.display_title = interpolate(wf.runName, ctx); |
| 599 | if (wf.concurrency) run.concurrency_group = interpolate(wf.concurrency.group, ctx); |
| 600 | } catch (e) { |
| 601 | console.warn("run-name/concurrency evaluation failed", e); |
| 602 | } |
| 603 | this.sql.exec( |
| 604 | "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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", |
| 605 | run.id, |
| 606 | run.run_number, |
| 607 | run.workflow_path, |
| 608 | run.workflow_name, |
| 609 | run.event, |
| 610 | run.ref, |
| 611 | run.sha, |
| 612 | run.actor, |
| 613 | run.status, |
| 614 | run.created_at, |
| 615 | run.updated_at, |
| 616 | run.pr_number, |
| 617 | run.attempt, |
| 618 | run.event_payload, |
| 619 | run.inputs, |
| 620 | run.concurrency_group, |
| 621 | run.display_title, |
| 622 | run.head_sha, |
| 623 | run.agent_origin ?? 0, |
| 624 | ); |
| 625 | if (run.concurrency_group && wf.concurrency) { |
| 626 | const cancelInProgress = evaluateValue(wf.concurrency.cancelInProgress, ctx); |
| 627 | const others = this.sql.exec<{ id: string }>("SELECT id FROM runs WHERE concurrency_group = ? AND status != 'completed' AND id != ?", run.concurrency_group, id).toArray(); |
| 628 | 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"); |
| 629 | } |
| 630 | await this.env.RUN_WORKFLOW.create({ id, params: { repoId: this.repo.meta.repoId, runId: id } }); |
| 631 | const m = this.repo.meta; |
| 632 | 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 }); |
| 633 | this.repo.broadcast("runs", { action: "created", runId: id }); |
| 634 | await this.onWorkflowRun(run, "requested"); |
| 635 | return toRun(run); |
| 636 | } |
| 637 | |
| 638 | /** A run by id, or by its number (github.run_id; the latest attempt). */ |
| 639 | private runRow(id: string): RunRow { |
| 640 | const r = /^\d+$/.test(id) |
| 641 | ? this.sql.exec<RunRow>("SELECT * FROM runs WHERE run_number = ? ORDER BY attempt DESC, created_at DESC LIMIT 1", Number(id)).toArray()[0] |
| 642 | : this.sql.exec<RunRow>("SELECT * FROM runs WHERE id = ?", id).toArray()[0]; |
| 643 | if (!r) throw new HttpError(404, "run not found"); |
| 644 | return r; |
| 645 | } |
| 646 | |
| 647 | private jobRow(id: string): JobRow { |
| 648 | const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", id).toArray()[0]; |
| 649 | if (!j) throw new HttpError(404, "job not found"); |
| 650 | return j; |
| 651 | } |
| 652 | |
| 653 | listRuns(opts: { workflow?: string; branch?: string; status?: string; event?: string; sha?: string; prNumber?: number; limit?: number; offset?: number } = {}): { runs: RunInfo[]; total: number } { |
| 654 | const where: string[] = []; |
| 655 | const args: unknown[] = []; |
| 656 | if (opts.workflow) { |
| 657 | where.push("(workflow_path = ? OR workflow_path LIKE ?)"); |
| 658 | args.push(opts.workflow, `%/${opts.workflow}`); |
| 659 | } |
| 660 | if (opts.branch) { |
| 661 | where.push("ref = ?"); |
| 662 | args.push(`refs/heads/${opts.branch}`); |
| 663 | } |
| 664 | if (opts.status) { |
| 665 | where.push(opts.status === "completed" || opts.status === "queued" || opts.status === "in_progress" ? "status = ?" : "conclusion = ?"); |
| 666 | args.push(opts.status); |
| 667 | } |
| 668 | if (opts.event) { |
| 669 | where.push("event = ?"); |
| 670 | args.push(opts.event); |
| 671 | } |
| 672 | if (opts.sha) { |
| 673 | where.push("(sha = ? OR head_sha = ?)"); |
| 674 | args.push(opts.sha, opts.sha); |
| 675 | } |
| 676 | if (opts.prNumber) { |
| 677 | where.push("pr_number = ?"); |
| 678 | args.push(opts.prNumber); |
| 679 | } |
| 680 | const w = where.length ? `WHERE ${where.join(" AND ")}` : ""; |
| 681 | const total = this.sql.exec<{ n: number }>(`SELECT count(*) AS n FROM runs ${w}`, ...args).one().n; |
| 682 | 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(); |
| 683 | return { runs: rows.map(toRun), total }; |
| 684 | } |
| 685 | |
| 686 | getRun(id: string): RunInfo & { jobs: JobInfo[] } { |
| 687 | const run = this.runRow(id); |
| 688 | return { ...toRun(run), jobs: this.jobs(run.id) }; |
| 689 | } |
| 690 | |
| 691 | jobs(runId: string): JobInfo[] { |
| 692 | if (/^\d+$/.test(runId)) runId = this.runRow(runId).id; |
| 693 | return this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? ORDER BY rowid", runId).toArray().map(toJob); |
| 694 | } |
| 695 | |
| 696 | getJob(id: string): JobInfo { |
| 697 | return toJob(this.jobRow(id)); |
| 698 | } |
| 699 | |
| 700 | async logs(jobId: string, afterSeq = -1): Promise<{ lines: Array<LogLine & { seq: number }>; complete: boolean }> { |
| 701 | const j = this.jobRow(jobId); |
| 702 | 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(); |
| 703 | if (rows.length === 0 && j.status === "completed") { |
| 704 | const archived = await getFirst(this.env.BUCKET, logKeys(this.repo.meta.repoId, jobId)); |
| 705 | if (archived && afterSeq < 0) return { lines: (await archived.json()) as Array<LogLine & { seq: number }>, complete: true }; |
| 706 | } |
| 707 | return { lines: rows, complete: j.status === "completed" }; |
| 708 | } |
| 709 | |
| 710 | private logKey(jobId: string) { |
| 711 | return logKeys(this.repo.meta.repoId, jobId)[0]; |
| 712 | } |
| 713 | |
| 714 | // ---------------------------------------------------------------- orchestration (called by RunWorkflow) |
| 715 | |
| 716 | /** |
| 717 | * The run's job graph: each job with the jobs it waits for (reusable workflow calls |
| 718 | * expanded). RunWorkflow starts a job as soon as its own needs have finished. |
| 719 | */ |
| 720 | async plan(runId: string): Promise<{ jobs: JobNode[]; error?: string; concurrencyBlocked: boolean }> { |
| 721 | const run = this.runRow(runId); |
| 722 | if (run.status === "completed") return { jobs: [], error: "run already completed", concurrencyBlocked: false }; |
| 723 | let exp: Expanded; |
| 724 | try { |
| 725 | exp = await this.expanded(run); |
| 726 | } catch (e) { |
| 727 | return { jobs: [], error: (e as Error).message, concurrencyBlocked: false }; |
| 728 | } |
| 729 | const { order, errors } = jobOrder(exp.workflow); |
| 730 | if (errors.length) return { jobs: [], error: errors.map((e) => e.message).join("; "), concurrencyBlocked: false }; |
| 731 | const blocked = |
| 732 | !!run.concurrency_group && |
| 733 | 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; |
| 734 | if (!blocked) this.sql.exec("UPDATE runs SET status = 'in_progress', updated_at = ? WHERE id = ? AND status = 'queued'", now(), runId); |
| 735 | return { jobs: order.flat().map((key) => ({ key, needs: exp.workflow.jobs[key]!.needs })), concurrencyBlocked: blocked }; |
| 736 | } |
| 737 | |
| 738 | /** The run's workflow with reusable workflow calls expanded (see ./reusable.ts). */ |
| 739 | private async expanded(run: RunRow): Promise<Expanded> { |
| 740 | const wf = (await this.loadWorkflows(run.sha)).find((w) => w.path === run.workflow_path); |
| 741 | if (!wf?.workflow) throw new ReusableError(`workflow file not found or invalid: ${wf?.errors.join("; ") || run.workflow_path}`); |
| 742 | return expandReusable(wf.workflow, { repoId: null, sha: run.sha }, (uses, from) => this.loadCalled(uses, from)); |
| 743 | } |
| 744 | |
| 745 | /** Load a called workflow: `./path` from the same commit, or `owner/repo/path@ref` from a repository on this instance. */ |
| 746 | private async loadCalled(uses: string, from: Source): Promise<{ workflow: Workflow; source: Source }> { |
| 747 | const ref = parseCallRef(uses); |
| 748 | const m = this.repo.meta; |
| 749 | let source: Source; |
| 750 | let text: string | null; |
| 751 | if (ref.kind === "local" && from.repoId === null) { |
| 752 | source = from; |
| 753 | const b = await this.repo.blob(from.sha, ref.path); |
| 754 | text = b && !b.binary ? fromUtf8(b.content) : null; |
| 755 | } else { |
| 756 | let repoId: string; |
| 757 | if (ref.kind === "local") repoId = from.repoId!; |
| 758 | else if (ref.owner.toLowerCase() === m.owner.toLowerCase() && ref.repo.toLowerCase() === m.name.toLowerCase()) repoId = m.repoId; |
| 759 | else { |
| 760 | const dir = new Directory(this.env); |
| 761 | // A transferred or renamed repository's workflows are found under the old name too. |
| 762 | const target = (await dir.repo(ref.owner, ref.repo)) ?? (await dir.movedRepo(ref.owner, ref.repo)); |
| 763 | // Like GitHub: public workflows can be shared; private ones only within the same owner. |
| 764 | 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}`); |
| 765 | repoId = target.id; |
| 766 | } |
| 767 | if (repoId === m.repoId) { |
| 768 | const sha = ref.kind === "local" ? from.sha : await this.repo.resolve(ref.ref); |
| 769 | if (!sha) throw new Error(`unknown ref ${ref.kind === "remote" ? ref.ref : ""}`); |
| 770 | source = { repoId: null, sha }; |
| 771 | const b = await this.repo.blob(sha, ref.path); |
| 772 | text = b && !b.binary ? fromUtf8(b.content) : null; |
| 773 | } else { |
| 774 | const stub = this.env.REPO.get(this.env.REPO.idFromName(repoId)); |
| 775 | const sha = ref.kind === "local" ? from.sha : await stub.resolve(ref.ref); |
| 776 | if (!sha) throw new Error(`unknown ref ${ref.kind === "remote" ? ref.ref : ""} in ${ref.kind === "remote" ? `${ref.owner}/${ref.repo}` : "the called repository"}`); |
| 777 | source = { repoId, sha }; |
| 778 | const b = await stub.blob(sha, ref.path); |
| 779 | text = b && !b.binary ? fromUtf8(b.content) : null; |
| 780 | } |
| 781 | } |
| 782 | if (text === null) throw new Error(`${uses} not found`); |
| 783 | const { workflow, errors } = parseWorkflow(text, ref.path); |
| 784 | const fatal = errors.filter((e) => !isWarning(e)); |
| 785 | if (!workflow || fatal.length) throw new Error(`${uses} is invalid: ${fatal.map((e) => `${e.path}: ${e.message}`).join("; ")}`); |
| 786 | return { workflow, source }; |
| 787 | } |
| 788 | |
| 789 | /** `needs` context for a job or call: keys are as written inside its own workflow. */ |
| 790 | private async needsResults(run: RunRow, exp: Expanded, keys: string[], scope: string | null): Promise<Record<string, NeedsResult>> { |
| 791 | const out: Record<string, NeedsResult> = {}; |
| 792 | for (const key of keys) out[localKey(key, scope)] = await this.resultOf(run, exp, key); |
| 793 | return out; |
| 794 | } |
| 795 | |
| 796 | /** Status functions for the `if:` of a job or call: from the results of all its ancestors, like GitHub. */ |
| 797 | private statusOf(run: RunRow, exp: Expanded, key: string): StatusFns { |
| 798 | const results = ancestors(exp.needs, exp.needs.get(key) ?? []).map((k) => combineResults(this.rowsOf(run, exp, k))); |
| 799 | return jobStatus(results, run.conclusion === "cancelled"); |
| 800 | } |
| 801 | |
| 802 | /** The job rows of a job, or of every job inside a reusable workflow call. */ |
| 803 | private rowsOf(run: RunRow, exp: Expanded, key: string): Array<JobRow & { continueOnError: boolean }> { |
| 804 | const rows = exp.calls.has(key) |
| 805 | ? this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND substr(job_key, 1, ?) = ?", run.id, key.length + 1, `${key}/`).toArray() |
| 806 | : this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND job_key = ?", run.id, key).toArray(); |
| 807 | return rows.map((r) => ({ ...r, continueOnError: !!r.continue_on_error })); |
| 808 | } |
| 809 | |
| 810 | /** Result and outputs of a job, or of a reusable workflow call (from its `on.workflow_call.outputs`). */ |
| 811 | private async resultOf(run: RunRow, exp: Expanded, key: string): Promise<NeedsResult> { |
| 812 | const call = exp.calls.get(key); |
| 813 | const rows = this.rowsOf(run, exp, key); |
| 814 | const result = combineResults(rows); |
| 815 | let outputs: Record<string, string> = {}; |
| 816 | if (!call) for (const r of rows) Object.assign(outputs, JSON.parse(r.outputs)); |
| 817 | if (call && Object.keys(call.workflow.on.workflow_call?.outputs ?? {}).length) { |
| 818 | const jobs: Record<string, NeedsResult> = {}; |
| 819 | for (const k of Object.keys(call.workflow.jobs)) jobs[k] = await this.resultOf(run, exp, `${key}/${k}`); |
| 820 | const { inputs } = await this.callContext(run, exp, key, false); |
| 821 | const ctx: ExprContext = { github: this.githubContext(run, localKey(key, call.parent), ""), jobs, inputs, vars: this.variables() }; |
| 822 | outputs = {}; |
| 823 | for (const [name, d] of Object.entries(call.workflow.on.workflow_call!.outputs!)) { |
| 824 | try { |
| 825 | const v = evaluateValue(d.value, ctx); |
| 826 | outputs[name] = v === undefined || v === null ? "" : typeof v === "object" ? JSON.stringify(v) : String(v); |
| 827 | } catch { |
| 828 | outputs[name] = ""; |
| 829 | } |
| 830 | } |
| 831 | } |
| 832 | return { result, outputs }; |
| 833 | } |
| 834 | |
| 835 | /** |
| 836 | * `inputs` and `secrets` inside a reusable workflow call (or at the top level when |
| 837 | * `callKey` is null): inputs from the caller's `with` and the declared defaults, |
| 838 | * secrets from `secrets: inherit` or the caller's mapping. |
| 839 | */ |
| 840 | private async callContext(run: RunRow, exp: Expanded, callKey: string | null, withSecrets: boolean): Promise<{ inputs: Record<string, unknown>; secrets: Record<string, string> }> { |
| 841 | if (callKey === null) return { inputs: JSON.parse(run.inputs), secrets: withSecrets ? { ...(await this.orgSecrets()), ...(await this.repoSecrets()) } : {} }; |
| 842 | const call = exp.calls.get(callKey)!; |
| 843 | const parent = await this.callContext(run, exp, call.parent, withSecrets); |
| 844 | 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() }; |
| 845 | const inputs: Record<string, unknown> = {}; |
| 846 | for (const [k, d] of Object.entries(call.workflow.on.workflow_call?.inputs ?? {})) { |
| 847 | let v = call.job.with && k in call.job.with ? evaluateValue(call.job.with[k], ctx) : evaluateValue(d.default, ctx); |
| 848 | if (v === undefined || v === null || v === "") { |
| 849 | if (d.required) throw new ReusableError(`jobs.${callKey}: input '${k}' is required by ${call.job.uses}`); |
| 850 | v = d.type === "boolean" ? false : d.type === "number" ? 0 : ""; |
| 851 | } |
| 852 | inputs[k] = d.type === "boolean" ? v === true || v === "true" : d.type === "number" ? Number(v) : v; |
| 853 | } |
| 854 | const secrets: Record<string, string> = {}; |
| 855 | if (withSecrets) { |
| 856 | if (call.job.secrets === "inherit") Object.assign(secrets, parent.secrets); |
| 857 | else for (const [name, expr] of Object.entries(call.job.secrets ?? {})) secrets[name.toUpperCase()] = String(evaluateValue(expr, { ...ctx, secrets: parent.secrets }) ?? ""); |
| 858 | for (const [name, d] of Object.entries(call.workflow.on.workflow_call?.secrets ?? {})) { |
| 859 | if (d.required && !secrets[name.toUpperCase()]) throw new ReusableError(`jobs.${callKey}: secret '${name}' is required by ${call.job.uses}`); |
| 860 | } |
| 861 | } |
| 862 | return { inputs, secrets }; |
| 863 | } |
| 864 | |
| 865 | /** |
| 866 | * Start every matrix instance of a job. Returns the ids the workflow must wait |
| 867 | * for (instances beyond max-parallel are queued and started as others finish). |
| 868 | */ |
| 869 | async startJobKey(runId: string, jobKey: string): Promise<{ jobIds: string[] }> { |
| 870 | const run = this.runRow(runId); |
| 871 | const exp = await this.expanded(run); |
| 872 | const job = exp.workflow.jobs[jobKey]!; |
| 873 | const owner = exp.workflowOf.get(jobKey)!; |
| 874 | const parent = exp.parentOf.get(jobKey) ?? null; |
| 875 | const chain: string[] = []; |
| 876 | for (let c = parent; c; c = exp.calls.get(c)!.parent) chain.unshift(c); |
| 877 | const prefix = chain.map((c) => exp.calls.get(c)!.job.name ?? localKey(c, exp.calls.get(c)!.parent)).join(" / "); |
| 878 | const label = (name: string) => (prefix ? `${prefix} / ${name}` : name); |
| 879 | const fail = (name: string, message: string, conclusion: Conclusion = "failure") => { |
| 880 | const id = newId(); |
| 881 | const t = now(); |
| 882 | 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); |
| 883 | this.repo.broadcast(`run:${runId}`, { action: "job", jobId: id }); |
| 884 | return { jobIds: [] as string[] }; |
| 885 | }; |
| 886 | const short = localKey(jobKey, parent); |
| 887 | let needs: Record<string, NeedsResult>; |
| 888 | let baseCtx: ExprContext; |
| 889 | let shouldRun: boolean; |
| 890 | try { |
| 891 | // The callers decide first: a skipped call, or one whose ancestors failed or were skipped, skips everything inside it. |
| 892 | for (const c of chain) { |
| 893 | const call = exp.calls.get(c)!; |
| 894 | const callNeeds = await this.needsResults(run, exp, call.needs, call.parent); |
| 895 | const outer = await this.callContext(run, exp, call.parent, false); |
| 896 | const callCtx: ExprContext = { github: this.githubContext(run, localKey(c, call.parent), ""), needs: callNeeds, inputs: outer.inputs, vars: this.variables() }; |
| 897 | if (!evaluateIf(call.job.if, callCtx, this.statusOf(run, exp, c))) return fail(label(job.name ?? short), "", "skipped"); |
| 898 | } |
| 899 | // Resolve the call's inputs and secrets now, so a missing required one fails here rather than in the runner. |
| 900 | if (parent) await this.callContext(run, exp, parent, true); |
| 901 | needs = await this.needsResults(run, exp, exp.needs.get(jobKey) ?? [], parent); |
| 902 | baseCtx = { github: this.githubContext(run, short, ""), needs, inputs: (await this.callContext(run, exp, parent, false)).inputs, vars: this.variables(), env: owner.env }; |
| 903 | shouldRun = evaluateIf(job.if, baseCtx, this.statusOf(run, exp, jobKey)); |
| 904 | } catch (e) { |
| 905 | return fail(label(job.name ?? short), e instanceof ReusableError ? e.message : `Error evaluating 'if': ${(e as Error).message}`); |
| 906 | } |
| 907 | if (!shouldRun) return fail(label(job.name ?? short), "", "skipped"); |
| 908 | |
| 909 | let combos: Array<Record<string, unknown>>; |
| 910 | try { |
| 911 | combos = expandMatrix(job.strategy?.matrix, baseCtx); |
| 912 | } catch (e) { |
| 913 | return fail(label(job.name ?? short), `Error expanding matrix: ${(e as Error).message}`); |
| 914 | } |
| 915 | if (combos.length > 256) return fail(label(job.name ?? short), "A matrix may produce at most 256 jobs."); |
| 916 | const failFast = evaluateValue(job.strategy?.failFast ?? true, baseCtx) !== false; |
| 917 | const maxParallel = Number(evaluateValue(job.strategy?.maxParallel ?? combos.length, baseCtx)) || combos.length; |
| 918 | const ids: string[] = []; |
| 919 | combos.forEach((matrix, i) => { |
| 920 | // Like GitHub, name, runs-on, timeout-minutes, continue-on-error and concurrency are evaluated for each matrix leg. |
| 921 | const ctx = { ...baseCtx, matrix, strategy: { "fail-fast": failFast, "job-index": i, "job-total": combos.length, "max-parallel": maxParallel } }; |
| 922 | let name = short; |
| 923 | if (!job.name && Object.keys(matrix).length) name = `${short} (${Object.values(matrix).map((v) => (typeof v === "object" ? JSON.stringify(v) : String(v))).join(", ")})`; |
| 924 | let leg: { labels: string[]; timeout: number; continueOnError: boolean; group: string | null; cancel: boolean; environment: string | null }; |
| 925 | try { |
| 926 | if (job.name) name = interpolate(job.name, ctx); |
| 927 | const continueOnError = evaluateValue(job.continueOnError, ctx); |
| 928 | const cancel = job.concurrency ? evaluateValue(job.concurrency.cancelInProgress, ctx) : false; |
| 929 | leg = { |
| 930 | labels: runnerLabels(job.runsOn, ctx), |
| 931 | timeout: Number(evaluateValue(job.timeoutMinutes, ctx)) || 360, |
| 932 | continueOnError: continueOnError === true || continueOnError === "true", |
| 933 | group: (job.concurrency ? interpolate(job.concurrency.group, ctx).trim() : "") || null, |
| 934 | cancel: cancel === true || cancel === "true", |
| 935 | // Evaluated once, here: its protection rules gate the launch, and runnerSpec deploys to this same name. |
| 936 | environment: job.environment ? interpolate(job.environment.name, ctx).trim() : null, |
| 937 | }; |
| 938 | } catch (e) { |
| 939 | fail(label(name), `Error evaluating the job for this matrix combination: ${(e as Error).message}`); |
| 940 | return; |
| 941 | } |
| 942 | const unsupported = leg.labels.filter((l) => !SUPPORTED_LABEL.test(l)); |
| 943 | const selfHosted = leg.labels.some((l) => l.toLowerCase() === "self-hosted"); |
| 944 | if (!leg.labels.length || (unsupported.length && !selfHosted)) { |
| 945 | 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.`); |
| 946 | return; |
| 947 | } |
| 948 | const id = newId(); |
| 949 | this.sql.exec( |
| 950 | "INSERT INTO jobs (id, run_id, job_key, name, status, matrix, strategy, timeout_minutes, labels, continue_on_error, concurrency_group, concurrency_cancel, environment) VALUES (?, ?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?, ?)", |
| 951 | id, |
| 952 | runId, |
| 953 | jobKey, |
| 954 | label(name), |
| 955 | JSON.stringify(matrix), |
| 956 | JSON.stringify(ctx.strategy), |
| 957 | Math.min(leg.timeout, 360), |
| 958 | selfHosted ? JSON.stringify(leg.labels.map((l) => l.toLowerCase())) : null, |
| 959 | leg.continueOnError ? 1 : 0, |
| 960 | leg.group, |
| 961 | leg.cancel ? 1 : 0, |
| 962 | leg.environment, |
| 963 | ); |
| 964 | ids.push(id); |
| 965 | }); |
| 966 | for (const id of ids.slice(0, maxParallel)) await this.launch(id); |
| 967 | this.repo.broadcast(`run:${runId}`, { action: "jobs" }); |
| 968 | return { jobIds: ids }; |
| 969 | } |
| 970 | |
| 971 | private checkName(run: RunRow, job: JobRow): string { |
| 972 | return `${run.workflow_name} / ${job.name}`; |
| 973 | } |
| 974 | |
| 975 | private async launch(jobId: string): Promise<void> { |
| 976 | const job = this.jobRow(jobId); |
| 977 | if (job.status === "completed" || !(await this.gate(job)) || !(await this.concurrencySlot(job))) return; |
| 978 | const run = this.runRow(job.run_id); |
| 979 | const m = this.repo.meta; |
| 980 | if (job.labels) { |
| 981 | // Self-hosted: park the job until a registered runner with matching labels claims it. |
| 982 | this.sql.exec("UPDATE jobs SET status = 'waiting', container = NULL WHERE id = ?", jobId); |
| 983 | 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}` }); |
| 984 | return; |
| 985 | } |
| 986 | // One repository can't take the whole fleet: past its share, the job waits for one of its own to finish. |
| 987 | const busy = this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM jobs WHERE labels IS NULL AND container IS NOT NULL AND status IN ('waiting', 'in_progress') AND id != ?", jobId).one().n; |
| 988 | if (busy >= MAX_HOSTED_JOBS) { |
| 989 | this.sql.exec("UPDATE jobs SET status = 'queued', fleet_held = 1 WHERE id = ?", jobId); |
| 990 | 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}` }); |
| 991 | return; |
| 992 | } |
| 993 | this.sql.exec("UPDATE jobs SET fleet_held = 0 WHERE id = ?", jobId); |
| 994 | const token = `djt_${m.repoId}_${jobId}_${randomToken(24)}`; |
| 995 | const container = `job-${jobId}`; |
| 996 | this.sql.exec("UPDATE jobs SET status = 'waiting', token_hash = ?, container = ? WHERE id = ?", await sha256(token), container, jobId); |
| 997 | 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}` }); |
| 998 | try { |
| 999 | const stub = this.env.RUNNER.get(this.env.RUNNER.idFromName(container)); |
| 1000 | await stub.startJob({ repoId: m.repoId, jobId, token, apiUrl: this.env.PUBLIC_URL }); |
| 1001 | } catch (e) { |
| 1002 | const message = (e as Error).message; |
| 1003 | if (message.includes(NO_INSTANCE)) return this.placementFailed(jobId, message); |
| 1004 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `Could not start a runner: ${message}` } }); |
| 1005 | } |
| 1006 | } |
| 1007 | |
| 1008 | |
| 1009 | // ---------------------------------------------------------------- environment protection rules |
| 1010 | |
| 1011 | private jobUrl(run: RunRow, jobId: string): string { |
| 1012 | const m = this.repo.meta; |
| 1013 | return `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}`; |
| 1014 | } |
| 1015 | |
| 1016 | /** |
| 1017 | * Whose run this is, for prevent_self_review, as account ids (logins can be renamed): who started |
| 1018 | * it and its earlier attempts, and for each of them that is an agent, the account that owns it. |
| 1019 | */ |
| 1020 | async selfIds(run: RunRow): Promise<Set<string>> { |
| 1021 | const actor = JSON.parse(run.actor) as Actor; |
| 1022 | const ids = new Set([actor.id, ...((run.prior_actors ? JSON.parse(run.prior_actors) : []) as string[])].filter(Boolean)); |
| 1023 | const dir = new Directory(this.env); |
| 1024 | for (const id of [...ids]) { |
| 1025 | const a = await dir.account(id).catch(() => null); |
| 1026 | if (a?.kind === "agent" && a.owner_id) ids.add(a.owner_id); |
| 1027 | } |
| 1028 | return ids; |
| 1029 | } |
| 1030 | |
| 1031 | /** |
| 1032 | * A job whose environment has required reviewers or a wait timer waits here, before a token is |
| 1033 | * minted or a runner started, in the status `approval`: no concurrency slot, no hosted slot. |
| 1034 | * Returns whether the job may go on to start. The branch policy is checked first, so a ref that |
| 1035 | * may never deploy fails at once instead of asking anyone. |
| 1036 | */ |
| 1037 | private async gate(job: JobRow): Promise<boolean> { |
| 1038 | if (!job.environment || job.gate_state === "released") return true; |
| 1039 | const rules = this.repo.deploys.protection(job.environment); |
| 1040 | if (!rules) { |
| 1041 | if (job.status === "approval") this.release(job); |
| 1042 | return true; |
| 1043 | } |
| 1044 | const run = this.runRow(job.run_id); |
| 1045 | const fail = async (message: string) => { |
| 1046 | await this.completeJob(job.id, { conclusion: "failure", outputs: {}, failure: { message } }); |
| 1047 | return false; |
| 1048 | }; |
| 1049 | if (!this.repo.deploys.refAllowed(job.environment, run.ref)) return fail(`${run.ref.replace(/^refs\/(heads|tags)\//, "")} is not allowed to deploy to ${job.environment} due to environment protection rules.`); |
| 1050 | const decision = rules.reviewers.length ? this.repo.deploys.decision(run.id, job.environment) : null; |
| 1051 | if (decision === "rejected") { |
| 1052 | const r = this.repo.deploys.reviews(run.id).find((x) => x.environment.toLowerCase() === job.environment!.toLowerCase() && x.state === "rejected")!; |
| 1053 | return fail(`The deployment to ${job.environment} was rejected by ${r.user.login}${r.comment ? `: ${r.comment}` : "."}`); |
| 1054 | } |
| 1055 | let since = job.gate_since; |
| 1056 | if (job.status !== "approval") { |
| 1057 | since = now(); |
| 1058 | this.sql.exec("UPDATE jobs SET status = 'approval', gate_since = ? WHERE id = ?", since, job.id); |
| 1059 | this.sql.exec("UPDATE runs SET status = 'waiting', updated_at = ? WHERE id = ? AND status != 'completed'", since, run.id); |
| 1060 | const actor = JSON.parse(run.actor) as Actor; |
| 1061 | this.repo.deploys.jobWaiting({ id: job.id, ref: run.ref, sha: run.sha, logUrl: this.jobUrl(run, job.id), actor: run.agent_origin ? { ...actor, kind: "agent" } : actor }, job.environment); |
| 1062 | this.repo.upsertCheck({ sha: run.head_sha ?? run.sha, name: this.checkName(run, job), status: "queued", runId: run.id, jobId: job.id, detailsUrl: this.jobUrl(run, job.id) }); |
| 1063 | // Reviewers hear about it once per run and environment, from its first job to wait. |
| 1064 | const first = !this.sql.exec("SELECT 1 FROM jobs WHERE run_id = ? AND environment = ? COLLATE NOCASE AND gate_since IS NOT NULL AND id != ?", run.id, job.environment, job.id).toArray().length; |
| 1065 | if (first && rules.reviewers.length && !decision) this.emitReview("requested", run, job.environment, rules.reviewers, actor, null, ""); |
| 1066 | this.repo.broadcast(`run:${run.id}`, { action: "job", jobId: job.id }); |
| 1067 | this.repo.broadcast("runs", { action: "waiting", runId: run.id }); |
| 1068 | } |
| 1069 | const timerEnds = Date.parse(since!) + rules.waitTimer * 60_000; |
| 1070 | if ((rules.reviewers.length && decision !== "approved") || Date.now() < timerEnds) return false; |
| 1071 | this.release({ ...job, status: "approval" }); |
| 1072 | return true; |
| 1073 | } |
| 1074 | |
| 1075 | /** The job's protection rules passed: it goes on to start, and its run stops waiting once no other job does. */ |
| 1076 | private release(job: JobRow): void { |
| 1077 | this.sql.exec("UPDATE jobs SET gate_state = 'released', status = 'queued' WHERE id = ?", job.id); |
| 1078 | this.syncWaiting(job.run_id); |
| 1079 | } |
| 1080 | |
| 1081 | private syncWaiting(runId: string): void { |
| 1082 | if (this.sql.exec("SELECT 1 FROM jobs WHERE run_id = ? AND status = 'approval'", runId).toArray().length) return; |
| 1083 | this.sql.exec("UPDATE runs SET status = 'in_progress', updated_at = ? WHERE id = ? AND status = 'waiting'", now(), runId); |
| 1084 | } |
| 1085 | |
| 1086 | private async wakeGate(runId: string, jobId: string): Promise<void> { |
| 1087 | try { |
| 1088 | await (await this.env.RUN_WORKFLOW.get(runId)).sendEvent({ type: `gate-${jobId}`, payload: {} }); |
| 1089 | } catch (e) { |
| 1090 | console.error("failed to wake a job waiting for approval", e); |
| 1091 | } |
| 1092 | } |
| 1093 | |
| 1094 | private emitReview(action: "requested" | "approved" | "rejected", run: RunRow, environment: string, reviewers: Reviewer[], by: Actor, state: string | null, comment: string): void { |
| 1095 | const m = this.repo.meta; |
| 1096 | const url = `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}`; |
| 1097 | const ids = Object.fromEntries(this.repo.deploys.environments().map((e) => [e.name.toLowerCase(), e.id])); |
| 1098 | this.repo.emit({ |
| 1099 | type: "deployment_review", |
| 1100 | action, |
| 1101 | repoId: m.repoId, |
| 1102 | owner: m.owner, |
| 1103 | repo: m.name, |
| 1104 | payload: { |
| 1105 | since: now(), |
| 1106 | environment: environment, |
| 1107 | environment_id: ids[environment.toLowerCase()] ?? null, |
| 1108 | comment, |
| 1109 | ...(state ? { state } : {}), |
| 1110 | workflow_run: { id: run.id, run_number: run.run_number, name: run.workflow_name, display_title: run.display_title, head_sha: run.head_sha ?? run.sha, head_branch: run.ref.replace(/^refs\/(heads|tags)\//, ""), html_url: url }, |
| 1111 | reviewers: reviewers.map((r) => ({ type: "User", reviewer: { login: r.login, id: r.id } })), |
| 1112 | ...(action === "requested" ? { requestor: { login: by.login } } : { approver: { login: by.login } }), |
| 1113 | }, |
| 1114 | actor: by, |
| 1115 | at: now(), |
| 1116 | }); |
| 1117 | } |
| 1118 | |
| 1119 | /** |
| 1120 | * Where a job waiting on its environment stands, for the run's Workflow: what it waits for and |
| 1121 | * until when. Re-checks the rules first, so a job whose time came starts here. |
| 1122 | */ |
| 1123 | async jobGate(jobId: string): Promise<{ status: JobRow["status"]; wait: "review" | "timer" | null; until: string | null }> { |
| 1124 | let job = this.jobRow(jobId); |
| 1125 | if (job.status === "approval") await this.launch(jobId); |
| 1126 | job = this.jobRow(jobId); |
| 1127 | if (job.status !== "approval") return { status: job.status, wait: null, until: null }; |
| 1128 | const rules = this.repo.deploys.protection(job.environment!) ?? { reviewers: [], waitTimer: 0, preventSelfReview: false }; |
| 1129 | const since = Date.parse(job.gate_since!); |
| 1130 | const deadline = since + APPROVAL_DAYS * 86400_000; |
| 1131 | if (Date.now() >= deadline) { |
| 1132 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The deployment to ${job.environment} wasn't approved within ${APPROVAL_DAYS} days.` } }); |
| 1133 | return { status: "completed", wait: null, until: null }; |
| 1134 | } |
| 1135 | const decided = this.repo.deploys.decision(job.run_id, job.environment!) === "approved"; |
| 1136 | if (rules.reviewers.length && !decided) return { status: "approval", wait: "review", until: new Date(deadline).toISOString() }; |
| 1137 | return { status: "approval", wait: "timer", until: new Date(since + rules.waitTimer * 60_000).toISOString() }; |
| 1138 | } |
| 1139 | |
| 1140 | /** The environments a run's jobs wait on, with who may approve them (GitHub's pending deployments). */ |
| 1141 | async pendingDeployments(runId: string, viewer: { id: string; login: string; mayReview?: boolean } | null): Promise<PendingDeployment[]> { |
| 1142 | const run = this.runRow(runId); |
| 1143 | const waiting = this.sql.exec<{ environment: string; since: string }>("SELECT environment, MIN(gate_since) AS since FROM jobs WHERE run_id = ? AND status = 'approval' GROUP BY environment COLLATE NOCASE", run.id).toArray(); |
| 1144 | const self = viewer ? await this.selfIds(run) : new Set<string>(); |
| 1145 | return waiting.flatMap((w) => { |
| 1146 | const env = this.repo.deploys.environments().find((e) => e.name.toLowerCase() === w.environment.toLowerCase()); |
| 1147 | if (!env) return []; |
| 1148 | const p = env.protection; |
| 1149 | const decided = this.repo.deploys.decision(run.id, env.name); |
| 1150 | // A listed reviewer who lost access to the repository no longer decides (the caller says so in mayReview). |
| 1151 | const listed = !!viewer && viewer.mayReview !== false && p.reviewers.some((r) => r.id === viewer.id); |
| 1152 | const selfBlocked = !!viewer && p.preventSelfReview && self.has(viewer.id); |
| 1153 | return [ |
| 1154 | { |
| 1155 | environment: { id: env.id, name: env.name }, |
| 1156 | waitTimer: p.waitTimer, |
| 1157 | waitTimerStartedAt: w.since, |
| 1158 | reviewers: p.reviewers, |
| 1159 | preventSelfReview: p.preventSelfReview, |
| 1160 | needsReview: p.reviewers.length > 0 && decided !== "approved", |
| 1161 | currentUserCanApprove: listed && !selfBlocked && p.reviewers.length > 0 && !decided, |
| 1162 | selfBlocked, |
| 1163 | }, |
| 1164 | ]; |
| 1165 | }); |
| 1166 | } |
| 1167 | |
| 1168 | /** |
| 1169 | * A required reviewer approves or rejects environments a run waits on. The caller checked that |
| 1170 | * `reviewer` is a person signed in or using a write token (never an agent or a job); here the |
| 1171 | * reviewer must be listed on every environment, and not whoever started the run when the |
| 1172 | * environment prevents self-review. One approval is enough. Approved jobs start once their |
| 1173 | * wait timer has also passed; rejected ones fail. |
| 1174 | */ |
| 1175 | async reviewDeployments(runId: string, input: { environmentIds: number[]; state: "approved" | "rejected"; comment: string }, reviewer: { id: string; login: string; mayReview: boolean }): Promise<{ reviews: DeploymentReview[]; deployments: Deployment[] }> { |
| 1176 | const run = this.runRow(runId); |
| 1177 | if (!reviewer.mayReview) throw new HttpError(403, "only people with access to this repository review its deployments"); |
| 1178 | if (!input.environmentIds.length) throw new HttpError(422, "environment_ids must name at least one environment"); |
| 1179 | const pending = await this.pendingDeployments(run.id, reviewer); |
| 1180 | const chosen = input.environmentIds.map((id) => { |
| 1181 | const p = pending.find((x) => x.environment.id === id); |
| 1182 | if (!p || !p.needsReview) throw new HttpError(422, `environment ${id} isn't waiting for a review in this run`); |
| 1183 | if (p.selfBlocked) throw new HttpError(403, `${p.environment.name} doesn't let whoever started a run approve it`); |
| 1184 | if (!p.currentUserCanApprove) throw new HttpError(403, `you aren't a required reviewer for ${p.environment.name}`); |
| 1185 | return p; |
| 1186 | }); |
| 1187 | const by: Actor = { id: reviewer.id, login: reviewer.login, kind: "user" }; |
| 1188 | const out: DeploymentReview[] = []; |
| 1189 | const jobIds: string[] = []; |
| 1190 | for (const p of chosen) { |
| 1191 | out.push(this.repo.deploys.addReview(run.id, p.environment.name, input.state, input.comment, reviewer)); |
| 1192 | this.emitReview(input.state, run, p.environment.name, p.reviewers, by, input.state, input.comment); |
| 1193 | const jobs = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND status = 'approval' AND environment = ? COLLATE NOCASE", run.id, p.environment.name).toArray(); |
| 1194 | for (const j of jobs) { |
| 1195 | jobIds.push(j.id); |
| 1196 | await this.launch(j.id); |
| 1197 | await this.wakeGate(run.id, j.id); |
| 1198 | } |
| 1199 | } |
| 1200 | this.repo.broadcast(`run:${run.id}`, { action: "reviewed" }); |
| 1201 | return { reviews: out, deployments: this.repo.deploys.jobDeployments(jobIds) }; |
| 1202 | } |
| 1203 | |
| 1204 | deploymentReviews(runId: string): DeploymentReview[] { |
| 1205 | return this.repo.deploys.reviews(this.runRow(runId).id); |
| 1206 | } |
| 1207 | |
| 1208 | /** |
| 1209 | * The job's container got no machine (its start failed, or it stopped before the runner reported |
| 1210 | * in). Put the job back in line and have the run's workflow start it again after a pause, or fail it |
| 1211 | * once it has waited MAX_PLACEMENT_TRIES times. |
| 1212 | */ |
| 1213 | private async placementFailed(jobId: string, why: string): Promise<void> { |
| 1214 | const job = this.jobRow(jobId); |
| 1215 | if (job.status === "completed") return; |
| 1216 | const tries = job.placement_tries + 1; |
| 1217 | if (tries > MAX_PLACEMENT_TRIES) { |
| 1218 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `No hosted runner could be started: Diffinery tried ${MAX_PLACEMENT_TRIES} times over about half an hour (${why}). Re-run the job to try again.` } }); |
| 1219 | return; |
| 1220 | } |
| 1221 | this.sql.exec("UPDATE jobs SET status = 'queued', container = NULL, token_hash = NULL, placement_tries = ? WHERE id = ?", tries, jobId); |
| 1222 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId }); |
| 1223 | try { |
| 1224 | const instance = await this.env.RUN_WORKFLOW.get(job.run_id); |
| 1225 | await instance.sendEvent({ type: `job-${jobId}`, payload: { retryInSeconds: placementDelay(tries) } }); |
| 1226 | } catch (e) { |
| 1227 | console.error("could not ask the run's workflow to start the job again", e); |
| 1228 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `Could not start a runner: ${why}` } }); |
| 1229 | } |
| 1230 | } |
| 1231 | |
| 1232 | /** The run's workflow starts a job again after placementFailed put it back in line. */ |
| 1233 | async relaunchJob(jobId: string): Promise<void> { |
| 1234 | if (this.jobRow(jobId).status === "queued") await this.launch(jobId); |
| 1235 | } |
| 1236 | |
| 1237 | /** |
| 1238 | * Job-level `concurrency`, like GitHub: at most one job of a group runs at a time in |
| 1239 | * the repository. With cancel-in-progress the newcomer cancels the group's other jobs; |
| 1240 | * without it the newcomer waits as `pending` and replaces any older pending job. |
| 1241 | * Returns whether the job may start now. |
| 1242 | */ |
| 1243 | private async concurrencySlot(job: JobRow): Promise<boolean> { |
| 1244 | if (!job.concurrency_group) return true; |
| 1245 | 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(); |
| 1246 | const pending = others.filter((o) => o.status === "pending"); |
| 1247 | const running = others.filter((o) => o.status !== "pending"); |
| 1248 | const failure = { message: `Canceling since a higher priority waiting request for '${job.concurrency_group}' exists` }; |
| 1249 | if (running.length && !job.concurrency_cancel) { |
| 1250 | this.sql.exec("UPDATE jobs SET status = 'pending' WHERE id = ?", job.id); |
| 1251 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId: job.id }); |
| 1252 | for (const o of pending) await this.completeJob(o.id, { conclusion: "cancelled", outputs: {}, failure }); |
| 1253 | return false; |
| 1254 | } |
| 1255 | // Take the group before cancelling anything, so a job that starts meanwhile waits for this one. |
| 1256 | this.sql.exec("UPDATE jobs SET status = 'waiting' WHERE id = ?", job.id); |
| 1257 | for (const o of [...pending, ...running]) { |
| 1258 | await this.stopContainer(o); |
| 1259 | await this.completeJob(o.id, { conclusion: "cancelled", outputs: {}, failure }); |
| 1260 | } |
| 1261 | return true; |
| 1262 | } |
| 1263 | |
| 1264 | async jobTimedOut(jobId: string): Promise<void> { |
| 1265 | const j = this.jobRow(jobId); |
| 1266 | if (j.status === "completed") return; |
| 1267 | await this.stopContainer(j); |
| 1268 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The job exceeded its timeout of ${j.timeout_minutes} minutes.` } }, "timed_out"); |
| 1269 | } |
| 1270 | |
| 1271 | async containerStopped(jobId: string, exitCode: number | null): Promise<void> { |
| 1272 | const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", jobId).toArray()[0]; |
| 1273 | if (!j || j.status === "completed") return; |
| 1274 | // Stopped while still waiting for its runner, without an exit code: the container never got a machine. |
| 1275 | if (j.status === "waiting" && !j.labels && !exitCode) return this.placementFailed(jobId, "the container stopped before the runner started"); |
| 1276 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: exitCode ? `The runner stopped unexpectedly (exit code ${exitCode}).` : "The runner's container stopped before the job finished, without an exit code (the platform restarted or replaced it)." } }); |
| 1277 | } |
| 1278 | |
| 1279 | /** |
| 1280 | * A self-hosted runner asks for work. Returns the oldest waiting job whose |
| 1281 | * runs-on labels are all offered by the runner, with a fresh job token. |
| 1282 | */ |
| 1283 | async claimJob(runnerName: string, runnerLabels: string[], caller: Caller): Promise<{ jobId: string; token: string; runId: string; name: string } | null> { |
| 1284 | if (!atLeast(caller.role, "admin")) throw new HttpError(403, "admin access is required to run self-hosted jobs"); |
| 1285 | return this.assignJob(runnerName, runnerLabels, null); |
| 1286 | } |
| 1287 | |
| 1288 | /** The oldest waiting job whose runs-on labels the runner all offers, handed to it with a fresh job token. */ |
| 1289 | private async assignJob(runnerName: string, runnerLabels: string[], runnerId: number | null): Promise<{ jobId: string; token: string; runId: string; name: string } | null> { |
| 1290 | const offered = new Set(["self-hosted", ...runnerLabels.map((l) => l.toLowerCase())]); |
| 1291 | const waiting = this.sql |
| 1292 | .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") |
| 1293 | .toArray(); |
| 1294 | const job = waiting.find((j) => (JSON.parse(j.labels!) as string[]).every((l) => offered.has(l))); |
| 1295 | if (!job) return null; |
| 1296 | const m = this.repo.meta; |
| 1297 | const token = `djt_${m.repoId}_${job.id}_${randomToken(24)}`; |
| 1298 | const container = `self-hosted:${runnerName.slice(0, 64)}`; |
| 1299 | this.sql.exec("UPDATE jobs SET token_hash = ?, container = ?, runner_id = ? WHERE id = ? AND container IS NULL", await sha256(token), container, runnerId, job.id); |
| 1300 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId: job.id }); |
| 1301 | return { jobId: job.id, token, runId: job.run_id, name: job.name }; |
| 1302 | } |
| 1303 | |
| 1304 | // ------------------------------------------------------------ registered self-hosted runners |
| 1305 | |
| 1306 | /** |
| 1307 | * GitHub's runner registration token: a repository admin's credential, good for an hour, that a |
| 1308 | * machine trades for its own runner token. So the machine never holds a person's token, and the |
| 1309 | * jobs it runs (as the same OS user) can't read one. |
| 1310 | */ |
| 1311 | async runnerRegistrationToken(caller: Caller): Promise<{ token: string; expires_at: string }> { |
| 1312 | if (caller.kind !== "user" || caller.permissions || !atLeast(caller.role, "admin")) throw new HttpError(403, "only repository admins can add self-hosted runners"); |
| 1313 | const token = `drg_${randomToken(32)}`; |
| 1314 | const expires = new Date(Date.now() + REGISTRATION_TTL_MS).toISOString(); |
| 1315 | const hash = await sha256(token); |
| 1316 | this.sql.exec("DELETE FROM runner_registrations WHERE expires_at < ?", now()); |
| 1317 | this.sql.exec("INSERT INTO runner_registrations (token_hash, created_by, expires_at) VALUES (?, ?, ?)", hash, caller.login, expires); |
| 1318 | return { token, expires_at: expires }; |
| 1319 | } |
| 1320 | |
| 1321 | /** Register a runner with a registration token; returns its own token, shown once. */ |
| 1322 | async registerRunner(registration: string, name: string, labels: string[], ephemeral: boolean): Promise<{ id: number; name: string; labels: string[]; ephemeral: boolean; token: string; registeredBy: string }> { |
| 1323 | const hash = await sha256(registration); |
| 1324 | const reg = this.sql.exec<{ created_by: string }>("SELECT created_by FROM runner_registrations WHERE token_hash = ? AND expires_at > ?", hash, now()).toArray()[0]; |
| 1325 | if (!reg) throw new HttpError(401, "invalid or expired registration token"); |
| 1326 | const clean = [...new Set(labels.map((l) => String(l).trim().toLowerCase()).filter((l) => l && l !== "self-hosted"))]; |
| 1327 | if (clean.length > 20) throw new HttpError(422, "a runner can offer at most 20 labels"); |
| 1328 | const bad = clean.find((l) => !/^[a-z0-9][a-z0-9._-]{0,63}$/.test(l)); |
| 1329 | if (bad) throw new HttpError(422, `invalid label ${JSON.stringify(bad.slice(0, 70))}: use letters, digits, '.', '_' and '-' (up to 64)`); |
| 1330 | const runnerName = name.trim().slice(0, 64) || "runner"; |
| 1331 | const count = this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM runners").one().n; |
| 1332 | if (count >= MAX_RUNNERS) throw new HttpError(429, `this repository has ${count} self-hosted runners, the most allowed; remove one first`); |
| 1333 | const id = this.sql.exec<{ id: number }>("INSERT INTO runners (name, labels, ephemeral, created_by, created_at) VALUES (?, ?, ?, ?, ?) RETURNING id", runnerName, JSON.stringify(clean), ephemeral ? 1 : 0, reg.created_by, now()).one().id; |
| 1334 | const token = `drn_${this.repo.meta.repoId}_${id}_${randomToken(32)}`; |
| 1335 | this.sql.exec("UPDATE runners SET token_hash = ? WHERE id = ?", await sha256(token), id); |
| 1336 | return { id, name: runnerName, labels: clean, ephemeral, token, registeredBy: reg.created_by }; |
| 1337 | } |
| 1338 | |
| 1339 | /** A registered runner asks for work with its own token. An ephemeral runner's token ends with the job it takes. */ |
| 1340 | async claimJobAsRunner(token: string): Promise<{ jobId: string; token: string; runId: string; name: string } | null> { |
| 1341 | const id = Number(token.split("_")[2]); |
| 1342 | const r = Number.isSafeInteger(id) ? this.sql.exec<RunnerRow>("SELECT * FROM runners WHERE id = ?", id).toArray()[0] : undefined; |
| 1343 | if (!r || !r.token_hash || r.token_hash !== (await sha256(token))) throw new HttpError(401, "this runner was removed, or its token is invalid; register it again"); |
| 1344 | this.sql.exec("UPDATE runners SET last_seen_at = ? WHERE id = ?", now(), r.id); |
| 1345 | const job = await this.assignJob(r.name, JSON.parse(r.labels) as string[], r.id); |
| 1346 | if (job && r.ephemeral) this.sql.exec("UPDATE runners SET token_hash = NULL WHERE id = ?", r.id); |
| 1347 | return job; |
| 1348 | } |
| 1349 | |
| 1350 | runners(caller: Caller): RunnerInfo[] { |
| 1351 | if (!atLeast(caller.role, "admin") || caller.permissions) throw new HttpError(403, "only repository admins can list self-hosted runners"); |
| 1352 | const busy = new Set(this.sql.exec<{ runner_id: number }>("SELECT DISTINCT runner_id FROM jobs WHERE runner_id IS NOT NULL AND status != 'completed'").toArray().map((x) => x.runner_id)); |
| 1353 | return this.sql.exec<RunnerRow>("SELECT * FROM runners ORDER BY id").toArray().map((r) => ({ |
| 1354 | id: r.id, |
| 1355 | name: r.name, |
| 1356 | os: "unknown", |
| 1357 | status: r.last_seen_at && Date.now() - Date.parse(r.last_seen_at) < RUNNER_ONLINE_MS ? "online" : "offline", |
| 1358 | busy: busy.has(r.id), |
| 1359 | ephemeral: !!r.ephemeral, |
| 1360 | labels: ["self-hosted", ...(JSON.parse(r.labels) as string[])].map((name, i) => ({ id: i + 1, name, type: i === 0 ? "read-only" : "custom" })), |
| 1361 | })); |
| 1362 | } |
| 1363 | |
| 1364 | /** Remove a runner: its token stops working at once. A job it is running carries on with its own job token. */ |
| 1365 | deleteRunner(id: number, caller: Caller): void { |
| 1366 | if (caller.kind !== "user" || caller.permissions || !atLeast(caller.role, "admin")) throw new HttpError(403, "only repository admins can remove self-hosted runners"); |
| 1367 | if (!this.sql.exec("DELETE FROM runners WHERE id = ? RETURNING id", id).toArray().length) throw new HttpError(404, "runner not found"); |
| 1368 | } |
| 1369 | |
| 1370 | private async stopContainer(j: JobRow): Promise<void> { |
| 1371 | if (!j.container || j.container.startsWith("self-hosted:")) return; |
| 1372 | try { |
| 1373 | await this.env.RUNNER.get(this.env.RUNNER.idFromName(j.container)).stopJob(); |
| 1374 | } catch (e) { |
| 1375 | console.warn("stop container failed", e); |
| 1376 | } |
| 1377 | } |
| 1378 | |
| 1379 | async finishRun(runId: string): Promise<RunInfo> { |
| 1380 | const run = this.runRow(runId); |
| 1381 | const jobs = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ?", runId).toArray(); |
| 1382 | let conclusion: RunRow["conclusion"] = run.conclusion === "cancelled" ? "cancelled" : "success"; |
| 1383 | if (conclusion !== "cancelled") { |
| 1384 | // A job with continue-on-error may fail without failing the run. |
| 1385 | if (jobs.some((j) => (j.conclusion === "failure" || j.conclusion === "timed_out") && !j.continue_on_error)) conclusion = "failure"; |
| 1386 | else if (jobs.some((j) => j.conclusion === "cancelled")) conclusion = "cancelled"; |
| 1387 | // As on GitHub, a run whose jobs were all skipped is skipped, not a success, so a |
| 1388 | // workflow_run follower that checks for success (a live test after a deploy) doesn't run. |
| 1389 | else if (jobs.length && jobs.every((j) => j.conclusion === "skipped")) conclusion = "skipped"; |
| 1390 | } |
| 1391 | const t = now(); |
| 1392 | this.sql.exec("UPDATE runs SET status = 'completed', conclusion = ?, updated_at = ? WHERE id = ?", conclusion, t, runId); |
| 1393 | const m = this.repo.meta; |
| 1394 | 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 }); |
| 1395 | this.repo.broadcast(`run:${runId}`, { action: "completed", conclusion }); |
| 1396 | this.repo.broadcast("runs", { action: "completed", runId }); |
| 1397 | await this.onWorkflowRun(this.runRow(runId), "completed"); |
| 1398 | return toRun(this.runRow(runId)); |
| 1399 | } |
| 1400 | |
| 1401 | /** |
| 1402 | * Starts the default branch's `on: workflow_run` workflows that follow `run`. Like |
| 1403 | * GitHub, they run on the default branch, must name the workflows they follow, and |
| 1404 | * chains stop after three levels (A -> B -> C -> D; D's runs trigger nothing). |
| 1405 | */ |
| 1406 | private async onWorkflowRun(run: RunRow, action: "requested" | "completed"): Promise<void> { |
| 1407 | if (this.chainDepth(run) >= 3) return; |
| 1408 | try { |
| 1409 | const sha = this.repo.getRef(this.repo.headRef); |
| 1410 | if (!sha) return; |
| 1411 | const followers = (await this.loadWorkflows(sha)).filter((w) => w.workflow?.on.workflow_run?.workflows?.length); |
| 1412 | if (!followers.length) return; |
| 1413 | const m = this.repo.meta; |
| 1414 | const payload = JSON.parse(run.event_payload) as { pull_request?: { head?: { ref?: string } } }; |
| 1415 | const branch = payload.pull_request?.head?.ref ?? run.ref.replace(/^refs\/(heads|tags)\//, ""); |
| 1416 | const actor = JSON.parse(run.actor) as Actor; |
| 1417 | const workflowRun = { |
| 1418 | id: run.run_number, |
| 1419 | name: run.workflow_name, |
| 1420 | display_title: run.display_title, |
| 1421 | path: run.workflow_path, |
| 1422 | run_number: run.run_number, |
| 1423 | run_attempt: run.attempt, |
| 1424 | event: run.event, |
| 1425 | status: run.status, |
| 1426 | conclusion: run.conclusion, |
| 1427 | head_branch: branch, |
| 1428 | head_sha: run.head_sha ?? run.sha, |
| 1429 | actor: { login: actor.login }, |
| 1430 | triggering_actor: { login: actor.login }, |
| 1431 | pull_requests: run.pr_number ? [{ number: run.pr_number }] : [], |
| 1432 | created_at: run.created_at, |
| 1433 | updated_at: run.updated_at, |
| 1434 | html_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}`, |
| 1435 | }; |
| 1436 | for (const w of followers) { |
| 1437 | if (!matchesTrigger(w.workflow!, { name: "workflow_run", action, workflow: run.workflow_name, branch })) continue; |
| 1438 | await this.createRun(w.workflow!, { |
| 1439 | event: "workflow_run", |
| 1440 | ref: this.repo.headRef, |
| 1441 | sha, |
| 1442 | payload: { action, workflow_run: workflowRun, workflow: { name: run.workflow_name, path: run.workflow_path }, repository: await this.repositoryPayload(), sender: { login: actor.login } }, |
| 1443 | caller: actor, |
| 1444 | title: w.workflow!.name ?? w.path, |
| 1445 | // A follower acts on what the run it follows produced (artifacts, outputs), so it stays an agent's. |
| 1446 | agentOrigin: !!run.agent_origin, |
| 1447 | }); |
| 1448 | } |
| 1449 | } catch (e) { |
| 1450 | console.warn("workflow_run trigger failed", e); |
| 1451 | } |
| 1452 | } |
| 1453 | |
| 1454 | /** How many workflow_run links led to `run` (0 for a run started by any other event). */ |
| 1455 | private chainDepth(run: RunRow): number { |
| 1456 | let depth = 0; |
| 1457 | for (let r: RunRow | undefined = run; r?.event === "workflow_run" && depth < 3; depth++) { |
| 1458 | const parent: string | number | undefined = (JSON.parse(r.event_payload) as { workflow_run?: { id?: string | number } }).workflow_run?.id; |
| 1459 | 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; |
| 1460 | } |
| 1461 | return depth; |
| 1462 | } |
| 1463 | |
| 1464 | async plannedError(runId: string, message: string): Promise<void> { |
| 1465 | const t = now(); |
| 1466 | 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 })); |
| 1467 | } |
| 1468 | |
| 1469 | // ---------------------------------------------------------------- cancellation and re-runs |
| 1470 | |
| 1471 | private async cancelInternal(runId: string, reason: string): Promise<void> { |
| 1472 | const run = this.runRow(runId); |
| 1473 | if (run.status === "completed") return; |
| 1474 | this.sql.exec("UPDATE runs SET conclusion = 'cancelled', updated_at = ? WHERE id = ?", now(), run.id); |
| 1475 | for (const j of this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND status != 'completed'", run.id).toArray()) { |
| 1476 | await this.stopContainer(j); |
| 1477 | await this.completeJob(j.id, { conclusion: "cancelled", outputs: {}, failure: { message: reason } }); |
| 1478 | } |
| 1479 | } |
| 1480 | |
| 1481 | async cancel(runId: string, caller: Caller): Promise<void> { |
| 1482 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "permission denied"); |
| 1483 | await this.cancelInternal(runId, `Cancelled by ${caller.login}`); |
| 1484 | } |
| 1485 | |
| 1486 | async rerun(runId: string, caller: Caller): Promise<RunInfo> { |
| 1487 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "permission denied"); |
| 1488 | if (caller.id.startsWith("job:")) throw new HttpError(403, "a workflow's GITHUB_TOKEN cannot re-run workflows"); |
| 1489 | const old = this.runRow(runId); |
| 1490 | const wf = (await this.loadWorkflows(old.sha)).find((w) => w.path === old.workflow_path)?.workflow; |
| 1491 | if (!wf) throw new HttpError(422, "workflow file no longer valid at this commit"); |
| 1492 | const run = await this.createRun(wf, { |
| 1493 | event: old.event, |
| 1494 | ref: old.ref, |
| 1495 | sha: old.sha, |
| 1496 | headSha: old.head_sha ?? undefined, |
| 1497 | payload: JSON.parse(old.event_payload), |
| 1498 | caller, |
| 1499 | prNumber: old.pr_number ?? undefined, |
| 1500 | inputs: JSON.parse(old.inputs), |
| 1501 | title: old.display_title, |
| 1502 | // Re-running code an agent wrote keeps it acting as the agent, whoever re-runs it. |
| 1503 | agentOrigin: !!old.agent_origin, |
| 1504 | }); |
| 1505 | // Whoever started an earlier attempt still can't approve this one where an environment prevents self-review. |
| 1506 | const prior = [...new Set([...((old.prior_actors ? JSON.parse(old.prior_actors) : []) as string[]), (JSON.parse(old.actor) as Actor).id])]; |
| 1507 | this.sql.exec("UPDATE runs SET attempt = ?, run_number = ?, prior_actors = ? WHERE id = ?", old.attempt + 1, old.run_number, JSON.stringify(prior), run!.id); |
| 1508 | return toRun(this.runRow(run!.id)); |
| 1509 | } |
| 1510 | |
| 1511 | // ---------------------------------------------------------------- runner protocol |
| 1512 | |
| 1513 | /** Validate a job or sandbox token. Returns what it grants. */ |
| 1514 | async runnerAuth( |
| 1515 | token: string, |
| 1516 | ): Promise<{ kind: "job"; jobId: string; runId: string; role: Caller["role"]; permissions: JobPermissions; actorKind: Actor["kind"] } | { kind: "sandbox"; sandboxId: string; caller: Caller; taskPerson?: string } | null> { |
| 1517 | const parts = token.split("_"); |
| 1518 | if (parts.length < 4) return null; |
| 1519 | const hash = await sha256(token); |
| 1520 | if (parts[0] === "djt") { |
| 1521 | const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", parts[2]!).toArray()[0]; |
| 1522 | if (!j || j.token_hash !== hash || j.status === "completed") return null; |
| 1523 | const grant = this.repo.runtime.grantFor(j.id); |
| 1524 | // Code an agent wrote, pushed or opened a pull request with runs as that agent (repo/token-policy.ts). |
| 1525 | const run = this.sql.exec<{ agent_origin: number }>("SELECT agent_origin FROM runs WHERE id = ?", j.run_id).toArray()[0]; |
| 1526 | const actorKind: Actor["kind"] = run?.agent_origin ? "agent" : "system"; |
| 1527 | return { kind: "job", jobId: j.id, runId: j.run_id, role: grant.role, permissions: grant.permissions, actorKind }; |
| 1528 | } |
| 1529 | if (parts[0] === "dst") { |
| 1530 | const s = this.sql.exec<{ id: string; secret: string; created_by: string; status: string; created_at: string; task_person: string | null }>("SELECT * FROM sandboxes WHERE id = ?", parts[2]!).toArray()[0]; |
| 1531 | if (!s || s.secret !== hash || s.status !== "running") return null; |
| 1532 | // A sandbox's token lasts a day at most, even if nobody closes the sandbox. |
| 1533 | if (Date.now() - Date.parse(s.created_at) > SANDBOX_TOKEN_TTL_MS) return null; |
| 1534 | return { kind: "sandbox", sandboxId: s.id, caller: JSON.parse(s.created_by) as Caller, ...(s.task_person ? { taskPerson: s.task_person } : {}) }; |
| 1535 | } |
| 1536 | return null; |
| 1537 | } |
| 1538 | |
| 1539 | private async authJob(jobId: string, token: string): Promise<JobRow> { |
| 1540 | const auth = await this.runnerAuth(token); |
| 1541 | if (!auth || auth.kind !== "job" || auth.jobId !== jobId) throw new HttpError(401, "invalid job token"); |
| 1542 | return this.jobRow(jobId); |
| 1543 | } |
| 1544 | |
| 1545 | /** Claims for an OIDC id token; side-effect free (unlike runnerSpec). */ |
| 1546 | async oidcClaims(jobId: string, token: string): Promise<{ allowed: boolean; github: Record<string, string> }> { |
| 1547 | const job = await this.authJob(jobId, token); |
| 1548 | if (job.status !== "in_progress") throw new HttpError(409, "job is not running"); |
| 1549 | const run = this.runRow(job.run_id); |
| 1550 | const exp = await this.expanded(run); |
| 1551 | const def: Job = exp.workflow.jobs[job.job_key]!; |
| 1552 | const wf = exp.workflowOf.get(job.job_key)!; |
| 1553 | const allowed = this.repo.runtime.grantFor(jobId).permissions["id-token"] === "write"; |
| 1554 | const ctx = this.githubContext(run, localKey(job.job_key, exp.parentOf.get(job.job_key) ?? null), token) as Record<string, Json>; |
| 1555 | const github: Record<string, string> = {}; |
| 1556 | for (const [k, v] of Object.entries(ctx)) if (typeof v === "string" || typeof v === "number") github[k] = String(v); |
| 1557 | // Ids don't change when a name is reused, so a cloud trust policy can pin them; the environment is the one |
| 1558 | // whose protection rules this job passed (runnerSpec records its deployment). |
| 1559 | github.repository_owner_id = String(numericId(this.repo.meta.ownerId)); |
| 1560 | const actor = JSON.parse(run.actor) as Actor; |
| 1561 | if (actor.id) github.actor_id = String(numericId(actor.id)); |
| 1562 | const env = this.sql.exec<{ environment: string }>("SELECT environment FROM deployments WHERE job_id = ?", jobId).toArray()[0]; |
| 1563 | if (env) github.environment = env.environment; |
| 1564 | return { allowed, github }; |
| 1565 | } |
| 1566 | |
| 1567 | /** |
| 1568 | * What a job token may do with container packages (the `packages` permission). |
| 1569 | * A workflow without a permissions block gets the repository's default: read in |
| 1570 | * repositories with read-only tokens (the default for new ones), write in those that |
| 1571 | * keep GitHub's old permissive default. A permissions map that leaves `packages` out gets none. |
| 1572 | */ |
| 1573 | async packagesPermission(token: string): Promise<"none" | "read" | "write"> { |
| 1574 | const auth = await this.runnerAuth(token); |
| 1575 | return auth?.kind === "job" ? auth.permissions.packages : "none"; |
| 1576 | } |
| 1577 | |
| 1578 | /** |
| 1579 | * A pull_request run for a pull request from another repository (a fork): its jobs |
| 1580 | * never get write or secrets. Decided from the pull request's row, not the event payload. |
| 1581 | */ |
| 1582 | private untrusted(run: RunRow): boolean { |
| 1583 | return run.event === "pull_request" && this.repo.policy.isForkPull(run.pr_number); |
| 1584 | } |
| 1585 | |
| 1586 | async runnerSpec(jobId: string, token: string): Promise<JobSpec> { |
| 1587 | const job = await this.authJob(jobId, token); |
| 1588 | const run = this.runRow(job.run_id); |
| 1589 | const exp = await this.expanded(run); |
| 1590 | const def: Job = exp.workflow.jobs[job.job_key]!; |
| 1591 | const wf = exp.workflowOf.get(job.job_key)!; |
| 1592 | const parent = exp.parentOf.get(job.job_key) ?? null; |
| 1593 | const short = localKey(job.job_key, parent); |
| 1594 | let scope: { inputs: Record<string, unknown>; secrets: Record<string, string> }; |
| 1595 | try { |
| 1596 | // A fork's code never sees the repository's secrets, as on GitHub, and an agent's only where an admin allows it. |
| 1597 | scope = await this.callContext(run, exp, parent, !this.untrusted(run) && (!run.agent_origin || this.repo.policy.settings().agentRunSecrets)); |
| 1598 | } catch (e) { |
| 1599 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: (e as Error).message } }); |
| 1600 | throw new HttpError(409, (e as Error).message); |
| 1601 | } |
| 1602 | const m = this.repo.meta; |
| 1603 | // Deployment environment (`environment:`): its branch policy, secrets and variables, and a deployment record. |
| 1604 | let environment: { secrets: Record<string, string>; vars: Record<string, string> } = { secrets: {}, vars: {} }; |
| 1605 | if (def.environment) { |
| 1606 | try { |
| 1607 | 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) }; |
| 1608 | // The name evaluated at launch, whose protection rules the job passed (older jobs evaluate it here). |
| 1609 | const name = job.environment ?? interpolate(def.environment.name, ctx); |
| 1610 | if (this.repo.deploys.protection(name) && job.gate_state !== "released") throw new Error(`The environment ${name} requires approval this job didn't get. Re-run the job to ask for it.`); |
| 1611 | const url = def.environment.url && !def.environment.url.includes("steps.") ? interpolate(def.environment.url, ctx) : undefined; |
| 1612 | const logUrl = `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}`; |
| 1613 | // Code an agent wrote deploys as that agent (environment rules hold it to the agent's limits). |
| 1614 | const actor = JSON.parse(run.actor) as Actor; |
| 1615 | 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 }); |
| 1616 | } catch (e) { |
| 1617 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: (e as Error).message } }); |
| 1618 | throw new HttpError(409, (e as Error).message); |
| 1619 | } |
| 1620 | } |
| 1621 | const t = now(); |
| 1622 | this.sql.exec("UPDATE jobs SET status = 'in_progress', started_at = ? WHERE id = ?", t, jobId); |
| 1623 | 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}` }); |
| 1624 | this.repo.broadcast(`run:${run.id}`, { action: "job", jobId }); |
| 1625 | // Only the secrets the job's expressions name reach the runner, as on GitHub (plus the runner's own debug switches). |
| 1626 | const used = referencedSecrets(def, wf.env); |
| 1627 | if (used !== "all") for (const name of RUNNER_SECRETS) used.add(name); |
| 1628 | const secrets: Record<string, string> = { ...secretsFor({ ...scope.secrets, ...(this.untrusted(run) ? {} : environment.secrets) }, used), GITHUB_TOKEN: token, DIFFINERY_TOKEN: token }; |
| 1629 | // The job token can do what the effective `permissions:` allow (job level wins, like GitHub). |
| 1630 | const permissions = permissionsForJob(exp, job.job_key, { untrusted: this.untrusted(run), repoDefault: this.repo.policy.settings().defaultPermissions }); |
| 1631 | this.repo.runtime.grant(jobId, permissions); |
| 1632 | const idToken = permissions["id-token"] === "write"; |
| 1633 | return { |
| 1634 | runId: run.id, |
| 1635 | jobId, |
| 1636 | jobKey: short, |
| 1637 | name: job.name, |
| 1638 | repo: { owner: m.owner, name: m.name, cloneUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, defaultBranch: m.defaultBranch }, |
| 1639 | sha: run.sha, |
| 1640 | ref: run.ref, |
| 1641 | workflowPath: run.workflow_path, |
| 1642 | github: this.githubContext(run, short, token), |
| 1643 | env: { ...wf.env, ...def.env }, |
| 1644 | defaults: { run: { ...wf.defaults.run, ...def.defaults.run } }, |
| 1645 | matrix: JSON.parse(job.matrix), |
| 1646 | strategy: JSON.parse(job.strategy), |
| 1647 | needs: await this.needsResults(run, exp, exp.needs.get(job.job_key) ?? [], parent), |
| 1648 | inputs: scope.inputs, |
| 1649 | vars: { ...this.variables(), ...environment.vars }, |
| 1650 | secrets, |
| 1651 | outputs: def.outputs, |
| 1652 | steps: def.steps.map((s) => ({ ...s, "timeout-minutes": s["timeout-minutes"] === undefined ? undefined : Number(s["timeout-minutes"]) })), |
| 1653 | container: def.container, |
| 1654 | services: def.services, |
| 1655 | timeoutMinutes: job.timeout_minutes, |
| 1656 | continueOnError: !!job.continue_on_error, |
| 1657 | idToken: !!idToken, |
| 1658 | apiUrl: this.env.PUBLIC_URL, |
| 1659 | token, |
| 1660 | runtime: await this.runtimeServices(run, job), |
| 1661 | }; |
| 1662 | } |
| 1663 | |
| 1664 | /** ACTIONS_RUNTIME_TOKEN and the service URLs; the token expires when the job would time out. */ |
| 1665 | private async runtimeServices(run: RunRow, job: JobRow): Promise<JobSpec["runtime"]> { |
| 1666 | if (!this.env.MASTER_KEY) return undefined; |
| 1667 | const origin = new URL(this.env.PUBLIC_URL).origin; |
| 1668 | const exp = Math.floor(Date.now() / 1000) + (job.timeout_minutes + 30) * 60; |
| 1669 | const { scopes } = this.repo.runtime.job(job.id); |
| 1670 | const token = await signRuntimeToken(this.env.MASTER_KEY, origin, { repo: this.repo.meta.repoId, run: run.id, job: job.id, exp }, scopes); |
| 1671 | return { token, resultsUrl: `${origin}/`, cacheUrl: `${origin}/_runtime/`, runtimeUrl: `${origin}/_runtime/` }; |
| 1672 | } |
| 1673 | |
| 1674 | async runnerLogs(jobId: string, token: string, lines: LogLine[]): Promise<void> { |
| 1675 | const job = await this.authJob(jobId, token); |
| 1676 | let chars = job.log_chars ?? 0; |
| 1677 | if (chars > MAX_JOB_LOG) return; |
| 1678 | let seq = this.sql.exec<{ n: number | null }>("SELECT max(seq) AS n FROM logs WHERE job_id = ?", jobId).one().n ?? -1; |
| 1679 | const out: Array<LogLine & { seq: number }> = []; |
| 1680 | for (const l of lines.slice(0, 5000)) { |
| 1681 | seq++; |
| 1682 | const t = String(l.t ?? "").slice(0, 40); |
| 1683 | const step = Number.isInteger(l.step) ? l.step : 0; |
| 1684 | let text = String(l.text ?? ""); |
| 1685 | if (text.length > 16384) text = `${text.slice(0, 16384)}…`; |
| 1686 | // Past the cap, one last line says so and the rest is dropped: the whole log is read into |
| 1687 | // memory when the job completes, to archive it. |
| 1688 | if (chars + text.length > MAX_JOB_LOG) { |
| 1689 | text = `The log is cut off here: a job's log is limited to ${MAX_JOB_LOG / 1048576} MiB of text.`; |
| 1690 | chars = MAX_JOB_LOG + 1; |
| 1691 | } else chars += text.length; |
| 1692 | this.sql.exec("INSERT INTO logs (job_id, seq, t, step, text) VALUES (?, ?, ?, ?, ?)", jobId, seq, t, step, text); |
| 1693 | out.push({ t, step, text, seq }); |
| 1694 | if (chars > MAX_JOB_LOG) break; |
| 1695 | } |
| 1696 | this.sql.exec("UPDATE jobs SET log_chars = ? WHERE id = ?", chars, jobId); |
| 1697 | this.repo.broadcast(`job:${jobId}`, { action: "logs", lines: out }); |
| 1698 | } |
| 1699 | |
| 1700 | async runnerStep(jobId: string, token: string, index: number, report: StepReport): Promise<void> { |
| 1701 | const job = await this.authJob(jobId, token); |
| 1702 | const steps = JSON.parse(job.steps) as Array<StepReport & { index: number }>; |
| 1703 | const existing = steps.findIndex((s) => s.index === index); |
| 1704 | const next = { ...(existing >= 0 ? steps[existing] : {}), ...report, index }; |
| 1705 | if (existing >= 0) steps[existing] = next; |
| 1706 | else steps.push(next); |
| 1707 | steps.sort((a, b) => a.index - b.index); |
| 1708 | this.sql.exec("UPDATE jobs SET steps = ? WHERE id = ?", JSON.stringify(steps), jobId); |
| 1709 | this.repo.broadcast(`job:${jobId}`, { action: "step", step: next }); |
| 1710 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId }); |
| 1711 | } |
| 1712 | |
| 1713 | async runnerComplete(jobId: string, token: string, completion: JobCompletion): Promise<void> { |
| 1714 | await this.authJob(jobId, token); |
| 1715 | await this.completeJob(jobId, completion); |
| 1716 | } |
| 1717 | |
| 1718 | private async completeJob(jobId: string, completion: JobCompletion, override?: "timed_out"): Promise<void> { |
| 1719 | const job = this.jobRow(jobId); |
| 1720 | if (job.status === "completed") return; |
| 1721 | const gated = job.status === "approval"; |
| 1722 | const run = this.runRow(job.run_id); |
| 1723 | const t = now(); |
| 1724 | const conclusion = override ?? completion.conclusion; |
| 1725 | this.sql.exec( |
| 1726 | "UPDATE jobs SET status = 'completed', conclusion = ?, completed_at = ?, outputs = ?, failure = ?, summary = ?, token_hash = NULL, started_at = COALESCE(started_at, ?) WHERE id = ?", |
| 1727 | conclusion, |
| 1728 | t, |
| 1729 | JSON.stringify(completion.outputs ?? {}), |
| 1730 | completion.failure ? JSON.stringify(completion.failure) : null, |
| 1731 | completion.summary ?? null, |
| 1732 | t, |
| 1733 | jobId, |
| 1734 | ); |
| 1735 | this.repo.deploys.jobCompleted(jobId, conclusion); |
| 1736 | const m = this.repo.meta; |
| 1737 | this.repo.upsertCheck({ |
| 1738 | sha: run.head_sha ?? run.sha, |
| 1739 | name: this.checkName(run, job), |
| 1740 | status: "completed", |
| 1741 | conclusion: conclusion === "timed_out" ? "timed_out" : conclusion, |
| 1742 | runId: run.id, |
| 1743 | jobId, |
| 1744 | detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}`, |
| 1745 | output: completion.failure ? { title: "Failed", summary: completion.failure.message } : { title: conclusion }, |
| 1746 | }); |
| 1747 | // Archive logs to R2 and free the Durable Object's storage. |
| 1748 | const lines = this.sql.exec<LogLine & { seq: number }>("SELECT seq, t, step, text FROM logs WHERE job_id = ? ORDER BY seq", jobId).toArray(); |
| 1749 | if (lines.length) { |
| 1750 | await this.env.BUCKET.put(this.logKey(jobId), JSON.stringify(lines), { httpMetadata: { contentType: "application/json" } }); |
| 1751 | this.sql.exec("DELETE FROM logs WHERE job_id = ?", jobId); |
| 1752 | } |
| 1753 | // Uploads the job left unfinished stop counting against the budgets, and their bytes go. |
| 1754 | await this.repo.runtime.sweepJob(jobId).catch((e) => console.error("runtime sweep failed", e)); |
| 1755 | this.repo.broadcast(`job:${jobId}`, { action: "completed", conclusion }); |
| 1756 | this.repo.broadcast(`run:${run.id}`, { action: "job", jobId }); |
| 1757 | |
| 1758 | // fail-fast: cancel unfinished siblings of a failed matrix job (unless the failed leg may continue on error). |
| 1759 | 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(); |
| 1760 | const strategy = JSON.parse(job.strategy || "{}") as { "fail-fast"?: boolean }; |
| 1761 | if ((conclusion === "failure" || conclusion === "timed_out") && !job.continue_on_error && strategy["fail-fast"] !== false && siblings.length) { |
| 1762 | for (const s of siblings) { |
| 1763 | await this.stopContainer(s); |
| 1764 | await this.completeJob(s.id, { conclusion: "cancelled", outputs: {}, failure: { message: `Cancelled because ${job.name} failed (fail-fast).` } }); |
| 1765 | } |
| 1766 | } else { |
| 1767 | const queued = siblings.find((s) => s.status === "queued"); |
| 1768 | if (queued) await this.launch(queued.id); |
| 1769 | } |
| 1770 | // An ephemeral runner is done once its one job is. |
| 1771 | if (job.runner_id) this.sql.exec("DELETE FROM runners WHERE id = ? AND ephemeral = 1 AND token_hash IS NULL", job.runner_id); |
| 1772 | // A hosted job hands its slot to the oldest job waiting for one. |
| 1773 | if (!job.labels && job.container) { |
| 1774 | const held = this.sql.exec<{ id: string }>("SELECT id FROM jobs WHERE fleet_held = 1 AND status = 'queued' ORDER BY rowid LIMIT 1").toArray()[0]; |
| 1775 | if (held) await this.launch(held.id); |
| 1776 | } |
| 1777 | // A job that held its concurrency group hands it to the oldest pending job (of any run). |
| 1778 | if (job.concurrency_group && (job.status === "waiting" || job.status === "in_progress")) { |
| 1779 | 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]; |
| 1780 | if (next) await this.launch(next.id); |
| 1781 | } |
| 1782 | try { |
| 1783 | const instance = await this.env.RUN_WORKFLOW.get(run.id); |
| 1784 | await instance.sendEvent({ type: `job-${jobId}`, payload: { conclusion } }); |
| 1785 | } catch (e) { |
| 1786 | console.error("failed to notify workflow", e); |
| 1787 | } |
| 1788 | // The Workflow waits for a job held for approval on its own event. |
| 1789 | if (gated) { |
| 1790 | this.syncWaiting(run.id); |
| 1791 | await this.wakeGate(run.id, jobId); |
| 1792 | } |
| 1793 | } |
| 1794 | |
| 1795 | // ---------------------------------------------------------------- secrets and variables |
| 1796 | |
| 1797 | private async repoSecrets(): Promise<Record<string, string>> { |
| 1798 | const out: Record<string, string> = {}; |
| 1799 | for (const r of this.sql.exec<{ name: string; ciphertext: string }>("SELECT name, ciphertext FROM secrets").toArray()) { |
| 1800 | out[r.name] = await decryptSecret(this.env.MASTER_KEY, `repo:${this.repo.meta.repoId}`, r.ciphertext); |
| 1801 | } |
| 1802 | return out; |
| 1803 | } |
| 1804 | |
| 1805 | private async orgSecrets(): Promise<Record<string, string>> { |
| 1806 | const m = this.repo.meta; |
| 1807 | if (m.ownerKind !== "org") return {}; |
| 1808 | // Only the secrets whose repository selection includes this repository (visibility all, private or selected). |
| 1809 | const rows = await this.env.DB.prepare( |
| 1810 | `SELECT s.name, s.ciphertext FROM org_secrets s WHERE s.org_id = ?1 AND (s.visibility = 'all' |
| 1811 | OR (s.visibility = 'private' AND EXISTS (SELECT 1 FROM repos r WHERE r.id = ?2 AND r.visibility != 'public')) |
| 1812 | OR (s.visibility = 'selected' AND EXISTS (SELECT 1 FROM org_secret_repos x WHERE x.org_id = s.org_id AND x.name = s.name AND x.repo_id = ?2)))`, |
| 1813 | ) |
| 1814 | .bind(m.ownerId, m.repoId) |
| 1815 | .all<{ name: string; ciphertext: string }>(); |
| 1816 | const out: Record<string, string> = {}; |
| 1817 | for (const r of rows.results) out[r.name] = await decryptSecret(this.env.MASTER_KEY, `org:${m.ownerId}`, r.ciphertext); |
| 1818 | return out; |
| 1819 | } |
| 1820 | |
| 1821 | async setSecret(name: string, value: string): Promise<void> { |
| 1822 | if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(name) || /^GITHUB_/i.test(name)) throw new HttpError(422, "invalid secret name"); |
| 1823 | 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()); |
| 1824 | } |
| 1825 | |
| 1826 | deleteSecret(name: string): void { |
| 1827 | this.sql.exec("DELETE FROM secrets WHERE name = ?", name.toUpperCase()); |
| 1828 | } |
| 1829 | |
| 1830 | secretNames(): Array<{ name: string; updatedAt: string }> { |
| 1831 | 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 })); |
| 1832 | } |
| 1833 | |
| 1834 | setVariable(name: string, value: string | null): void { |
| 1835 | if (value === null) this.sql.exec("DELETE FROM variables WHERE name = ?", name.toUpperCase()); |
| 1836 | else this.sql.exec("INSERT OR REPLACE INTO variables (name, value, updated_at) VALUES (?, ?, ?)", name.toUpperCase(), value, now()); |
| 1837 | } |
| 1838 | |
| 1839 | /** Variables for `vars.*`: the organization's, overridden by the repository's. `withOrg = false` lists only the repository's own. */ |
| 1840 | variables(withOrg = true): Record<string, string> { |
| 1841 | const own = Object.fromEntries(this.sql.exec<{ name: string; value: string }>("SELECT name, value FROM variables").toArray().map((r) => [r.name, r.value])); |
| 1842 | if (!withOrg) return own; |
| 1843 | const org = Object.fromEntries(this.sql.exec<{ name: string; value: string }>("SELECT name, value FROM org_variables").toArray().map((r) => [r.name, r.value])); |
| 1844 | return { ...org, ...own }; |
| 1845 | } |
| 1846 | |
| 1847 | // ---------------------------------------------------------------- cache and artifacts |
| 1848 | |
| 1849 | /** |
| 1850 | * An older runner's cache (the runner API before the toolkit's cache service). |
| 1851 | * Entries are stored as `<ref>\n<key>` and scoped like the cache service's |
| 1852 | * (repo/runtime.ts): a job saves to its own ref and restores from its own ref, |
| 1853 | * then the pull request's base, then the default branch, so one branch or pull |
| 1854 | * request never sees another's caches. Entries saved before scoping (no ref) |
| 1855 | * are never restored, so they go a week after their last use. |
| 1856 | */ |
| 1857 | async cacheLookup(jobId: string, token: string, key: string, restoreKeys: string[]): Promise<{ key: string; stored: string } | null> { |
| 1858 | await this.authJob(jobId, token); |
| 1859 | const { scopes } = this.repo.runtime.job(jobId); |
| 1860 | // An entry whose object is gone is forgotten, and the lookup moves on. |
| 1861 | for (let attempt = 0; attempt < 5; attempt++) { |
| 1862 | const hit = this.legacyCacheMatch(scopes, key, restoreKeys); |
| 1863 | if (!hit) return null; |
| 1864 | if (!(await headFirst(this.env.BUCKET, this.cacheR2Keys(hit)))) { |
| 1865 | this.sql.exec("DELETE FROM caches WHERE key = ?", hit); |
| 1866 | continue; |
| 1867 | } |
| 1868 | this.sql.exec("UPDATE caches SET last_used = ? WHERE key = ?", now(), hit); |
| 1869 | return { key: hit.slice(hit.indexOf("\n") + 1), stored: hit }; |
| 1870 | } |
| 1871 | return null; |
| 1872 | } |
| 1873 | |
| 1874 | /** In each scope in turn: the exact key, then the newest entry for each restore key prefix. */ |
| 1875 | private legacyCacheMatch(scopes: string[], key: string, restoreKeys: string[]): string | null { |
| 1876 | for (const scope of scopes) { |
| 1877 | const exact = this.sql.exec<{ key: string }>("SELECT key FROM caches WHERE key = ?", `${scope}\n${key}`).toArray()[0]; |
| 1878 | if (exact) return exact.key; |
| 1879 | for (const prefix of restoreKeys) { |
| 1880 | const p = `${scope}\n${prefix}`; |
| 1881 | 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]; |
| 1882 | if (r) return r.key; |
| 1883 | } |
| 1884 | } |
| 1885 | return null; |
| 1886 | } |
| 1887 | |
| 1888 | /** Where a job saves `key` (its own ref's scope): the stored key and its R2 keys. */ |
| 1889 | async cacheSaveKeys(jobId: string, token: string, key: string): Promise<{ stored: string; r2Keys: string[] }> { |
| 1890 | await this.authJob(jobId, token); |
| 1891 | const stored = `${this.repo.runtime.job(jobId).scopes[0]}\n${key}`; |
| 1892 | return { stored, r2Keys: this.cacheR2Keys(stored) }; |
| 1893 | } |
| 1894 | |
| 1895 | /** The R2 keys of entry `stored`, when the job may read its scope. */ |
| 1896 | async cacheReadKeys(jobId: string, token: string, stored: string): Promise<string[] | null> { |
| 1897 | await this.authJob(jobId, token); |
| 1898 | const nl = stored.indexOf("\n"); |
| 1899 | if (nl < 0 || !this.repo.runtime.job(jobId).scopes.includes(stored.slice(0, nl))) return null; |
| 1900 | return this.cacheR2Keys(stored); |
| 1901 | } |
| 1902 | |
| 1903 | /** Where an older runner's cache entry is stored: the current key, then the old layout's. */ |
| 1904 | cacheR2Keys(key: string): string[] { |
| 1905 | return legacyCacheKeys(this.repo.meta.repoId, key); |
| 1906 | } |
| 1907 | |
| 1908 | /** Records a saved entry; returns the R2 keys of the entries it evicted, for the caller to delete. */ |
| 1909 | async cacheSaved(jobId: string, token: string, key: string, size: number): Promise<string[]> { |
| 1910 | const { stored } = await this.cacheSaveKeys(jobId, token, key); |
| 1911 | this.sql.exec("INSERT OR REPLACE INTO caches (key, size, created_at, last_used) VALUES (?, ?, ?, ?)", stored, size, now(), now()); |
| 1912 | // Like the cache service (repo/runtime.ts): entries unused for a week go, then the |
| 1913 | // least recently used ones above 10 GB per repository. |
| 1914 | 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); |
| 1915 | for (const k of stale) this.sql.exec("DELETE FROM caches WHERE key = ?", k); |
| 1916 | const evicted = stale.flatMap((k) => this.cacheR2Keys(k)); |
| 1917 | let total = this.sql.exec<{ n: number | null }>("SELECT sum(size) AS n FROM caches").one().n ?? 0; |
| 1918 | while (total > CACHE_LIMIT_BYTES) { |
| 1919 | const old = this.sql.exec<{ key: string; size: number }>("SELECT key, size FROM caches ORDER BY last_used LIMIT 1").one(); |
| 1920 | this.sql.exec("DELETE FROM caches WHERE key = ?", old.key); |
| 1921 | evicted.push(...this.cacheR2Keys(old.key)); |
| 1922 | total -= old.size; |
| 1923 | } |
| 1924 | return evicted; |
| 1925 | } |
| 1926 | |
| 1927 | /** |
| 1928 | * Where job `jobId` (authenticated by its token) stores artifact `name` of its run, and how many bytes |
| 1929 | * it may store there (what is left of the artifact quota). Refused when the run |
| 1930 | * already has an artifact of that name from the artifact service. |
| 1931 | */ |
| 1932 | async artifactUploadKeys(jobId: string, token: string, name: string): Promise<{ r2Keys: string[]; room: number }> { |
| 1933 | const job = await this.authJob(jobId, token); |
| 1934 | this.repo.runtime.legacyArtifactCheck(job.run_id, name); |
| 1935 | return { r2Keys: this.artifactR2Keys(job.run_id, name), room: await this.repo.runtime.artifactRoom() }; |
| 1936 | } |
| 1937 | |
| 1938 | /** Records an older runner's artifact, stored at its legacy key, with the artifact service's (repo/runtime.ts). */ |
| 1939 | async artifactSaved(jobId: string, token: string, name: string, size: number): Promise<{ id: number }> { |
| 1940 | const job = await this.authJob(jobId, token); |
| 1941 | return { id: this.repo.runtime.legacyArtifactSaved(job.run_id, jobId, name, size).id }; |
| 1942 | } |
| 1943 | |
| 1944 | /** Where job `jobId` (authenticated by its token) reads artifact `name` of its run. */ |
| 1945 | async artifactKeysFor(jobId: string, token: string, name: string): Promise<string[]> { |
| 1946 | const job = await this.authJob(jobId, token); |
| 1947 | return this.artifactR2Keys(job.run_id, name); |
| 1948 | } |
| 1949 | |
| 1950 | /** Where an older runner's artifact is stored: the current key, then the old layout's. */ |
| 1951 | artifactR2Keys(runId: string, name: string): string[] { |
| 1952 | return legacyArtifactKeys(this.repo.meta.repoId, runId, name); |
| 1953 | } |
| 1954 | |
| 1955 | /** |
| 1956 | * A run's artifacts: those uploaded through the artifact service (zip files, or |
| 1957 | * single unarchived files) and older runners' tarballs. |
| 1958 | */ |
| 1959 | artifacts(runId: string): Array<{ name: string; size: number; createdAt: string; r2Keys: string[]; id: number; contentType: string; fileName: string }> { |
| 1960 | runId = this.runRow(runId).id; |
| 1961 | return this.repo.runtime |
| 1962 | .runArtifacts(runId) |
| 1963 | .map((a) => ({ |
| 1964 | name: a.name, |
| 1965 | size: a.size, |
| 1966 | createdAt: a.createdAt, |
| 1967 | r2Keys: artifactObjectKeys(this.repo.meta.repoId, a), |
| 1968 | id: a.id, |
| 1969 | contentType: a.mimeType, |
| 1970 | fileName: artifactFileName(a), |
| 1971 | })) |
| 1972 | .sort((a, b) => a.name.localeCompare(b.name)); |
| 1973 | } |
| 1974 | |
| 1975 | // ---------------------------------------------------------------- sandboxes for agents |
| 1976 | |
| 1977 | /** |
| 1978 | * `task` is set only by agentic/ (never from a request): the sandbox of an agent task, started for |
| 1979 | * the agent `caller` by the person `task.personId`. No public sandbox call reaches it. An agent |
| 1980 | * review's sandbox is `readOnly`: stored with a `read` caller, its git credential can't push. |
| 1981 | */ |
| 1982 | async createSandbox(caller: Caller, ref?: string, task?: { personId: string; readOnly?: boolean }): Promise<{ id: string; ref: string }> { |
| 1983 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "write access is required to create a sandbox"); |
| 1984 | // A review's sandbox acts with read access only: its git credential can clone but never push. |
| 1985 | if (task?.readOnly) caller = { ...caller, role: "read" }; |
| 1986 | if (!task) { |
| 1987 | const since = new Date(Date.now() - SANDBOX_TOKEN_TTL_MS).toISOString(); |
| 1988 | const open = this.sql |
| 1989 | .exec<{ n: number }>("SELECT count(*) AS n FROM sandboxes WHERE kind = 'api' AND status = 'running' AND created_at > ? AND json_extract(created_by, '$.id') = ?", since, caller.id) |
| 1990 | .one().n; |
| 1991 | if (open >= MAX_OPEN_SANDBOXES) throw new HttpError(429, `you have ${open} open sandboxes in this repository, the most allowed; close one first`); |
| 1992 | } |
| 1993 | const sha = await this.repo.resolve(ref ?? "HEAD"); |
| 1994 | if (!sha) throw new HttpError(422, "unknown ref"); |
| 1995 | const id = newId(); |
| 1996 | const m = this.repo.meta; |
| 1997 | const token = `dst_${m.repoId}_${id}_${randomToken(24)}`; |
| 1998 | const secret = randomToken(32); |
| 1999 | this.sql.exec( |
| 2000 | "INSERT INTO sandboxes (id, secret, created_by, ref, created_at, status, kind, task_person) VALUES (?, ?, ?, ?, ?, 'running', ?, ?)", |
| 2001 | id, |
| 2002 | await sha256(token), |
| 2003 | JSON.stringify(caller), |
| 2004 | ref ?? m.defaultBranch, |
| 2005 | now(), |
| 2006 | task ? "task" : "api", |
| 2007 | task?.personId ?? null, |
| 2008 | ); |
| 2009 | const stub = this.env.SANDBOX.get(this.env.SANDBOX.idFromName(id)); |
| 2010 | try { |
| 2011 | await stub.boot({ |
| 2012 | DIFFINERY_CLONE_URL: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, |
| 2013 | DIFFINERY_REF: ref ?? m.defaultBranch, |
| 2014 | DIFFINERY_JOB_TOKEN: token, |
| 2015 | DIFFINERY_SANDBOX_SECRET: secret, |
| 2016 | DIFFINERY_API_URL: this.env.PUBLIC_URL, |
| 2017 | GIT_AUTHOR_NAME: caller.login, |
| 2018 | GIT_AUTHOR_EMAIL: `${caller.login}@users.noreply.diffinery.com`, |
| 2019 | GIT_COMMITTER_NAME: caller.login, |
| 2020 | GIT_COMMITTER_EMAIL: `${caller.login}@users.noreply.diffinery.com`, |
| 2021 | }, secret, id); |
| 2022 | } catch (e) { |
| 2023 | console.error("sandbox boot failed", e); |
| 2024 | this.sql.exec("UPDATE sandboxes SET status = 'failed' WHERE id = ?", id); |
| 2025 | await stub.shutdown().catch(() => {}); |
| 2026 | throw new HttpError(503, "the sandbox container did not start; try again"); |
| 2027 | } |
| 2028 | return { id, ref: ref ?? m.defaultBranch }; |
| 2029 | } |
| 2030 | |
| 2031 | getSandbox(id: string, caller: Caller): { id: string; ref: string; status: string; createdAt: string } { |
| 2032 | const s = this.sql.exec<{ id: string; ref: string; status: string; created_at: string; created_by: string; kind: string }>("SELECT * FROM sandboxes WHERE id = ?", id).toArray()[0]; |
| 2033 | // A task's sandbox holds a credential for the person's model access: nobody reaches it through the sandbox API. |
| 2034 | if (!s || s.kind === "task") throw new HttpError(404, "sandbox not found"); |
| 2035 | const owner = JSON.parse(s.created_by) as Caller; |
| 2036 | if (owner.id !== caller.id && !atLeast(caller.role, "admin")) throw new HttpError(403, "not your sandbox"); |
| 2037 | return { id: s.id, ref: s.ref, status: s.status, createdAt: s.created_at }; |
| 2038 | } |
| 2039 | |
| 2040 | async closeSandbox(id: string, caller: Caller): Promise<void> { |
| 2041 | this.getSandbox(id, caller); |
| 2042 | this.sql.exec("UPDATE sandboxes SET status = 'closed' WHERE id = ?", id); |
| 2043 | await this.env.SANDBOX.get(this.env.SANDBOX.idFromName(id)).shutdown(); |
| 2044 | } |
| 2045 | |
| 2046 | /** Close an agent task's sandbox (agentic/), which revokes its git credential. */ |
| 2047 | async closeTaskSandbox(id: string): Promise<void> { |
| 2048 | const s = this.sql.exec<{ kind: string }>("SELECT kind FROM sandboxes WHERE id = ?", id).toArray()[0]; |
| 2049 | if (s?.kind !== "task") return; |
| 2050 | this.sql.exec("UPDATE sandboxes SET status = 'closed' WHERE id = ?", id); |
| 2051 | await this.env.SANDBOX.get(this.env.SANDBOX.idFromName(id)).shutdown(); |
| 2052 | } |
| 2053 | } |